diff --git a/go.mod b/go.mod index 32bcfb5..782e8ed 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.25.0 require ( github.com/ThreeDotsLabs/watermill v1.5.2 - github.com/go-chi/chi/v5 v5.2.5 + github.com/go-chi/chi/v5 v5.3.0 github.com/go-chi/render v1.0.3 github.com/pkg/errors v0.9.1 github.com/stretchr/testify v1.11.1 diff --git a/go.sum b/go.sum index 0f2f296..b006e78 100644 --- a/go.sum +++ b/go.sum @@ -7,6 +7,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug= github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0= +github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM= +github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto= github.com/go-chi/render v1.0.3 h1:AsXqd2a1/INaIfUSKq3G5uA8weYx20FOsM7uSoCyyt4= github.com/go-chi/render v1.0.3/go.mod h1:/gr3hVkmYR0YlEy3LxCuVRFzEu9Ruok+gFqbIofjao0= github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= diff --git a/pkg/http/sse.go b/pkg/http/sse.go index 251ddd5..514db03 100644 --- a/pkg/http/sse.go +++ b/pkg/http/sse.go @@ -3,6 +3,7 @@ package http import ( "context" "net/http" + "sync" "github.com/go-chi/render" "github.com/pkg/errors" @@ -169,13 +170,25 @@ func (h sseHandler) handleEventStream(w http.ResponseWriter, r *http.Request) { responsesChan := make(chan interface{}) + // The producer goroutine reads from r (including chi.URLParam via the + // StreamAdapter). chi pools and resets the request's *Context once + // ServeHTTP returns, so the goroutine must not outlive this handler. + // wg.Wait below keeps ServeHTTP blocked until the goroutine exits. + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() defer func() { h.logger.Trace("Closing SSE handler", nil) close(responsesChan) }() - responsesChan <- response + select { + case responsesChan <- response: + case <-r.Context().Done(): + return + } h.logger.Trace("Listening for messages", nil) @@ -189,16 +202,15 @@ func (h sseHandler) handleEventStream(w http.ResponseWriter, r *http.Request) { msg.Ack() nextResponse, ok := h.streamAdapter.NextStreamResponse(r, msg) + if !ok { + continue + } + h.logger.Trace("Stream responding on message", watermill.LogFields{"uuid": msg.UUID}) select { + case responsesChan <- nextResponse: case <-r.Context().Done(): return - default: - } - - if ok { - h.logger.Trace("Stream responding on message", watermill.LogFields{"uuid": msg.UUID}) - responsesChan <- nextResponse } case <-r.Context().Done(): return @@ -210,4 +222,8 @@ func (h sseHandler) handleEventStream(w http.ResponseWriter, r *http.Request) { marshaler: h.config.Marshaler, } responder.Respond(w, r, responsesChan) + + // Block ServeHTTP until the producer goroutine has fully exited, so chi + // never resets the request context while the goroutine still reads r. + wg.Wait() }