Feature/evict slow clients

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
+)

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=

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

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)

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++ {