How to dispatch messages to individual clients?

[hspaay]

Edit from maintainer: this was originally an issue with the title "Race condition in server using sub.Client.Send()", which was subsequently deemed to be a confusion on how to use the library. Discussion from issue left here for reference. The library primitives referenced in the issue discussion are part of v0.8.0 of go-sse – they may be subject to change or removal in the future. The ideas and approaches presented here should still remain relevant even after future updates.

There seems to be a race condition sending messages from the server to the client.

In sse.Server OnSession a subscription is created as follow:

gosse := &sse.Server{}
gosse.OnSession = func(sseSession *sse.Session) (sse.Subscription, bool) {
sub := sse.Subscription{
	Client:      sseSession,
	LastEventID: sseSession.LastEventID,
	Topics:      []string{sse.DefaultTopic},
   }
}

In a separate gorouting, messages are send from the server to this client:

sseMsg := sse.Message{}
sseMsg.AppendData(myPayload)
sseMsg.ID, _ = sse.NewID(eventID)
sseMsg.Type, err = sse.NewType(eventType)
err = sub.Client.Send(&sseMsg)
_ = sub.Client.Flush()

This works most of the time, but running it with go test -race exposes a race condition:

WARNING: DATA RACE
Write at 0x00c0001764d2 by goroutine 380:
  net/http.(*http2responseWriter).handlerDone()
      /snap/go/10627/src/net/http/h2_bundle.go:6860 +0x6d
  net/http.(*http2serverConn).runHandler.func1()
      /snap/go/10627/src/net/http/h2_bundle.go:6230 +0x404
  runtime.deferreturn()
      /snap/go/10627/src/runtime/panic.go:477 +0x30
  net/http.(*http2serverConn).scheduleHandler.func1()
      /snap/go/10627/src/net/http/h2_bundle.go:6167 +0x5d

Previous read at 0x00c0001764d2 by goroutine 381:
  net/http.(*http2responseWriterState).writeChunk()
      /snap/go/10627/src/net/http/h2_bundle.go:6562 +0xa84
  net/http.http2chunkWriter.Write()
      /snap/go/10627/src/net/http/h2_bundle.go:6428 +0x48
  bufio.(*Writer).Flush()
      /snap/go/10627/src/bufio/bufio.go:642 +0xf0
  net/http.(*http2responseWriter).FlushError()
      /snap/go/10627/src/net/http/h2_bundle.go:6693 +0x9e
  github.com/tmaxmax/go-sse.flusherErrorWrapper.Flush()
      /home/henk/go/pkg/mod/github.com/tmaxmax/go-sse@v0.9.0-pre.2/session.go:160 +0x46
  github.com/tmaxmax/go-sse.(*flusherErrorWrapper).Flush()
      <autogenerated>:1 +0x17
  github.com/tmaxmax/go-sse.(*Session).Flush()
      /home/henk/go/pkg/mod/github.com/tmaxmax/go-sse@v0.9.0-pre.2/session.go:61 +0x83
  github.com/hiveot/hub/runtime/transports/httpstransport.(*HttpsTransport).Start.NewSSEServer.NewGoSSEServer.func1.1()
      /home/henk/dev/hiveot/hub/runtime/transports/httpstransport/sseserver/GoSseServer.go:58 +0x57a

The documentation of the Send function states that the interface does not have to be thread-safe:
(server.go:26)

// The Subscription struct is used to subscribe to a given provider.
type Subscription struct {
	// The client to which messages are sent. The implementation of the interface does not have to be
	// thread-safe – providers will not call methods on it concurrently.
	Client MessageWriter

[tmaxmax]

sse.Subscription.Client.Send is not meant to be called from outside a Provider. The idea is that this is a low-level building primitive for providers, in that a provider makes sure access to a subscription client is synchronized. This seems like misusage of the library.

It would be helpful if you could provide more context to your use case (and maybe more code) – what are you trying to achieve? There should be a solution which doesn't imply a direct Send.

Apologies for the late response.

[hspaay]

"sse.Subscription.Client.Send is not meant to be called from outside a Provider."

Ooh okay.

What I'm trying to accomplish is to have a server send an sse event to a specific client. The server code with the subscription can be found here: https://github.com/hiveot/hub/blob/main/runtime/transports/httpstransport/sseserver/GoSseServer.go

In case you like some more context:

• project: https://github.com/hiveot/hub
• overview: https://github.com/hiveot/hub/blob/main/docs/design-overview.md
• message flow: https://github.com/hiveot/hub/blob/main/docs/event-flow.jpg

This project has a runtime that accepts requests from clients using https connections, and uses SSE as a return channel. The clients can be IoT devices, web users or other services. When a client establishes a SSE connections, the server is notified using the go-sse onSession callback (line 12). It obtains the runtime session (line 15) for this client and requests a channel from this session (line 36). A go-sse SSE subscription is created to be able to send messages to the client (line 28). Messages received on the session channel (line 41) are pushed to the client (line 53).

The go-sse examples show a broadcast solution and I didn't see an example of sending an SSE event to a specific client. Hence the use of subscription.

If there is a better way to send messages to an individual sse connection, then I'd be happy to change this to follow the prescribed approach.

[tmaxmax]

A simple way to achieve this with the current design of the library is to assign each client to its own topic. For each IoT device/user/service you should find a way to generate a unique ID and use that as the topic of the event. For example, what you could do is to generate a new UUID for every new request. With OnSession it would look like:

s := sse.Server{
    OnSession: func(s *sse.Session) (sse.Subscriber, bool) {
        id := uuid.New() // some UUID library of choice
        // or fetch ID from a database using the s.Req data, or other method

        return sse.Subscriber{
            Client: s,
            LastEventID: s.LastEventID,
            Topics: []string{id.String()},
        }, true
    }
}

If you have multiple topics you want a client to be subscribed to, you can easily emulate that:

requestedTopics := getTopics(s.Req) // get the topics from the request in some way
topics := make([]string, 0, len(requestedTopics))
for _, rt := range requestedTopics {
    // generate a unique topic name using the client's ID and the topic names it wishes to subscribe to.
    topics = append(topics, id.String() + ">>" + rt)
}
return sse.Subscription{
    // ...
    Topics: topics
}, true

In order to find these IDs later, you should keep a thread-safe registry of them somewhere or have a deterministic manner of regenerating them. If you have a database with all the clients, for example, you could retrieve the ID by some unique device name which you obtain through other means. The issue of generating these IDs changes on a case-by-case basis – you should find out the way you can obtain that in your product.

If you also want to send a message to all clients subscribed to a specific topic, you may also hold a mapping between topics and all the IDs of the clients subscribed to that topic. This should also be easy if you already have a database with the users/devices or have a reliable deterministic way of generating the IDs.

This is how I'd approach the issue. It should be the least amount of work because it enables you to leverage both the already existing go-sse infrastructure (no need to rewrite neither the HTTP handler nor the provider) and also that of your own (your database and domain specific knowledge).

Please let me know if this helps you find a way forward with your problem! Given that this was an issue beforehand and the thread has some other unnecessary comments, please mark as answer if it does help, for discoverability.

[hspaay]

After some more experimentation I found that this approach unfortunately does not work, as it isn't interoperable with other sse client implementations. Specifically htmx with sse. For example:

A user interface using golang templates and htmx with sse displays the value of a single IoT property as follows:

<span sse-swap="{{$.ThingID}}/{{$.PropName}}"
      hx-swap="innerHTML"
      hx-target="this"
>
</span>

The golang backend publishes sse events with the eventName: ThingID/PropertyName.
The frontend uses the browser htmx with sse extension. There is no subscription.

The frontend should only receive events which it is authorized to receive based on its authentication. This requires the ability for the backend to selectively send events to clients without using subscriptions (that uses event names). The event names are already in use to determine which information to display.

Hope that explains the need for the backend to send events to individual connections. The concept of subscription is an application level decision that doesn't always apply. Selecting the sse connection to send events on is a transport level decision.

[tmaxmax]

I see, thank you for the breakdown!

I'll look at this in more detail at the end of the week to gain a better understanding and see what should be done here. This is a use-case which go-sse should support in some way.

Many thanks for the repeated contributions through issues and discussions! I'll be back with an answer.

[hspaay]

If I may make a suggestion, it could help to split the code. As the core the pure transport with an API that can be freely used to send events to individual connections, using just eventName, data and id. Additional 'plugins' can then be added on top of this core to enhance functionality such as subscription based messaging.

If the plugin api's can be kept identical then maybe it would be possible to add a middleware type chain of plugins to 'stack' functionality. For example, have plugins for logging, subscription, filtering, authorization, encoding of server sent events.

Lets say the plugin API is (nothing revolutionary):

send(eventName string, id string, payload string) error

The core and all plugins implement this api so the core can be used by the application with and without any plugins.

The go-sse server would provide an API to registering plugin instances and pass any events being published through the chain until the core is reached. Perhaps provide multiple ways to instantiate the server, one without any plugins and one with default plugins for encoding and logging. Additional plugins can be added at any time. I don't see a need of removing plugins.

One use-case to consider is to allow plugin selection after the connection is made to support different chains based on the header fields of incoming connections. Perhaps support separate plugins for different type of payload encoding.

The nice thing is that it is easier to add, change and test functionality as long as each plugin only has a single responsibility.

This all looks like quite a bit of work however, so maybe a shortcut could be to support sending messages using subscription.Client.Send().

[hspaay]

Thank you I'll give it a try.