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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
package amqptest
import (
"errors"
"fmt"
"testing"
"time"
"github.com/NeowayLabs/wabbit"
"github.com/NeowayLabs/wabbit/amqptest/server"
"github.com/pborman/uuid"
)
var rabbitmqPort = "35672"
// WaitOK blocks until rabbitmq can accept connections on
// <ctn ip address>:5672
func waitRabbitOK(amqpuri string) error {
var err error
var counter = 0
dial:
if counter > 120 {
panic("Impossible to connect to rabbitmq")
}
_, err = Dial(amqpuri)
if err != nil {
time.Sleep(500 * time.Millisecond)
counter++
goto dial
}
return nil
}
// TestDial test a simple connection to rabbitmq.
// If the rabbitmq disconnects will not be tested here!
func TestDial(t *testing.T) {
amqpuri := "amqp://guest:guest@localhost:35672/%2f"
// Should fail
conn, err := Dial(amqpuri)
if err == nil {
t.Error("No backend started... Should fail")
return
}
server := server.NewServer(amqpuri)
server.Start()
err = waitRabbitOK(amqpuri)
if err != nil {
t.Error(err)
return
}
conn, err = Dial(amqpuri)
if err != nil || conn == nil {
t.Error(err)
return
}
server.Stop()
}
func TestAutoRedial(t *testing.T) {
var err error
amqpuri := "amqp://guest:guest@localhost:35672/%2f"
server := server.NewServer(amqpuri)
server.Start()
defer server.Stop()
err = waitRabbitOK(amqpuri)
if err != nil {
t.Error(err)
return
}
conn, err := Dial(amqpuri)
if err != nil {
t.Error(err)
return
}
defer conn.Close()
redialErrors := make(chan wabbit.Error)
done := make(chan bool)
conn.AutoRedial(redialErrors, done)
// required goroutine to consume connection error messages
go func() {
for {
// discards the connection errors
<-redialErrors
}
}()
server.Stop()
// concurrently starts the rabbitmq after 1 second
go func() {
time.Sleep(10 * time.Second)
server.Start()
}()
select {
case <-time.After(20 * time.Second):
err = errors.New("Timeout exceeded. AMQP reconnect failed")
case <-done:
err = nil
}
if err != nil {
t.Errorf("Client doesn't reconnect in 3 seconds: %s", err.Error())
return
}
}
func TestConnMock(t *testing.T) {
var conn wabbit.Conn
// rabbitmq.Conn satisfies wabbit.Conn interface
conn = &Conn{}
if conn == nil {
t.Error("Maybe wabbit.Conn interface does not mock amqp.Conn correctly")
}
}
func TestConnCloseBeforeServerStop(t *testing.T) {
tests := []struct {
name string
sleep time.Duration
}{
{
name: "10 ms sleep before server.Stop()",
sleep: 10 * time.Millisecond,
},
{
name: "100 ms sleep before server.Stop()",
sleep: 100 * time.Millisecond,
},
{
name: "1000 ms sleep before server.Stop()",
sleep: time.Second,
},
}
for _, tc := range tests {
tc := tc // capture range variable.
t.Run(tc.name, func(t *testing.T) {
// Start a server.
uri := fmt.Sprintf("amqp://guest:guest@localhost:35672/%s",
uuid.New())
server := server.NewServer(uri)
server.Start()
defer server.Stop()
// Dial to the server.
conn, err := Dial(uri)
if err != nil {
t.Fatalf("Dial(): %v", err)
}
conn.Close()
// Sleep before stopping the server.
time.Sleep(tc.sleep)
})
}
}
|
