Repinoid
e83d804501
Параллельный enqueue + закрытие канала не должен паниковать «send on closed channel». go test -race ./... зелёный.
46 lines
1.4 KiB
Go
46 lines
1.4 KiB
Go
// sender_test.go — тест безопасности enqueue ↔ closeAndWait (фикс 🔴3).
|
|
//
|
|
// Раньше поздний колбэк paho мог упасть «send on closed channel» при закрытии
|
|
// канала во время enqueue. Мьютекс + closed-флаг должны это исключать.
|
|
package bridge
|
|
|
|
import (
|
|
"io"
|
|
"log/slog"
|
|
"sync"
|
|
"testing"
|
|
)
|
|
|
|
func TestDispatcherEnqueueCloseNoPanic(t *testing.T) {
|
|
log := slog.New(slog.NewTextHandler(io.Discard, nil))
|
|
// client nil — enqueue/closeAndWait его не трогают (только worker'ы, которых не запускаем).
|
|
d := newSQSDispatcher(nil, "http://queue", log)
|
|
|
|
// Параллельный enqueue, пока основной поток закрывает канал.
|
|
var wg sync.WaitGroup
|
|
stop := make(chan struct{})
|
|
for i := 0; i < 4; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for {
|
|
select {
|
|
case <-stop:
|
|
return
|
|
default:
|
|
d.enqueue(TelemetryEnvelope{Namespace: "ns", DeviceID: "d"})
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
d.closeAndWait()
|
|
close(stop)
|
|
wg.Wait()
|
|
|
|
// После закрытия enqueue обязан вернуть false (канал закрыт), а не паниковать.
|
|
if d.enqueue(TelemetryEnvelope{Namespace: "ns", DeviceID: "d"}) {
|
|
t.Error("enqueue после close должен вернуть false")
|
|
}
|
|
}
|