Skip to content

test(amqp091): join connectionWatcher goroutines in test teardown #109

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bithckr merged 2 commits into from
Jun 10, 2026
Merged

test(amqp091): join connectionWatcher goroutines in test teardown #109

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c5d360d
test(amqp091): join connectionWatcher goroutines in teardown
miotte May 30, 2026
9e4bbe9
test(amqp091): pass context as first parameter to stopWatcher
miotte Jun 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion internal/provider/connectors/amqp091/amqp091.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,10 @@ type BrokerDetails struct {
tlsConfig *tls.Config
tlsEnabled bool
shutdownChan chan bool
// watcherWG tracks the connectionWatcher goroutine so callers (notably
// tests) can wait for it to exit after a Disconnect, rather than letting
// it outlive the connection.
watcherWG sync.WaitGroup
}

func init() {
Expand Down Expand Up @@ -507,7 +511,11 @@ func (prov *amqp091provider) Connect(ctx context.Context, cf *pb.ConnectionConfi
return &pb.Error{Message: bdErr.Error()}
}
prov.connections.Add(bd.ClientIdentifier, &bd)
go bd.connectionWatcher()
bd.watcherWG.Add(1)
go func() {
defer bd.watcherWG.Done()
bd.connectionWatcher()
}()

return nil
}
Expand Down
64 changes: 64 additions & 0 deletions internal/provider/connectors/amqp091/amqp091_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,35 @@ const testContentTypeJSON = "application/json"
const testContentEncodingText = "text"
const testXMatchAny = "any"

// stopWatcher disconnects the client identified by ctx and blocks until its
// connectionWatcher goroutine has fully exited. Tests that call prov.Connect
// should defer this so the watcher cannot outlive the test and either race the
// package-level NewAmqpConn091 swap (read in connect()) or panic calling
// IsClosed() on a stale mock from its 30s fallback branch. It is a no-op if the
// connection was already disconnected.
func stopWatcher(ctx context.Context, p provider.Provider) {
prov, ok := p.(*amqp091provider)
if !ok {
return
}
id, err := GetClientIdentifier(ctx)
if err != nil {
return
}
bdu, ok := prov.connections.Get(id)
if !ok {
return
}
bd := bdu.(*BrokerDetails)
// Skip the disconnect if the test already triggered one: a second
// disconnectClientByIdentifier would block on the buffered shutdownChan.
// The watcher still stops on the existing clientDisconnect, so just wait.
if !bd.clientDisconnect.Load() {
prov.disconnectClientByIdentifier(id)
}
bd.watcherWG.Wait()
}

func init() {
// Register the MockProvider with the Provider factory.

Expand Down Expand Up @@ -173,6 +202,7 @@ func TestConnect(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)

assert.Nil(t, err)

Expand Down Expand Up @@ -203,6 +233,7 @@ func TestConnect_Error(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)

assert.NotNil(t, err)

Expand All @@ -225,6 +256,7 @@ func Test_Connect_NoClient(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)

assert.NotNil(t, err)
assert.Contains(t, err.GetMessage(), "noclient")
Expand Down Expand Up @@ -257,6 +289,7 @@ func TestConnect_TLS_SkipVerify(t *testing.T) {
cc := &pb.ConnectionConfiguration{}
cc.Tls = true
err := prov.Connect(ctx, cc, true)
defer stopWatcher(ctx, prov)

assert.Nil(t, err)

Expand Down Expand Up @@ -291,6 +324,7 @@ func TestConnect_TLS_WithCert(t *testing.T) {
cc.Tls = true
cc.CaCertificate = []byte("asdf")
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)

assert.Nil(t, err)

Expand Down Expand Up @@ -323,6 +357,7 @@ func TestConnect_Stats(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

stats := prov.Stats()
Expand Down Expand Up @@ -358,6 +393,7 @@ func setupProviderWithSimpleConn(t *testing.T) (provider.Provider, context.Conte
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

cleanup := func() {
Expand Down Expand Up @@ -440,6 +476,7 @@ func Test_Ack_AckErr(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

subjects := make([]string, 0)
Expand Down Expand Up @@ -504,6 +541,7 @@ func Test_Retry_NoMsg(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)
msg := pb.Message{}
err = prov.Retry(ctx, &pb.Source{}, msg.GetUuid(), 10)
Expand Down Expand Up @@ -537,6 +575,7 @@ func Test_DeadLetter_NoMsg(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)
msg := pb.Message{}
err = prov.DeadLetter(ctx, &pb.Source{}, msg.GetUuid())
Expand Down Expand Up @@ -607,6 +646,7 @@ func Test_Ack(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

subjects := make([]string, 0)
Expand Down Expand Up @@ -698,6 +738,7 @@ func Test_Nack(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

subjects := make([]string, 0)
Expand Down Expand Up @@ -787,6 +828,7 @@ func Test_Nack_NackErr(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

subjects := make([]string, 0)
Expand Down Expand Up @@ -882,6 +924,7 @@ func Test_Retry(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address"}}
Expand Down Expand Up @@ -969,6 +1012,7 @@ func Test_RetryFailure(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address"}}
Expand Down Expand Up @@ -1072,6 +1116,7 @@ func Test_RetryFailure_DeclareErrorsStillSuccess(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

opts := make(map[string]string)
Expand Down Expand Up @@ -1251,6 +1296,7 @@ func Test_DLQ(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

subjects := make([]string, 0)
Expand Down Expand Up @@ -1490,6 +1536,7 @@ func Test_Subscribe_Options(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)
var msg *pb.Message

Expand Down Expand Up @@ -1600,6 +1647,7 @@ func Test_Subscribe_NoSubjectsNoFilters(t *testing.T) {
cc.AdminPort = int32(i) //nolint:gosec

err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)
var msg *pb.Message

Expand Down Expand Up @@ -1676,6 +1724,7 @@ func Test_Subscribe_UnsupportedOptions(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

src := &pb.Source{Address: &pb.Address{Name: "addressname"}, Options: options}
Expand Down Expand Up @@ -1717,6 +1766,7 @@ func Test_Disconnect(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)
prov.Disconnect(ctx)

Expand Down Expand Up @@ -1764,6 +1814,7 @@ func Test_WaitForConnect(t *testing.T) {

cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)
errs <- newAmqp091Error("chanerr", 1) // simulate an error
time.Sleep(500 * time.Millisecond)
Expand Down Expand Up @@ -1822,6 +1873,7 @@ func Test_Publish(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

go func() {
Expand Down Expand Up @@ -1883,6 +1935,7 @@ func Test_Publish_Error(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

go func() {
Expand Down Expand Up @@ -1950,6 +2003,7 @@ func Test_PublishOne(t *testing.T) {

suberr := prov.PublishOne(ctx, msg)

stopWatcher(ctx, prov)
NewAmqpConn091 = oldNewAmqpConn091

assert.Nil(t, suberr)
Expand Down Expand Up @@ -2000,6 +2054,7 @@ func Test_PublishOneFailed(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

suberr := prov.PublishOne(ctx, msg)
Expand Down Expand Up @@ -2045,6 +2100,7 @@ func Test_PublishOneFailedNewChannel(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

suberr := prov.PublishOne(ctx, msg)
Expand Down Expand Up @@ -2086,6 +2142,7 @@ func Test_PublishOneFailedConnClosed(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

suberr := prov.PublishOne(ctx, msg)
Expand Down Expand Up @@ -2165,6 +2222,7 @@ func Test_Publish_ErrorDeclareExchange(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

go func() {
Expand Down Expand Up @@ -2270,6 +2328,7 @@ func Test_ClientExists(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

exists := prov.ClientExists("1234")
Expand Down Expand Up @@ -2302,6 +2361,7 @@ func Test_ClientExists_false(t *testing.T) {
ctx := context.Background()
cc := &pb.ConnectionConfiguration{}
err := prov.Connect(ctx, cc, false)
defer stopWatcher(ctx, prov)
assert.Nil(t, err)

exists := prov.ClientExists("4321")
Expand Down Expand Up @@ -2656,6 +2716,7 @@ func Test_Subscribe_Queue_DeclareOnly(t *testing.T) {

err := prov.Connect(ctx, cc, false)
assert.Nil(t, err)
defer stopWatcher(ctx, prov)

mc := make(chan *pb.Message)
defer close(mc)
Expand Down Expand Up @@ -2968,6 +3029,7 @@ func Test_Publish_ContextCancellation_ExitsPromptly(t *testing.T) {
defer cancel()
cc := &pb.ConnectionConfiguration{}
assert.Nil(t, prov.Connect(ctx, cc, false))
defer stopWatcher(ctx, prov)

// messageChannel is never closed and never has messages. The only exit
// from the Publish select loop is ctx.Done() (post-fix) or an AMQP error
Expand Down Expand Up @@ -3033,6 +3095,7 @@ func Test_Publish_NotifyCloseChannelsAreBuffered(t *testing.T) {
defer cancel()
cc := &pb.ConnectionConfiguration{}
assert.Nil(t, prov.Connect(ctx, cc, false))
defer stopWatcher(ctx, prov)

messageChannel := make(chan *pb.Message)
errChan := make(chan *pb.Error, 1)
Expand Down Expand Up @@ -3136,6 +3199,7 @@ func Test_QueueSubscribe_NotifyCloseChannelsAreBuffered(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
assert.Nil(t, prov.Connect(ctx, cc, false))
defer stopWatcher(ctx, prov)

src := &pb.Source{
Name: "queue",
Expand Down
Loading