diff --git a/event.go b/event.go index dbe1c353..f3264ba1 100644 --- a/event.go +++ b/event.go @@ -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 } @@ -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() @@ -234,7 +238,7 @@ 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 @@ -242,44 +246,48 @@ func (eventState *eventMonitoringState) monitorEvents(c *Client, opts EventsOpti ) 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: @@ -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() @@ -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 @@ -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 @@ -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 } diff --git a/event_test.go b/event_test.go index 7a48b1c6..983c68d9 100644 --- a/event_test.go +++ b/event_test.go @@ -13,6 +13,7 @@ import ( "os" "slices" "strings" + "sync/atomic" "testing" "time" @@ -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") + } +}