go-sse

Lightweight, fully spec-compliant HTML5 server-sent events library.

Table of contents

• go-sse
  • Table of contents
  • Installation and usage
  • Implementing a server
    • Providers and why they are vital
    • Meet Joe, the default provider
    • Publish your first event
    • The server-side "Hello world"
  • Using the client
    • Creating a client
    • Initiating a connection
    • Subscribing to events
    • Establishing the connection
    • Connection lost?
    • The "Hello world" server's client
  • License
  • Contributing

Installation and usage

Install the package using go get:

go get -u github.com/tmaxmax/go-sse

It is strongly recommended to use tagged versions of go-sse in your projects. The master branch has tested but unreleased and maybe undocumented changes, which may break backwards compatibility - use with caution.

The library provides both server-side and client-side implementations of the protocol. The implementations are completely decoupled and unopinionated: you can connect to a server created using go-sse from the browser and you can connect to any server that emits events using the client!

If you are not familiar with the protocol or not sure how it works, read MDN's guide for using server-sent events. The spec is also useful read!

Implementing a server

Providers and why they are vital

First, a server instance has to be created:

import "github.com/tmaxmax/go-sse"

s := sse.NewServer()

The sse.Server type also implements the http.Handler interface, but a server is framework-agnostic: See the ServeHTTP implementation to learn how to implement your own custom logic.

The NewServer constructor actually takes in additional options:

package sse

func NewServer(options ...ServerOption) *Server

One of them is the WithProvider option:

func WithProvider(provider Provider) Option

A provider is an implementation of the publish-subscribe messaging system:

type Provider interface {
    // Publish a message to all subscribers. A message contains the event and some additional information - read further and see the documentation.
    Publish(msg *Message) error
    // Add a new subscriber that is unsubscribed when the context is done.
    Subscribe(ctx context.Context, sub Subscription) error
    // Cleanup all resources and stop publishing messages or accepting subscriptions.
    Stop() error
}

The provider is what dispatches events to clients. When you publish a message (an event), the provider distributes it to all connections (subscribers). It is the central piece of the server: it determines the maximum number of clients your server can handle, the latency between broadcasting events and receiving them client-side and the maximum message throughput supported by your server. As different use cases have different needs, go-sse allows to plug in your own system. Some examples of such external systems are:

• RabbitMQ streams
• Redis pub-sub
• Apache Kafka
• Your own! For example, you can mock providers in testing.

If an external system is required, an adapter that satisfies the Provider interface must be created so it can then be used with go-sse. To implement such an adapter, read the Provider documentation for implementation requirements! And maybe share them with others: go-sse is built with reusability in mind!

But in most cases the power and scalability that these external systems bring is not necessary, so go-sse comes with a default provider builtin. Read further!

Meet Joe, the default provider

The server still works by default, without a provider. go-sse brings you Joe: the trusty, pure Go pub-sub implementation, who handles all your events by default! Befriend Joe as following:

import "github.com/tmaxmax/go-sse"

joe := sse.NewJoe()

and he'll dispatch events all day! By default, he has no memory of what events he has received, but you can help him remember and replay older messages to new clients using a ReplayProvider:

type ReplayProvider interface {
    // Put a new event in the provider's buffer.
    // It takes a pointer to a pointer because a new event may be created,
    // in case the ReplayProvider automatically sets IDs.
    Put(msg **Message)
    // Replay valid events to a subscriber.
    Replay(sub Subscription)
}

go-sse provides two replay providers by default, which both hold the events in-memory: the ValidReplayProvider and FiniteReplayProvider. The first replays events that are valid, not expired, the second replays a finite number of the most recent events. For example:

sse.NewJoe(sse.JoeConfig{
    ReplayProvider: server.NewValidReplayProvider(),
    ReplayGCInterval: time.Minute,
})

will tell Joe to replay all valid events and clean up the expired ones each minute! Replay providers can do so much more (for example, add IDs to events automatically): read the docs on how to use the existing ones and how to implement yours.

You can also implement your own replay providers: maybe you need persistent storage for your events? Or event validity is determined based on other criterias than expiry time? And if you think your replay provider may be useful to others, you are encouraged to share it!

go-sse created the ReplayProvider interface mainly for Joe, but it encourages you to integrate it with your own Provider implementations, where suitable.

Publish your first event

To publish events from the server, we use the sse.Message struct:

import "github.com/tmaxmax/go-sse"

m := sse.Message{}
m.AppendText("Hello world!", "Nice\nto see you.")

Now let's send it to our clients:

var s *sse.Server

s.Publish(&m)

This is how clients will receive our event:

data: Hello world!
data: Nice
data: to see you.

If we use a replay provider, such as ValidReplayProvider, this event will expire immediately and it also doesn't have an ID. Let's solve this:

m.SetID(sse.MustEventID("unique"))
m.SetTTL(5 * time.Minute)

Now the event will look like this:

id: unique
data: Hello world!
data: Nice
data: to see you.

And the ValidReplayProvider will stop replaying it after 5 minutes!

An EventID type is also exposed, which is a special type that denotes an event's ID. An ID must not have newlines, so we use a special function that validates the ID beforehand. MustEventID panics, but there's also NewEventID, which returns an error indicating whether the value was successfully converted to an ID or not:

id, err := sse.NewEventID("invalid\nID")

Here, err will be non-nil and id will be an invalid value: nothing will be sent to clients if you set an event's ID using that value!

Either way, IDs and expiry times can also be retrieved, so replay providers can use them to determine from which IDs to replay messages and which messages are still valid:

fmt.Println(m.ID(), m.ExpiresAt())

Setting the event's name (or type) is equally easy:

ok := m.SetName("The event's name")

Names cannot have newlines, so the returned boolean flag indicates whether the name was valid and set. Read the docs to find out more about messages and how to use them!

The server-side "Hello world"

Now, let's put everything that we've learned together! We'll create a server that sends a "Hello world!" message every second to all its clients, with Joe's help:

package main

import (
    "log"
    "net/http"
    "time"

    "github.com/tmaxmax/go-sse"
)

func main() {
    s := sse.NewServer()

    go func() {
        m := sse.Message{}
        m.AppendText("Hello world")

        for range time.Tick(time.Second) {
            _ = s.Publish(&m)
        }
    }()

    if err := http.ListenAndServe(":8000", s); err != nil {
        log.Fatalln(err)
    }
}

Joe is our default provider here, as no provider is given to the server constructor. The server is already an http.Handler so we can use it directly with http.ListenAndServe.

Also see a more complex example!

This is by far a complete presentation, make sure to read the docs in order to use go-sse to its full potential!

Using the client

Creating a client

We will use the sse.Client type for connecting to event streams:

type Client struct {
    HTTPClient              *http.Client
    OnRetry                 backoff.Notify
    ResponseValidator       ResponseValidator
    MaxRetries              int
    DefaultReconnectionTime time.Duration
}

As you can see, it uses a net/http client. It also uses the cenkalti/backoff library for implementing auto-reconnect when a connection to a server is lost. Read the client docs and the Backoff library's docs to find out how to configure the client. We'll use the default client the package provides for further examples.

Initiating a connection

We must first create an http.Request - yup, a fully customizable request:

req, err := http.NewRequestWithContext(ctx, http.MethodGet, "host", nil)

Any kind of request is valid as long as your server handler supports it: you can do a GET, a POST, send a body; do whatever! The context is used as always for cancellation - to stop receiving events you will have to cancel the context. Let's initiate a connection with this request:

import "github.com/tmaxmax/go-sse"

conn := sse.DefaultClient.NewConnection(req)
// you can also do client.NewConnection(req)
// it is an utility function that calls the
// NewConnection method on the default client

Subscribing to events

Great! Let's imagine the event stream looks as following:

data: some unnamed event

event: I have a name
data: some data

event: Another name
data: some data

To receive the unnamed events, we subscribe to them as following:

unsubscribe := conn.SubscribeMessages(func (event sse.Event) {
    // do something with the event
})

To receive the events named "I have a name":

unsubscribe := conn.SubscribeEvent("I have a name", func (event sse.Event) {
    // do something with the event
})

If you want to subscribe to all events, regardless of their name:

unsubscribe := conn.SubscribeToAll(func (event sse.Event) {
    // do something with the event
})

All Susbcribe methods return a function that when called tells the connection to stop calling the corresponding callback.

In order to work with events, the sse.Event type has some fields and methods exposed:

type Event struct {
    LastEventID string
    Name        string
    Data        []byte
}

func (e Event) String() { return string(e.Data) }

Pretty self-explanatory, but make sure to read the docs!

Now, with this knowledge, let's subscribe to all unnamed events and, when the connection is established, print their data to os.Stdout:

out := log.New(os.Stdout, "", 0)

unsubscribe := conn.SubscribeMessages(func(event sse.Event) {
    out.Printf("Received an unnamed event: %s\n", event)
})

We use a log.Logger instance because it synchronizes prints with a mutex: for each event the callback is called in a separate goroutine, so access to shared resources must be synchronized by the client.

Establishing the connection

Great, we are subscribed now! Let's start receiving events:

err := conn.Connect()

By calling Connect, the request created above will be sent to the server, and if successful, the subscribed callbacks will be called when new events are received. Connect returns only after all callbacks have finished executing. To stop calling a certain callback, call the unsubscribe function returned when subscribing. You can also subscribe new callbacks after calling Connect from a different goroutine.

Connection lost?

Either way, after receiving so many events, something went wrong and the server is temporarily down. Oh no! As a last hope, it has sent us the following event:

retry: 60000
: that's a minute in milliseconds and this
: is a comment which is ignored by the client

Not a sweat, though! The connection will automatically be reattempted after a minute, when we'll hope the server's back up again. Canceling the request's context will cancel any reconnection attempt, too.

If the server doesn't set a retry time, the client's DefaultReconnectionTime is used.

The "Hello world" server's client

Let's use what we know to create a client for the prevoius server example:

package main

import (
    "log"
    "net/http"
    "os"

    "github.com/tmaxmax/go-sse"
)

func main() {
    r, _ := http.NewRequest(http.MethodGet, "http://localhost:8000", nil)
    conn := sse.NewConnection(r)
    out := log.New(os.Stdout, "", 0)

    conn.SubscribeMessages(func(ev sse.Event) {
        out.Printf("%s\n\n", ev)
    })

    if err := conn.Connect(); err != nil {
        out.Println(err)
    }
}

Yup, this is it! We are using the default client to receive all the unnamed events from the server. The output will look like this, when both programs are run in parallel:

Hello world!

Hello world!

Hello world!

Hello world!

...

See the complex example's client too!

License

This project is licensed under the MIT license.

Contributing

The library's in its early stages, so contributions are vital - I'm so glad you wish to improve go-sse! Maybe start by opening an issue first, to describe the intended modifications and further discuss how to integrate them. Open PRs to the master branch and wait for CI to complete. If all is clear, your changes will soon be merged! Also, make sure your changes come with an extensive set of tests and the code is formatted.

Thank you for contributing!