Repository navigation
Expand file tree
/
Copy pathevents_adopt_test.go
More file actions
133 lines (126 loc) · 3.13 KB
/
Copy pathevents_adopt_test.go
File metadata and controls
133 lines (126 loc) · 3.13 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
package uio
import (
"errors"
"io"
"net"
"testing"
"time"
)
func expectPeerClosed(t *testing.T, conn net.Conn) {
t.Helper()
if err := conn.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
t.Fatal(err)
}
var buffer [1]byte
_, err := conn.Read(buffer[:])
if err == nil {
t.Fatal("peer remained open")
}
if timeout, ok := err.(net.Error); ok && timeout.Timeout() {
t.Fatalf("peer was not closed: %v", err)
}
if !errors.Is(err, io.EOF) && !errors.Is(err, net.ErrClosed) {
var operationError *net.OpError
if !errors.As(err, &operationError) {
t.Fatalf("peer read = %v, want closed connection", err)
}
}
}
func TestAdoptTransfersTCPConnection(t *testing.T) {
events := &Events{Pollers: 1}
started := make(chan struct{})
opened := make(chan Conn, 1)
received := make(chan string, 1)
events.OnStart = func(*Events) { close(started) }
events.OnOpen = func(conn Conn) { opened <- conn }
events.OnData = func(conn Conn) error {
payload := make([]byte, conn.InboundBuffered())
_, err := conn.Read(payload)
if err == nil {
received <- string(payload)
}
return err
}
serveDone := make(chan error, 1)
go func() { serveDone <- events.Serve() }()
t.Cleanup(func() {
_ = events.Close(nil)
select {
case <-serveDone:
case <-time.After(time.Second):
t.Error("Events.Serve did not stop")
}
})
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("Events.Serve did not start")
}
client, server := tcpConnectionPair(t)
defer client.Close()
const userdata = "adopted"
adopted, err := events.Adopt(server, userdata)
if err != nil {
t.Fatal(err)
}
select {
case conn := <-opened:
if conn != adopted || conn.Userdata() != userdata {
t.Fatalf("OnOpen connection = %v/%v, want adopted userdata", conn, conn.Userdata())
}
case <-time.After(time.Second):
t.Fatal("Adopt did not call OnOpen")
}
if _, err = client.Write([]byte("payload")); err != nil {
t.Fatal(err)
}
select {
case payload := <-received:
if payload != "payload" {
t.Fatalf("received payload = %q", payload)
}
case <-time.After(time.Second):
t.Fatal("adopted connection did not receive data")
}
}
func TestAdoptClosesConnectionWhenEventsNotServing(t *testing.T) {
client, server := tcpConnectionPair(t)
defer client.Close()
if conn, err := (&Events{}).Adopt(server, nil); conn != nil || !errors.Is(err, net.ErrClosed) {
t.Fatalf("Adopt() = %v, %v; want nil, net.ErrClosed", conn, err)
}
expectPeerClosed(t, client)
}
func tcpConnectionPair(t *testing.T) (net.Conn, net.Conn) {
t.Helper()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer listener.Close()
accepted := make(chan net.Conn, 1)
acceptErr := make(chan error, 1)
go func() {
conn, err := listener.Accept()
if err != nil {
acceptErr <- err
return
}
accepted <- conn
}()
client, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal(err)
}
select {
case server := <-accepted:
return client, server
case err := <-acceptErr:
client.Close()
t.Fatal(err)
case <-time.After(time.Second):
client.Close()
t.Fatal("listener did not accept connection")
}
return nil, nil
}