Detect disconnected clients

[frederikhors]

Somethining I think we need ASAP is to detect clients disconnected both with eventSource.close() or without.

If I use this code:

func newSSEHandler(ssePingInterval time.Duration) *sse.Server {
	sseServer := sse.New()

	sseServer.AutoReplay = false

	sseServer.AutoStream = true

	go func() {
		for range time.Tick(ssePingInterval) {
			for i := range sseServer.Streams {
				println(i, "stream is still present...")
				sseServer.Publish(i, &sse.Event{Data: []byte("test")})
			}
		}
	}()

	return sseServer
}

even if my only client (a browser) closes the connection (with eventSource.close()), the server does not notice even after several minutes and does not close the stream.

[codewinch]

This happens at a higher level than this library. Perhaps you can read up a bit on how context works: https://pkg.go.dev/context#Context (EDIT: there's a great link in that section, https://go.dev/blog/pipelines )

The easiest way is, in your handler, to start up a goroutine that just waits for

go func() {
     <-ctx.Done() // throw away the value received
     shutdown() // this can be another function inside the same handler
}

[frederikhors]

Oh wow. Can you please help me understand where I can check for context to be Done? In that loop I wrote?

[codewinch]

No, I don't think issues are for asking questions. If you want to propose a documentation update, please do that instead.

[frederikhors]

If you want to propose a documentation update, please do that instead.

How?

I mean, I'm making PR for this project. And this is something we need to address because we can garbage out our Streams slice.

[codewinch]

Sorry if I came across as rude. Without seeing your entire app, it's unclear to me how/when sseServer gets instantiated (during server startup?), how you're tracking the streams, how your http router is accepting streams and passing off to the handler, etc.

I am suggesting that you might want to close this issue to create a new issue (or change this issue) to request a simple example of how to manage client disconnects to be added to the README.

[frederikhors]

Ok, I understand what you mean that I do for this issue and I will do it as soon as I understand if it can already be done now or if it is something that is missing and that must be written.

In the meantime, I'll paste the code of my simple app, a few lines as you can see.

package main

import (
	"github.com/r3labs/sse/v2"
	"github.com/go-chi/chi/v5"
	"github.com/go-chi/chi/v5/middleware"
)

func main() {
	sseServer := NewSSEHandler()

	router := chi.NewRouter()

	router.Use(middleware.Logger, middleware.Recoverer)
  
	router.Group(func(r chi.Router) {
		r.Get("/sse", sseServer.HTTPHandler)
	})

  httpServer := &http.Server{
		Addr:    ":3000",
		Handler: router,
		// ReadTimeout:  serverReadTimeout,
		// WriteTimeout: serverWriteTimeout,
	}

  err := httpServer.ListenAndServe()
  if !errors.Is(err, http.ErrServerClosed) {
    panic(err)
  }
}

func NewSSEHandler() *sse.Server {
	sseServer := sse.New()

	sseServer.AutoReplay = false

	sseServer.AutoStream = true

	go func() {
		for range time.Tick(10 * time.Second) {
			for i := range sseServer.Streams {
				println(time.Now().Unix(), i, "stream is still present...")

				sseServer.Publish(i, &sse.Event{Data: []byte("test")})
			}
		}
	}()

	return sseServer
}

With this simple code I need to detect (not necessarily immediately, I can wait even a few seconds and even a few minutes if necessary) when a client disconnects both with .close() method and when browser window is closed.

This not only for the ping loop but for many other reasons.

@codewinch is still valid your teory to use Go's context in this case? If yes, how? I will create a PR for docs for this.

[codewinch]

Hi @frederikhors , this is how I do it within my resource handler from my route:

go func() {
    // Received Browser Disconnection
    <-ctx.Done()
    shutDown(errors.New("Exit"))
    return
}()

[frederikhors]

Ok. I don't understand where do you put that code. In the router handler?

[codewinch]

I am calling sseServer.HTTPHandler(w,r) at the end of my own handler, where I handle all of these housekeeping items myself.

For example, instead of what you have here:

r.Get("/sse", sseServer.HTTPHandler)

Mine is more like:

r.Get("/sse/:various/:things", function() error {

   // do various things like authentication

  func shutDown(err error) {
        // do things to tear down connections,
        // close database conns, etc.
  }

  go func() {
        // Received Browser Disconnection
        <-ctx.Done()
        shutDown(errors.New("Exit"))
        return
  }()
 
  sseServer.HTTPHandler(w,r)

  return nil
})

[frederikhors]

Are you declaring func shutDown(err error) { inside the func?

[frederikhors]

And do you mean <-request.Context().Done()?

[frederikhors]

I think I understand what you mean and I'm trying with the eventsource.close() method, but it doesn't work. I never get the <-r.Context().Done() (I put a println("test") right after it and it doesn't appear).

[frederikhors]

It works. Thanks. And the amazing work of @purehyperbole is also removing the subscribers from the Stream! I am fascinated by it! Amazing!

The thing to fix now is the auto-elimination of the stream when it is no longer needed, the opposite of AutoStream = true in short.

[codewinch]

And do you mean <-request.Context().Done()?

it depends on your framework; often frameworks wrap the "real" http context as described within the link above within their own, and you might need to access it by something like that.

Are you declaring func shutDown(err error) { inside the func?

Yes, so it has access to all of the surrounding private variables without needing them to be passed in, but you could make it a separate function if it was more convenient (or if that outer function was getting too big!)

It works. Thanks.

That's great!! Can you close this issue then and possibly propose an update to the README? Thanks Frederik!