diff --git a/go.mod b/go.mod index aa376f2..ec64693 100644 --- a/go.mod +++ b/go.mod @@ -1,9 +1,15 @@ module github.com/r3labs/sse/v2 -go 1.13 +go 1.25.0 require ( github.com/stretchr/testify v1.7.0 - golang.org/x/net v0.0.0-20191116160921-f9c825593386 // indirect gopkg.in/cenkalti/backoff.v1 v1.1.0 ) + +require ( + github.com/davecgh/go-spew v1.1.0 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + golang.org/x/net v0.55.0 // indirect + gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect +) diff --git a/go.sum b/go.sum index 31187d6..a084334 100644 --- a/go.sum +++ b/go.sum @@ -5,11 +5,8 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= -golang.org/x/net v0.0.0-20191116160921-f9c825593386 h1:ktbWvQrW08Txdxno1PiDpSxPXG6ndGsfnJjRRtkM0LQ= -golang.org/x/net v0.0.0-20191116160921-f9c825593386/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= -golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= -golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= gopkg.in/cenkalti/backoff.v1 v1.1.0 h1:Arh75ttbsvlpVA7WtVpH4u9h6Zl46xuptxqLxPiSo4Y= gopkg.in/cenkalti/backoff.v1 v1.1.0/go.mod h1:J6Vskwqd+OMVJl8C33mmtxTBs2gyzfv7UDAkHu8BrjI= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= diff --git a/server.go b/server.go index d1b27af..4094f70 100644 --- a/server.go +++ b/server.go @@ -29,6 +29,10 @@ type Server struct { AutoStream bool // Enables automatic replay for each new subscriber that connects AutoReplay bool + // Evicts a subscriber immediately when its event buffer is full instead of + // blocking the stream's fan-out goroutine on it. Applied to streams at + // creation time; changing it afterwards does not affect existing streams. + EvictSlowClients bool // Specifies the function to run when client subscribe or un-subscribe OnSubscribe func(streamID string, sub *Subscriber) @@ -82,7 +86,7 @@ func (s *Server) CreateStream(id string) *Stream { return s.streams[id] } - str := newStream(id, s.BufferSize, s.AutoReplay, s.AutoStream, s.OnSubscribe, s.OnUnsubscribe) + str := newStream(id, s.BufferSize, s.AutoReplay, s.AutoStream, s.EvictSlowClients, s.OnSubscribe, s.OnUnsubscribe) str.run() s.streams[id] = str diff --git a/stream.go b/stream.go index bfbcb9b..b3608a1 100644 --- a/stream.go +++ b/stream.go @@ -25,25 +25,33 @@ type Stream struct { AutoReplay bool isAutoStream bool + // evictSlowClients controls behaviour when a subscriber's event buffer is full. + // If true, the subscriber is immediately removed rather than blocking the + // fan-out goroutine. This prevents one stalled client from freezing all other + // connected clients. Set once at stream creation; never mutated afterwards, + // since run() reads it from its own goroutine without synchronization. + evictSlowClients bool + // Specifies the function to run when client subscribe or un-subscribe OnSubscribe func(streamID string, sub *Subscriber) OnUnsubscribe func(streamID string, sub *Subscriber) } // newStream returns a new stream -func newStream(id string, buffSize int, replay, isAutoStream bool, onSubscribe, onUnsubscribe func(string, *Subscriber)) *Stream { +func newStream(id string, buffSize int, replay, isAutoStream, evictSlowClients bool, onSubscribe, onUnsubscribe func(string, *Subscriber)) *Stream { return &Stream{ - ID: id, - AutoReplay: replay, - subscribers: make([]*Subscriber, 0), - isAutoStream: isAutoStream, - register: make(chan *Subscriber), - deregister: make(chan *Subscriber), - event: make(chan *Event, buffSize), - quit: make(chan struct{}), - Eventlog: make(EventLog, 0), - OnSubscribe: onSubscribe, - OnUnsubscribe: onUnsubscribe, + ID: id, + AutoReplay: replay, + subscribers: make([]*Subscriber, 0), + isAutoStream: isAutoStream, + evictSlowClients: evictSlowClients, + register: make(chan *Subscriber), + deregister: make(chan *Subscriber), + event: make(chan *Event, buffSize), + quit: make(chan struct{}), + Eventlog: make(EventLog, 0), + OnSubscribe: onSubscribe, + OnUnsubscribe: onUnsubscribe, } } @@ -60,22 +68,27 @@ func (str *Stream) run() { // Remove closed subscriber case subscriber := <-str.deregister: - i := str.getSubIndex(subscriber) - if i != -1 { - str.removeSubscriber(i) - } - - if str.OnUnsubscribe != nil { - go str.OnUnsubscribe(str.ID, subscriber) - } + str.unsubscribe(subscriber) // Publish event to subscribers case event := <-str.event: if str.AutoReplay { str.Eventlog.Add(event) } - for i := range str.subscribers { - str.subscribers[i].connection <- event + for i := len(str.subscribers) - 1; i >= 0; i-- { + if str.evictSlowClients { + // Non-blocking: if the buffer is full, evict the subscriber + // immediately rather than blocking the fan-out goroutine. + // unsubscribe is used instead of close() to avoid + // deadlocking run() on its own deregister channel. + select { + case str.subscribers[i].connection <- event: + default: + str.unsubscribe(str.subscribers[i]) + } + } else { + str.subscribers[i].connection <- event + } } // Shutdown if the server closes @@ -126,6 +139,22 @@ func (str *Stream) addSubscriber(eventid int, url *url.URL) *Subscriber { return sub } +// unsubscribe removes sub if it is still present and, only in that case, +// notifies OnUnsubscribe. Shared by the deregister path and slow-client +// eviction so a given subscriber is never reported removed more than once. +func (str *Stream) unsubscribe(sub *Subscriber) { + i := str.getSubIndex(sub) + if i == -1 { + return + } + + str.removeSubscriber(i) + + if str.OnUnsubscribe != nil { + go str.OnUnsubscribe(str.ID, sub) + } +} + func (str *Stream) removeSubscriber(i int) { atomic.AddInt32(&str.subscriberCount, -1) close(str.subscribers[i].connection) diff --git a/stream_test.go b/stream_test.go index 1c89a6e..a359845 100644 --- a/stream_test.go +++ b/stream_test.go @@ -16,7 +16,7 @@ import ( // Maybe fix this in the future so we can test with -race enabled func TestStreamAddSubscriber(t *testing.T) { - s := newStream("test", 1024, true, false, nil, nil) + s := newStream("test", 1024, true, false, false, nil, nil) s.run() defer s.close() @@ -34,7 +34,7 @@ func TestStreamAddSubscriber(t *testing.T) { } func TestStreamRemoveSubscriber(t *testing.T) { - s := newStream("test", 1024, true, false, nil, nil) + s := newStream("test", 1024, true, false, false, nil, nil) s.run() defer s.close() @@ -47,7 +47,7 @@ func TestStreamRemoveSubscriber(t *testing.T) { } func TestStreamSubscriberClose(t *testing.T) { - s := newStream("test", 1024, true, false, nil, nil) + s := newStream("test", 1024, true, false, false, nil, nil) s.run() defer s.close() @@ -59,7 +59,7 @@ func TestStreamSubscriberClose(t *testing.T) { } func TestStreamDisableAutoReplay(t *testing.T) { - s := newStream("test", 1024, true, false, nil, nil) + s := newStream("test", 1024, true, false, false, nil, nil) s.run() defer s.close() @@ -74,7 +74,7 @@ func TestStreamDisableAutoReplay(t *testing.T) { func TestStreamMultipleSubscribers(t *testing.T) { var subs []*Subscriber - s := newStream("test", 1024, true, false, nil, nil) + s := newStream("test", 1024, true, false, false, nil, nil) s.run() for i := 0; i < 10; i++ {