Skip to content

event: close idle streams when the final listener is removed #1223

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Rajvira10 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

event: close idle streams when the final listener is removed #1223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 55 additions & 29 deletions event.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ func (c *Client) RemoveEventListener(listener chan *APIEvents) error {
return err
}
if c.eventMonitor.listernersCount() == 0 {
c.eventMonitor.disableEventMonitoring()
c.eventMonitor.disableEventMonitoring(nil)
}
return nil
}
Expand Down Expand Up @@ -209,15 +209,19 @@ func (eventState *eventMonitoringState) enableEventMonitoring(c *Client, opts Ev
atomic.StoreInt64(&eventState.lastSeen, 0)
eventState.C = make(chan *APIEvents, 100)
eventState.errC = make(chan error, 1)
go eventState.monitorEvents(c, opts)
go eventState.monitorEvents(c, opts, eventState.C, eventState.errC)
}
return nil
}

func (eventState *eventMonitoringState) disableEventMonitoring() {
func (eventState *eventMonitoringState) disableEventMonitoring(eventChan chan *APIEvents) {
eventState.Lock()
defer eventState.Unlock()

if eventChan != nil && eventState.C != eventChan {
return
}

eventState.closeListeners()

eventState.Wait()
Expand All @@ -234,52 +238,56 @@ func (eventState *eventMonitoringState) disableEventMonitoring() {
}
}

func (eventState *eventMonitoringState) monitorEvents(c *Client, opts EventsOptions) {
func (eventState *eventMonitoringState) monitorEvents(c *Client, opts EventsOptions, eventChan chan *APIEvents, errChan chan error) {
const (
noListenersTimeout = 5 * time.Second
noListenersInterval = 10 * time.Millisecond
noListenersMaxTries = noListenersTimeout / noListenersInterval
)

var err error
for i := time.Duration(0); i < noListenersMaxTries && eventState.noListeners(); i++ {
for i := time.Duration(0); i < noListenersMaxTries && eventState.monitoring(eventChan) && eventState.noListeners(); i++ {
time.Sleep(10 * time.Millisecond)
}

if !eventState.monitoring(eventChan) {
return
}

if eventState.noListeners() {
// terminate if no listener is available after 5 seconds.
// Prevents goroutine leak when RemoveEventListener is called
// right after AddEventListener.
eventState.disableEventMonitoring()
eventState.disableEventMonitoring(eventChan)
return
}

if err = eventState.connectWithRetry(c, opts); err != nil {
if err = eventState.connectWithRetry(c, opts, eventChan, errChan); err != nil {
// terminate if connect failed
eventState.disableEventMonitoring()
eventState.disableEventMonitoring(eventChan)
return
}
for eventState.isEnabled() {
for eventState.monitoring(eventChan) {
timeout := time.After(100 * time.Millisecond)
select {
case ev, ok := <-eventState.C:
case ev, ok := <-eventChan:
if !ok {
return
}
if ev == EOFEvent {
go eventState.disableEventMonitoring()
go eventState.disableEventMonitoring(eventChan)
return
}
go func(ev *APIEvents) {
eventState.updateLastSeen(ev)
eventState.sendEvent(ev)
}(ev)
case err = <-eventState.errC:
case err = <-errChan:
if errors.Is(err, ErrNoListeners) {
eventState.disableEventMonitoring()
eventState.disableEventMonitoring(eventChan)
return
} else if err != nil {
defer func() { go eventState.monitorEvents(c, opts) }()
defer func() { go eventState.monitorEvents(c, opts, eventChan, errChan) }()
return
}
case <-timeout:
Expand All @@ -288,28 +296,39 @@ func (eventState *eventMonitoringState) monitorEvents(c *Client, opts EventsOpti
}
}

func (eventState *eventMonitoringState) connectWithRetry(c *Client, opts EventsOptions) error {
var retries int
eventState.RLock()
eventChan := eventState.C
errChan := eventState.errC
eventState.RUnlock()
closeConn, err := c.eventHijack(opts, atomic.LoadInt64(&eventState.lastSeen), eventChan, errChan)
for ; err != nil && retries < maxMonitorConnRetries; retries++ {
func (eventState *eventMonitoringState) connectWithRetry(c *Client, opts EventsOptions, eventChan chan *APIEvents, errChan chan error) error {
var closeConn func()
var err error
for retries := 0; retries <= maxMonitorConnRetries; retries++ {
if !eventState.monitoring(eventChan) {
return ErrNoListeners
}
closeConn, err = c.eventHijack(opts, atomic.LoadInt64(&eventState.lastSeen), eventChan, errChan)
if err == nil || retries == maxMonitorConnRetries {
break
}
waitTime := int64(retryInitialWaitTime * math.Pow(2, float64(retries)))
time.Sleep(time.Duration(waitTime) * time.Millisecond)
eventState.RLock()
eventChan = eventState.C
errChan = eventState.errC
eventState.RUnlock()
closeConn, err = c.eventHijack(opts, atomic.LoadInt64(&eventState.lastSeen), eventChan, errChan)
}
eventState.Lock()
defer eventState.Unlock()
if !eventState.enabled || eventState.C != eventChan {
if closeConn != nil {
closeConn()
}
return ErrNoListeners
}
eventState.closeConn = closeConn
return err
}

// monitoring reports whether the connection belongs to the current session.
func (eventState *eventMonitoringState) monitoring(eventChan chan *APIEvents) bool {
eventState.RLock()
defer eventState.RUnlock()
return eventState.enabled && eventState.C == eventChan
}

func (eventState *eventMonitoringState) noListeners() bool {
eventState.RLock()
defer eventState.RUnlock()
Expand Down Expand Up @@ -389,8 +408,8 @@ func (c *Client) eventHijack(opts EventsOptions, startTime int64, eventChan chan
keepRunning := int32(1)
//lint:ignore SA1019 the alternative doesn't quite work, so keep using the deprecated thing.
go func(res *http.Response, conn *httputil.ClientConn) {
defer conn.Close()
defer res.Body.Close()
defer conn.Close()
decoder := json.NewDecoder(res.Body)
for atomic.LoadInt32(&keepRunning) == 1 {
var event APIEvents
Expand All @@ -404,7 +423,12 @@ func (c *Client) eventHijack(opts EventsOptions, startTime int64, eventChan chan
c.eventMonitor.RUnlock()
break
}
errChan <- err
c.eventMonitor.RLock()
if c.eventMonitor.enabled && c.eventMonitor.C == eventChan {
errChan <- err
}
c.eventMonitor.RUnlock()
break
}
if event.Time == 0 {
continue
Expand All @@ -419,6 +443,8 @@ func (c *Client) eventHijack(opts EventsOptions, startTime int64, eventChan chan
}(res, conn)
return func() {
atomic.StoreInt32(&keepRunning, 0)
// Close the socket first to interrupt Decode without draining the body.
dial.Close()
}, nil
}

Expand Down
117 changes: 117 additions & 0 deletions event_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"os"
"slices"
"strings"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -316,3 +317,119 @@ func TestEventListenerReAdding(t *testing.T) {
// Give the goroutine of the first eventHijack() time to handle the EOF.
time.Sleep(10 * time.Millisecond)
}

func TestRemoveEventListenerClosesIdleConnection(t *testing.T) {
for name, newServer := range map[string]func(http.Handler) *httptest.Server{
"http": httptest.NewServer,
"tls": httptest.NewTLSServer,
} {
t.Run(name, func(t *testing.T) {
disconnected := make(chan struct{}, 2)
server := newServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(`{"status":"start","time":1}` + "\n"))
w.(http.Flusher).Flush()
<-r.Context().Done()
disconnected <- struct{}{}
}))
defer server.Close()
defer server.CloseClientConnections()
client, err := NewClient(server.URL)
if err != nil {
t.Fatal(err)
}
if server.TLS != nil {
client.TLSConfig = server.Client().Transport.(*http.Transport).TLSClientConfig
}
listener := make(chan *APIEvents, 1)
for range 2 {
if err := client.AddEventListener(listener); err != nil {
t.Fatal(err)
}
select {
case <-listener:
case <-time.After(5 * time.Second):
t.Fatal("did not receive event")
}
if err := client.RemoveEventListener(listener); err != nil {
t.Fatal(err)
}
select {
case <-disconnected:
case <-time.After(2 * time.Second):
t.Fatal("idle event connection remained open after removing the final listener")
}
}
})
}
}

func TestEventListenerReAddingDuringConnect(t *testing.T) {
firstRequest := make(chan struct{})
releaseFirst := make(chan struct{})
firstClosed := make(chan struct{})
secondClosed := make(chan struct{})
var requests atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
first := requests.Add(1) == 1
if first {
close(firstRequest)
<-releaseFirst
}
w.Write([]byte(`{"status":"start","time":1}` + "\n"))
w.(http.Flusher).Flush()
<-r.Context().Done()
if first {
close(firstClosed)
} else {
close(secondClosed)
}
}))
defer server.Close()
defer server.CloseClientConnections()
// Also release the handler if the test fails before the handshake completes.
defer func() {
select {
case <-releaseFirst:
default:
close(releaseFirst)
}
}()
client, err := NewClient(server.URL)
if err != nil {
t.Fatal(err)
}
listener := make(chan *APIEvents, 1)
if err := client.AddEventListener(listener); err != nil {
t.Fatal(err)
}
select {
case <-firstRequest:
case <-time.After(5 * time.Second):
t.Fatal("first request did not start")
}
if err := client.RemoveEventListener(listener); err != nil {
t.Fatal(err)
}
if err := client.AddEventListener(listener); err != nil {
t.Fatal(err)
}
select {
case <-listener:
case <-time.After(5 * time.Second):
t.Fatal("new listener did not receive event")
}
close(releaseFirst)
select {
case <-firstClosed:
case <-time.After(2 * time.Second):
t.Fatal("connection from the previous session remained open")
}
if err := client.RemoveEventListener(listener); err != nil {
t.Fatal(err)
}
select {
case <-secondClosed:
case <-time.After(2 * time.Second):
t.Fatal("current session connection remained open")
}
}