MiFaZhan
84 lines
2.3 KiB
Go
84 lines
2.3 KiB
Go
package rabbitmq
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"net"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestConfirmedExpiration(t *testing.T) {
|
|
for _, item := range []struct {
|
|
delay time.Duration
|
|
want string
|
|
}{{0, ""}, {-time.Second, ""}, {time.Second, "1000"}, {1500 * time.Millisecond, "1500"}} {
|
|
if got := confirmedExpiration(item.delay); got != item.want {
|
|
t.Fatalf("expiration(%v)=%q, want %q", item.delay, got, item.want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestConfirmedConnectionHandshakeTimeout(t *testing.T) {
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer listener.Close()
|
|
finished := make(chan struct{})
|
|
go func() {
|
|
defer close(finished)
|
|
connection, acceptErr := listener.Accept()
|
|
if acceptErr != nil {
|
|
return
|
|
}
|
|
defer connection.Close()
|
|
_ = connection.SetDeadline(time.Now().Add(3 * time.Second))
|
|
_, _ = io.Copy(io.Discard, connection)
|
|
}()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
|
defer cancel()
|
|
started := time.Now()
|
|
connection, cleanup, err := openConfirmedConnection(ctx, "amqp://guest:guest@"+listener.Addr().String()+"/")
|
|
if cleanup != nil {
|
|
cleanup()
|
|
}
|
|
if connection != nil || !errors.Is(err, context.DeadlineExceeded) {
|
|
t.Fatalf("expected handshake timeout, connection=%v err=%v", connection, err)
|
|
}
|
|
if time.Since(started) > 2*time.Second {
|
|
t.Fatal("handshake exceeded the context deadline")
|
|
}
|
|
select {
|
|
case <-finished:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("timed out socket was not closed")
|
|
}
|
|
}
|
|
|
|
func TestConfirmedConnectionCanceledContext(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
connection, cleanup, err := openConfirmedConnection(ctx, "invalid-dsn")
|
|
if connection != nil || cleanup != nil || !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("expected cancellation before dialing, connection=%v err=%v", connection, err)
|
|
}
|
|
}
|
|
|
|
func TestConsumerCloserUnregister(t *testing.T) {
|
|
closerMu.Lock()
|
|
initial := len(closers)
|
|
closerMu.Unlock()
|
|
for attempt := 0; attempt < 1000; attempt++ {
|
|
unregister := registerCloser(func() error { return nil })
|
|
unregister()
|
|
unregister()
|
|
}
|
|
closerMu.Lock()
|
|
defer closerMu.Unlock()
|
|
if len(closers) != initial {
|
|
t.Fatalf("stale consumer closers retained: initial=%d actual=%d", initial, len(closers))
|
|
}
|
|
}
|