Notify server when client connections are closed

handlers/events_handlers.go

@@ -78,15 +78,19 @@ func streamEventsToResponse(logger lager.Logger, w http.ResponseWriter, eventCha
 
 	var event models.Event
 	eventID := 0
-	closeNotifier := w.(http.CloseNotifier).CloseNotify()
+	done := make(chan bool, 1)
+	go func() {
+		rw.ReadFrom(conn)
+		done <- true
+	}()
 
 	for {
 		select {
 		case event = <-eventChan:
 		case err := <-errorChan:
 			logger.Error("failed-to-get-next-event", err)
 			return
-		case <-closeNotifier:
+		case <-done:
 			logger.Debug("received-close-notify")
 			return
 		}

handlers/events_handlers_test.go

@@ -18,6 +18,7 @@ import (
 	"code.cloudfoundry.org/lager/lagertest"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/onsi/gomega/gbytes"
 	"github.com/vito/go-sse/sse"
 )
 
@@ -133,6 +134,23 @@ var _ = Describe("Event Handlers", func() {
 					Expect(response.Header.Get("Cache-Control")).To(Equal("no-cache, no-store, must-revalidate"))
 				})
 
+				It("detects closed clients on the server", func() {
+					reader := sse.NewReadCloser(response.Body)
+
+					hub.Emit(&eventfakes.FakeEvent{Token: "A"})
+					encodedPayload := base64.StdEncoding.EncodeToString([]byte("A"))
+
+					Expect(reader.Next()).To(Equal(sse.Event{
+						ID:   "0",
+						Name: "fake",
+						Data: []byte(encodedPayload),
+					}))
+
+					reader.Close()
+
+					Eventually(logger).Should(gbytes.Say("received-close-notify"))
+				})
+
 				Context("when the source provides an unmarshalable event", func() {
 					It("closes the event stream to the client", func() {
 						hub.Emit(eventfakes.UnmarshalableEvent{Fn: func() {}})