49 lines
920 B
Go
49 lines
920 B
Go
package events
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
)
|
|
|
|
func TestBroker_DeliversToAllSubscribers(t *testing.T) {
|
|
b := NewBroker()
|
|
a, cancelA := b.Subscribe()
|
|
defer cancelA()
|
|
c, cancelC := b.Subscribe()
|
|
defer cancelC()
|
|
|
|
b.Publish(Services)
|
|
assert.Equal(t, Services, <-a)
|
|
assert.Equal(t, Services, <-c)
|
|
}
|
|
|
|
func TestBroker_CancelStopsDelivery(t *testing.T) {
|
|
b := NewBroker()
|
|
ch, cancel := b.Subscribe()
|
|
cancel()
|
|
cancel() // safe to call twice
|
|
b.Publish(Agents)
|
|
assert.Empty(t, ch)
|
|
}
|
|
|
|
func TestBroker_SlowSubscriberDoesNotBlockPublish(t *testing.T) {
|
|
b := NewBroker()
|
|
_, cancel := b.Subscribe()
|
|
defer cancel()
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
for range 1000 {
|
|
b.Publish(Services) // must return even though nobody reads
|
|
}
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("Publish blocked on a slow subscriber")
|
|
}
|
|
}
|