diff --git a/internal/broadcasters.go b/internal/broadcasters.go index cde0056b..c3ddc09d 100644 --- a/internal/broadcasters.go +++ b/internal/broadcasters.go @@ -51,11 +51,18 @@ func (p *channelPair[V]) send(value V) { } } -// close stops any send in progress and closes sendCh. Call it once per subscriber. -func (p *channelPair[V]) close() { +// close stops any send in progress and closes sendCh. If final is not nil, close first sends *final to +// sendCh if it has room. Call it once per subscriber. +func (p *channelPair[V]) close(final *V) { close(p.done) p.lock.Lock() defer p.lock.Unlock() + if final != nil { + select { + case p.sendCh <- *final: + default: + } + } close(p.sendCh) } @@ -92,7 +99,7 @@ func (b *Broadcaster[V]) RemoveListener(ch <-chan V) { }) b.lock.Unlock() for _, pair := range removed { - pair.close() + pair.close(nil) } } @@ -115,11 +122,21 @@ func (b *Broadcaster[V]) Broadcast(value V) { // Close closes all current subscriber channels. func (b *Broadcaster[V]) Close() { + b.closeAll(nil) +} + +// CloseWith sends value to each current subscriber that has room for it, then closes all current subscriber +// channels. It does not wait for a subscriber to read. +func (b *Broadcaster[V]) CloseWith(value V) { + b.closeAll(&value) +} + +func (b *Broadcaster[V]) closeAll(final *V) { b.lock.Lock() subscribers := b.subscribers b.subscribers = nil b.lock.Unlock() for _, pair := range subscribers { - pair.close() + pair.close(final) } } diff --git a/internal/broadcasters_test.go b/internal/broadcasters_test.go index bd541d5c..03a6a992 100644 --- a/internal/broadcasters_test.go +++ b/internal/broadcasters_test.go @@ -174,6 +174,43 @@ func testBroadcasterGenerically[V any](t *testing.T, broadcasterFactory func() * }) } + t.Run("close with value delivers it to a listener with room", func(t *testing.T) { + withBroadcaster(t, func(b *Broadcaster[V]) { + ch := b.AddListener() + + // Close with a final value. + value := valueFactory() + b.CloseWith(value) + + // The listener should receive the value, and then the channel should close. + assert.Equal(t, value, th.RequireValue(t, ch, timeout)) + th.AssertChannelClosed(t, ch, timeout) + }) + }) + + t.Run("close with value does not wait for a full listener", func(t *testing.T) { + withBroadcaster(t, func(b *Broadcaster[V]) { + ch := b.AddListener() + for i := 0; i < subscriberChannelBufferLength; i++ { + b.Broadcast(valueFactory()) + } + + // Close with a final value without reading from the listener. + closeDone := make(chan struct{}) + go func() { + b.CloseWith(valueFactory()) + close(closeDone) + }() + + // CloseWith should return, and the listener should get only the values that fit. + th.AssertChannelClosed(t, closeDone, timeout) + for i := 0; i < subscriberChannelBufferLength; i++ { + th.RequireValue(t, ch, timeout) + } + th.AssertChannelClosed(t, ch, timeout) + }) + }) + t.Run("hasListeners", func(t *testing.T) { withBroadcaster(t, func(b *Broadcaster[V]) { assert.False(t, b.HasListeners()) diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go index 9e0a3eeb..2d8723cb 100644 --- a/internal/datasystem/fdv2_datasystem.go +++ b/internal/datasystem/fdv2_datasystem.go @@ -27,8 +27,9 @@ type broadcasters struct { changeSetBroadcaster *internal.Broadcaster[subsystems.ChangeSet] } -func (b *broadcasters) Close() { - b.dataSourceStatus.Close() +// Close sends status to each data source status listener that has room for it, then closes every broadcaster. +func (b *broadcasters) Close(status interfaces.DataSourceStatus) { + b.dataSourceStatus.CloseWith(status) b.dataStoreStatus.Close() b.flagChangeEvent.Close() b.changeSetBroadcaster.Close() @@ -682,10 +683,9 @@ func (f *FDv2) Stop() error { f.cancel() } f.wg.Wait() - // Do not broadcast Off. Status listeners learn of shutdown when their channels close. - _, _ = f.maybeUpdateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{}) + status, _ := f.maybeUpdateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{}) _ = f.store.Close() - f.broadcasters.Close() + f.broadcasters.Close(status) return nil } diff --git a/internal/datasystem/fdv2_datasystem_test.go b/internal/datasystem/fdv2_datasystem_test.go index 631d1244..5d0aa959 100644 --- a/internal/datasystem/fdv2_datasystem_test.go +++ b/internal/datasystem/fdv2_datasystem_test.go @@ -7,6 +7,7 @@ import ( th "github.com/launchdarkly/go-test-helpers/v3" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/launchdarkly/go-server-sdk/v7/interfaces" "github.com/launchdarkly/go-server-sdk/v7/subsystems" @@ -69,6 +70,34 @@ func TestFDv2WaitForEndsWhenStopped(t *testing.T) { assert.Less(t, time.Since(timeStart), 500*time.Millisecond) } +func TestFDv2StatusListenerReceivesOffWhenStopped(t *testing.T) { + system := makeFallbackTestFDv2(t) + provider := &dataStatusProvider{system: system} + listener := provider.AddStatusListener() + + // Stop the system. + require.NoError(t, system.Stop()) + + // The listener should receive Off, and then its channel should close. + assert.Equal(t, interfaces.DataSourceStateOff, th.RequireValue(t, listener, time.Second).State) + th.AssertChannelClosed(t, listener, time.Second) +} + +func TestFDv2WaitForOffReturnsTrueWhenStopped(t *testing.T) { + system := makeFallbackTestFDv2(t) + provider := &dataStatusProvider{system: system} + go func() { + <-time.After(100 * time.Millisecond) + _ = system.Stop() + }() + + // Wait for Off while the system stops. + success := provider.WaitFor(interfaces.DataSourceStateOff, time.Second) + + // WaitFor should see Off. + assert.True(t, success) +} + func TestFDv2GetStatusReturnsWhileStatusListenerIsFull(t *testing.T) { system := makeFallbackTestFDv2(t) t.Cleanup(func() { _ = system.Stop() })