// Copyright (c) 2019 Andy Pan

// Copyright (c) 2017 Joshua J Baker

//

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

package gnet

import (

"bufio"

"bytes"

"context"

"encoding/binary"

"errors"

"fmt"

"io"

"math/rand"

"net"

"runtime"

"sync"

"sync/atomic"

"testing"

"time"

"github.com/stretchr/testify/assert"

"github.com/stretchr/testify/require"

"go.uber.org/zap"

gerr "github.com/panjf2000/gnet/v2/pkg/errors"

"github.com/panjf2000/gnet/v2/pkg/logging"

bbPool "github.com/panjf2000/gnet/v2/pkg/pool/bytebuffer"

goPool "github.com/panjf2000/gnet/v2/pkg/pool/goroutine"

)

var streamLen = 1024 * 1024

func TestServe(t *testing.T) {

// start an engine

// connect 10 clients

// each client will pipe random data for 1-3 seconds.

// the writes to the engine will be random sizes. 0KB - 1MB.

// the engine will echo back the data.

// waits for graceful connection closing.

t.Run("poll", func(t *testing.T) {

t.Run("tcp", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", false, false, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", false, false, true, false, false, 10, LeastConnections)

})

})

t.Run("tcp-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", false, false, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", false, false, true, true, false, 10, LeastConnections)

})

})

t.Run("tcp-async-writev", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", false, false, false, true, true, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", false, false, true, true, true, 10, LeastConnections)

})

})

t.Run("udp", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "udp", ":9991", false, false, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "udp", ":9992", false, false, true, false, false, 10, LeastConnections)

})

})

t.Run("udp-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "udp", ":9991", false, false, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "udp", ":9992", false, false, true, true, false, 10, LeastConnections)

})

})

t.Run("unix", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", false, false, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", false, false, true, false, false, 10, SourceAddrHash)

})

})

t.Run("unix-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", false, false, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", false, false, true, true, false, 10, SourceAddrHash)

})

})

t.Run("unix-async-writev", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", false, false, false, true, true, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", false, false, true, true, true, 10, SourceAddrHash)

})

})

})

t.Run("poll-reuseport", func(t *testing.T) {

t.Run("tcp", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", true, false, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", true, false, true, false, false, 10, LeastConnections)

})

})

t.Run("tcp-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", true, false, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", true, false, true, true, false, 10, LeastConnections)

})

})

t.Run("tcp-async-writev", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", true, false, false, true, true, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", true, false, true, true, true, 10, LeastConnections)

})

})

t.Run("udp", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "udp", ":9991", true, false, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "udp", ":9992", true, false, true, false, false, 10, LeastConnections)

})

})

t.Run("udp-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "udp", ":9991", true, false, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "udp", ":9992", true, false, true, true, false, 10, LeastConnections)

})

})

t.Run("unix", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", true, false, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", true, false, true, false, false, 10, LeastConnections)

})

})

t.Run("unix-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", true, false, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", true, false, true, true, false, 10, LeastConnections)

})

})

t.Run("unix-async-writev", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", true, false, false, true, true, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", true, false, true, true, true, 10, LeastConnections)

})

})

})

t.Run("poll-reuseaddr", func(t *testing.T) {

t.Run("tcp", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", false, true, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", false, true, true, false, false, 10, LeastConnections)

})

})

t.Run("tcp-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "tcp", ":9991", false, true, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "tcp", ":9992", false, true, true, false, false, 10, LeastConnections)

})

})

t.Run("udp", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "udp", ":9991", false, true, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "udp", ":9992", false, true, true, false, false, 10, LeastConnections)

})

})

t.Run("udp-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "udp", ":9991", false, true, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "udp", ":9992", false, true, true, true, false, 10, LeastConnections)

})

})

t.Run("unix", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", false, true, false, false, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", false, true, true, false, false, 10, LeastConnections)

})

})

t.Run("unix-async", func(t *testing.T) {

t.Run("1-loop", func(t *testing.T) {

testServe(t, "unix", "gnet1.sock", false, true, false, true, false, 10, RoundRobin)

})

t.Run("N-loop", func(t *testing.T) {

testServe(t, "unix", "gnet2.sock", false, true, true, true, false, 10, LeastConnections)

})

})

})

}

type testServer struct {

*BuiltinEventEngine

tester *testing.T

eng Engine

network string

addr string

multicore bool

async bool

writev bool

nclients int

started int32

connected int32

clientActive int32

disconnected int32

workerPool *goPool.Pool

}

func (s *testServer) OnBoot(eng Engine) (action Action) {

s.eng = eng

return

}

func (s *testServer) OnOpen(c Conn) (out []byte, action Action) {

c.SetContext(c)

atomic.AddInt32(&s.connected, 1)

out = []byte("sweetness\r\n")

require.NotNil(s.tester, c.LocalAddr(), "nil local addr")

require.NotNil(s.tester, c.RemoteAddr(), "nil remote addr")

return

}

func (s *testServer) OnClose(c Conn, err error) (action Action) {

if err != nil {

logging.Debugf("error occurred on closed, %v\n", err)

}

if s.network != "udp" {

require.Equal(s.tester, c.Context(), c, "invalid context")

}

atomic.AddInt32(&s.disconnected, 1)

if atomic.LoadInt32(&s.connected) == atomic.LoadInt32(&s.disconnected) &&

atomic.LoadInt32(&s.disconnected) == int32(s.nclients) {

action = Shutdown

s.workerPool.Release()

}

return

}

func (s *testServer) OnTraffic(c Conn) (action Action) {

if s.async {

buf := bbPool.Get()

_, _ = c.WriteTo(buf)

if s.network == "tcp" || s.network == "unix" {

// just for test

_ = c.InboundBuffered()

_ = c.OutboundBuffered()

_, _ = c.Discard(1)

_ = s.workerPool.Submit(

func() {

if s.writev {

mid := buf.Len() / 2

bs := make([][]byte, 2)

bs[0] = buf.B[:mid]

bs[1] = buf.B[mid:]

_ = c.AsyncWritev(bs, func(c Conn, err error) error {

logging.Debugf("conn=%s done writev: %v", c.RemoteAddr().String(), err)

return nil

})

} else {

_ = c.AsyncWrite(buf.Bytes(), func(c Conn, err error) error {

logging.Debugf("conn=%s done write: %v", c.RemoteAddr().String(), err)

return nil

})

}

})

return

} else if s.network == "udp" {

_ = s.workerPool.Submit(

func() {

_ = c.AsyncWrite(buf.Bytes(), nil)

})

return

}

return

}

buf, _ := c.Next(-1)

_, _ = c.Write(buf)

return

}

func (s *testServer) OnTick() (delay time.Duration, action Action) {

if atomic.CompareAndSwapInt32(&s.started, 0, 1) {

for i := 0; i < s.nclients; i++ {

atomic.AddInt32(&s.clientActive, 1)

go func() {

startClient(s.tester, s.network, s.addr, s.multicore, s.async)

atomic.AddInt32(&s.clientActive, -1)

}()

}

}

if s.network == "udp" && atomic.LoadInt32(&s.clientActive) == 0 {

action = Shutdown

return

}

delay = time.Second / 5

return

}

func testServe(t *testing.T, network, addr string, reuseport, reuseaddr, multicore, async, writev bool, nclients int, lb LoadBalancing) {

ts := &testServer{

tester: t,

network: network,

addr: addr,

multicore: multicore,

async: async,

writev: writev,

nclients: nclients,

workerPool: goPool.Default(),

}

err := Run(ts,

network+"://"+addr,

WithLockOSThread(async),

WithMulticore(multicore),

WithReusePort(reuseport),

WithReuseAddr(reuseaddr),

WithTicker(true),

WithTCPKeepAlive(time.Minute*1),

WithTCPNoDelay(TCPDelay),

WithLoadBalancing(lb))

assert.NoError(t, err)

}

func startClient(t *testing.T, network, addr string, multicore, async bool) {

rand.Seed(time.Now().UnixNano())

c, err := net.Dial(network, addr)

require.NoError(t, err)

defer c.Close()

rd := bufio.NewReader(c)

if network != "udp" {

msg, err := rd.ReadBytes('\n')

require.NoError(t, err)

require.Equal(t, string(msg), "sweetness\r\n", "bad header")

}

duration := time.Duration((rand.Float64()*2+1)*float64(time.Second)) / 2

t.Logf("test duration: %dms", duration/time.Millisecond)

start := time.Now()

for time.Since(start) < duration {

reqData := make([]byte, streamLen)

if network == "udp" {

reqData = reqData[:1024]

}

_, err = rand.Read(reqData)

require.NoError(t, err)

_, err = c.Write(reqData)

require.NoError(t, err)

respData := make([]byte, len(reqData))

_, err = io.ReadFull(rd, respData)

require.NoError(t, err)

if !async {

// require.Equalf(t, reqData, respData, "response mismatch with protocol:%s, multi-core:%t, content of bytes: %d vs %d", network, multicore, string(reqData), string(respData))

require.Equalf(

t,

reqData,

respData,

"response mismatch with protocol:%s, multi-core:%t, length of bytes: %d vs %d",

network,

multicore,

len(reqData),

len(respData),

)

}

}

}

// NOTE: TestServeMulticast can fail with "write: no buffer space available" on wifi interface.

func TestServeMulticast(t *testing.T) {

t.Run("IPv4", func(t *testing.T) {

// 224.0.0.169 is an unassigned address from the Local Network Control Block

// https://www.iana.org/assignments/multicast-addresses/multicast-addresses.xhtml#multicast-addresses-1

t.Run("udp-multicast", func(t *testing.T) {

testMulticast(t, "224.0.0.169:9991", false, false, -1, 10)

})

t.Run("udp-multicast-reuseport", func(t *testing.T) {

testMulticast(t, "224.0.0.169:9991", true, false, -1, 10)

})

t.Run("udp-multicast-reuseaddr", func(t *testing.T) {

testMulticast(t, "224.0.0.169:9991", false, true, -1, 10)

})

})

t.Run("IPv6", func(t *testing.T) {

iface, err := findLoopbackInterface()

require.NoError(t, err)

if iface.Flags&net.FlagMulticast != net.FlagMulticast {

t.Skip("multicast is not supported on loopback interface")

}

// ff02::3 is an unassigned address from Link-Local Scope Multicast Addresses

// https://www.iana.org/assignments/ipv6-multicast-addresses/ipv6-multicast-addresses.xhtml#link-local

t.Run("udp-multicast", func(t *testing.T) {

testMulticast(t, fmt.Sprintf("[ff02::3%%%s]:9991", iface.Name), false, false, iface.Index, 10)

})

t.Run("udp-multicast-reuseport", func(t *testing.T) {

testMulticast(t, fmt.Sprintf("[ff02::3%%%s]:9991", iface.Name), true, false, iface.Index, 10)

})

t.Run("udp-multicast-reuseaddr", func(t *testing.T) {

testMulticast(t, fmt.Sprintf("[ff02::3%%%s]:9991", iface.Name), false, true, iface.Index, 10)

})

})

}

func findLoopbackInterface() (*net.Interface, error) {

ifaces, err := net.Interfaces()

if err != nil {

return nil, err

}

for _, iface := range ifaces {

if iface.Flags&net.FlagLoopback == net.FlagLoopback {

return &iface, nil

}

}

return nil, errors.New("no loopback interface")

}

func testMulticast(t *testing.T, addr string, reuseport, reuseaddr bool, index, nclients int) {

ts := &testMcastServer{

t: t,

addr: addr,

nclients: nclients,

}

options := []Option{

WithReuseAddr(reuseaddr),

WithReusePort(reuseport),

WithSocketRecvBuffer(2 * nclients * 1024), // enough space to receive messages from nclients to eliminate dropped packets

WithTicker(true),

}

if index != -1 {

options = append(options, WithMulticastInterfaceIndex(index))

}

err := Run(ts, "udp://"+addr, options...)

assert.NoError(t, err)

}

type testMcastServer struct {

*BuiltinEventEngine

t *testing.T

mcast sync.Map

addr string

nclients int

started int32

active int32

}

func (s *testMcastServer) startMcastClient() {

rand.Seed(time.Now().UnixNano())

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

defer cancel()

c, err := net.Dial("udp", s.addr)

require.NoError(s.t, err)

defer c.Close()

ch := make(chan []byte, 10000)

s.mcast.Store(c.LocalAddr().String(), ch)

duration := time.Duration((rand.Float64()*2+1)*float64(time.Second)) / 2

s.t.Logf("test duration: %dms", duration/time.Millisecond)

start := time.Now()

for time.Since(start) < duration {

reqData := make([]byte, 1024)

_, err = rand.Read(reqData)

require.NoError(s.t, err)

_, err = c.Write(reqData)

require.NoError(s.t, err)

// Workaround for MacOS "write: no buffer space available" error messages

// https://developer.apple.com/forums/thread/42334

time.Sleep(time.Millisecond)

select {

case respData := <-ch:

require.Equalf(s.t, reqData, respData, "response mismatch, length of bytes: %d vs %d", len(reqData), len(respData))

case <-ctx.Done():

require.Fail(s.t, "timeout receiving message")

return

}

}

}

func (s *testMcastServer) OnTraffic(c Conn) (action Action) {

buf, _ := c.Next(-1)

b := make([]byte, len(buf))

copy(b, buf)

ch, ok := s.mcast.Load(c.RemoteAddr().String())

require.True(s.t, ok)

ch.(chan []byte) <- b

return

}

func (s *testMcastServer) OnTick() (delay time.Duration, action Action) {

if atomic.CompareAndSwapInt32(&s.started, 0, 1) {

for i := 0; i < s.nclients; i++ {

atomic.AddInt32(&s.active, 1)

go func() {

s.startMcastClient()

atomic.AddInt32(&s.active, -1)

}()

}

}

if atomic.LoadInt32(&s.active) == 0 {

action = Shutdown

return

}

delay = time.Second / 5

return

}

type testMulticastBindServer struct {

*BuiltinEventEngine

}

func (t *testMulticastBindServer) OnTick() (delay time.Duration, action Action) {

action = Shutdown

return

}

func TestMulticastBindIPv4(t *testing.T) {

ts := &testMulticastBindServer{}

iface, err := findLoopbackInterface()

require.NoError(t, err)

err = Run(ts, "udp://224.0.0.169:9991",

WithMulticastInterfaceIndex(iface.Index),

WithTicker(true))

assert.NoError(t, err)

}

func TestMulticastBindIPv6(t *testing.T) {

ts := &testMulticastBindServer{}

iface, err := findLoopbackInterface()

require.NoError(t, err)

err = Run(ts, fmt.Sprintf("udp://[ff02::3%%%s]:9991", iface.Name),

WithMulticastInterfaceIndex(iface.Index),

WithTicker(true))

assert.NoError(t, err)

}

func TestDefaultGnetServer(t *testing.T) {

svr := BuiltinEventEngine{}

svr.OnBoot(Engine{})

svr.OnOpen(nil)

svr.OnClose(nil, nil)

svr.OnTraffic(nil)

svr.OnTick()

}

type testBadAddrServer struct {

*BuiltinEventEngine

}

func (t *testBadAddrServer) OnBoot(_ Engine) (action Action) {

return Shutdown

}

func TestBadAddresses(t *testing.T) {

events := new(testBadAddrServer)

err := Run(events, "tulip://howdy")

assert.Error(t, err)

err = Run(events, "howdy")

assert.Error(t, err)

err = Run(events, "tcp://")

assert.NoError(t, err)

}

func TestTick(t *testing.T) {

testTick("tcp", ":9989", t)

}

type testTickServer struct {

*BuiltinEventEngine

count int

}

func (t *testTickServer) OnTick() (delay time.Duration, action Action) {

if t.count == 25 {

action = Shutdown

return

}

t.count++

delay = time.Millisecond * 10

return

}

func testTick(network, addr string, t *testing.T) {

events := &testTickServer{}

start := time.Now()

opts := Options{Ticker: true}

err := Run(events, network+"://"+addr, WithOptions(opts))

assert.NoError(t, err)

dur := time.Since(start)

if dur < 250&time.Millisecond || dur > time.Second {

t.Logf("bad ticker timing: %d", dur)

}

}

func TestWakeConn(t *testing.T) {

testWakeConn(t, "tcp", ":9990")

}

type testWakeConnServer struct {

*BuiltinEventEngine

tester *testing.T

network string

addr string

conn chan Conn

c Conn

wake bool

}

func (t *testWakeConnServer) OnOpen(c Conn) (out []byte, action Action) {

t.conn <- c

return

}

func (t *testWakeConnServer) OnClose(c Conn, err error) (action Action) {

action = Shutdown

return

}

func (t *testWakeConnServer) OnTraffic(c Conn) (action Action) {

_, _ = c.Write([]byte("Waking up."))

action = -1

return

}

func (t *testWakeConnServer) OnTick() (delay time.Duration, action Action) {

if !t.wake {

t.wake = true

delay = time.Millisecond * 100

go func() {

conn, err := net.Dial(t.network, t.addr)

require.NoError(t.tester, err)

defer conn.Close()

r := make([]byte, 10)

_, err = conn.Read(r)

require.NoError(t.tester, err)

}()

return

}

t.c = <-t.conn

_ = t.c.Wake(func(c Conn, err error) error {

logging.Debugf("conn=%s done wake: %v", c.RemoteAddr().String(), err)

return nil

})

delay = time.Millisecond * 100

return

}

func testWakeConn(t *testing.T, network, addr string) {

svr := &testWakeConnServer{tester: t, network: network, addr: addr, conn: make(chan Conn, 1)}

logger := zap.NewExample()

err := Run(svr, network+"://"+addr,

WithTicker(true),

WithNumEventLoop(2*runtime.NumCPU()),

WithLogger(logger.Sugar()),

WithSocketRecvBuffer(4*1024),

WithSocketSendBuffer(4*1024),

WithReadBufferCap(2000),

WithWriteBufferCap(2000))

assert.NoError(t, err)

_ = logger.Sync()

}

func TestShutdown(t *testing.T) {

testShutdown(t, "tcp", ":9991")

}

type testShutdownServer struct {

*BuiltinEventEngine

tester *testing.T

network string

addr string

count int

clients int64

N int

}

func (t *testShutdownServer) OnOpen(c Conn) (out []byte, action Action) {

atomic.AddInt64(&t.clients, 1)

return

}

func (t *testShutdownServer) OnClose(c Conn, err error) (action Action) {

atomic.AddInt64(&t.clients, -1)

return

}

func (t *testShutdownServer) OnTick() (delay time.Duration, action Action) {

if t.count == 0 {

// start clients

for i := 0; i < t.N; i++ {

go func() {

conn, err := net.Dial(t.network, t.addr)

require.NoError(t.tester, err)

defer conn.Close()

_, err = conn.Read([]byte{0})

require.Error(t.tester, err)

}()

}

} else if int(atomic.LoadInt64(&t.clients)) == t.N {

action = Shutdown

}

t.count++

delay = time.Second / 20

return

}

func testShutdown(t *testing.T, network, addr string) {

events := &testShutdownServer{tester: t, network: network, addr: addr, N: 10}

err := Run(events, network+"://"+addr, WithTicker(true), WithReadBufferCap(512), WithWriteBufferCap(512))

assert.NoError(t, err)

require.Equal(t, int(events.clients), 0, "did not call close on all clients")

}

func TestCloseActionError(t *testing.T) {

testCloseActionError(t, "tcp", ":9992")

}

type testCloseActionErrorServer struct {

*BuiltinEventEngine

tester *testing.T

network, addr string

action bool

}

func (t *testCloseActionErrorServer) OnClose(c Conn, err error) (action Action) {

action = Shutdown

return

}

func (t *testCloseActionErrorServer) OnTraffic(c Conn) (action Action) {

n := c.InboundBuffered()

buf, _ := c.Peek(n)

_, _ = c.Write(buf)

_, _ = c.Discard(n)

action = Close

return

}

func (t *testCloseActionErrorServer) OnTick() (delay time.Duration, action Action) {

if !t.action {

t.action = true

delay = time.Millisecond * 100

go func() {

conn, err := net.Dial(t.network, t.addr)

require.NoError(t.tester, err)

defer conn.Close()

data := []byte("Hello World!")

_, _ = conn.Write(data)

_, err = conn.Read(data)

require.NoError(t.tester, err)

}()

return

}

delay = time.Millisecond * 100

return

}

func testCloseActionError(t *testing.T, network, addr string) {

events := &testCloseActionErrorServer{tester: t, network: network, addr: addr}

err := Run(events, network+"://"+addr, WithTicker(true))

assert.NoError(t, err)

}

func TestShutdownActionError(t *testing.T) {

testShutdownActionError(t, "tcp", ":9993")

}

type testShutdownActionErrorServer struct {

*BuiltinEventEngine

tester *testing.T

network, addr string

action bool

}

func (t *testShutdownActionErrorServer) OnTraffic(c Conn) (action Action) {

buf, _ := c.Peek(-1)

_, _ = c.Write(buf)

_, _ = c.Discard(-1)

action = Shutdown

return

}

func (t *testShutdownActionErrorServer) OnTick() (delay time.Duration, action Action) {

if !t.action {

t.action = true

delay = time.Millisecond * 100

go func() {

conn, err := net.Dial(t.network, t.addr)

require.NoError(t.tester, err)

defer conn.Close()

data := []byte("Hello World!")

_, _ = conn.Write(data)

_, err = conn.Read(data)

require.NoError(t.tester, err)

}()

return

}

delay = time.Millisecond * 100

return

}

func testShutdownActionError(t *testing.T, network, addr string) {

events := &testShutdownActionErrorServer{tester: t, network: network, addr: addr}

err := Run(events, network+"://"+addr, WithTicker(true))

assert.NoError(t, err)

}

func TestCloseActionOnOpen(t *testing.T) {

testCloseActionOnOpen(t, "tcp", ":9994")

}

type testCloseActionOnOpenServer struct {

*BuiltinEventEngine

tester *testing.T

network, addr string

action bool

}

func (t *testCloseActionOnOpenServer) OnOpen(c Conn) (out []byte, action Action) {

action = Close

return

}

func (t *testCloseActionOnOpenServer) OnClose(c Conn, err error) (action Action) {

action = Shutdown

return

}

func (t *testCloseActionOnOpenServer) OnTick() (delay time.Duration, action Action) {

if !t.action {

t.action = true

delay = time.Millisecond * 100

go func() {

conn, err := net.Dial(t.network, t.addr)

require.NoError(t.tester, err)

defer conn.Close()

}()

return

}

delay = time.Millisecond * 100

return

}

func testCloseActionOnOpen(t *testing.T, network, addr string) {

events := &testCloseActionOnOpenServer{tester: t, network: network, addr: addr}

err := Run(events, network+"://"+addr, WithTicker(true))

assert.NoError(t, err)

}

func TestShutdownActionOnOpen(t *testing.T) {

testShutdownActionOnOpen(t, "tcp", ":9995")

}

type testShutdownActionOnOpenServer struct {

*BuiltinEventEngine

tester *testing.T

network, addr string

action bool

}

func (t *testShutdownActionOnOpenServer) OnOpen(c Conn) (out []byte, action Action) {

action = Shutdown

return

}

func (t *testShutdownActionOnOpenServer) OnShutdown(s Engine) {

dupFD, err := s.Dup()

logging.Debugf("dup fd: %d with error: %v\n", dupFD, err)

}

func (t *testShutdownActionOnOpenServer) OnTick() (delay time.Duration, action Action) {

if !t.action {

t.action = true

delay = time.Millisecond * 100

go func() {

conn, err := net.Dial(t.network, t.addr)

require.NoError(t.tester, err)

defer conn.Close()

}()

return

}

delay = time.Millisecond * 100

return

}

func testShutdownActionOnOpen(t *testing.T, network, addr string) {

events := &testShutdownActionOnOpenServer{tester: t, network: network, addr: addr}

err := Run(events, network+"://"+addr, WithTicker(true))

assert.NoError(t, err)

}

func TestUDPShutdown(t *testing.T) {

testUDPShutdown(t, "udp4", ":9000")

}

type testUDPShutdownServer struct {

*BuiltinEventEngine

tester *testing.T

network string

addr string

tick bool

}

func (t *testUDPShutdownServer) OnTraffic(c Conn) (action Action) {

buf, _ := c.Peek(-1)

_, _ = c.Write(buf)

_, _ = c.Discard(-1)

action = Shutdown

return

}

func (t *testUDPShutdownServer) OnTick() (delay time.Duration, action Action) {

if !t.tick {

t.tick = true

delay = time.Millisecond * 100

go func() {

conn, err := net.Dial(t.network, t.addr)

require.NoError(t.tester, err)

defer conn.Close()

data := []byte("Hello World!")

_, err = conn.Write(data)

require.NoError(t.tester, err)

_, err = conn.Read(data)

require.NoError(t.tester, err)

}()

return

}

delay = time.Millisecond * 100

return

}

func testUDPShutdown(t *testing.T, network, addr string) {

svr := &testUDPShutdownServer{tester: t, network: network, addr: addr}

err := Run(svr, network+"://"+addr, WithTicker(true))

assert.NoError(t, err)

}

func TestCloseConnection(t *testing.T) {

testCloseConnection(t, "tcp", ":9996")

}

type testCloseConnectionServer struct {

*BuiltinEventEngine

tester *testing.T

network, addr string

action bool

}

func (t *testCloseConnectionServer) OnClose(c Conn, err error) (action Action) {

action = Shutdown

return

}

func (t *testCloseConnectionServer) OnTraffic(c Conn) (action Action) {

buf, _ := c.Peek(-1)

_, _ = c.Write(buf)