Example SSE server in Golang

sse.go

// Copyright (c) 2017 Ismael Celis

// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:

// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.

package main

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

// Example SSE server in Golang.
//     $ go run sse.go

type Broker struct {

	// Events are pushed to this channel by the main events-gathering routine
	Notifier chan []byte

	// New client connections
	newClients chan chan []byte

	// Closed client connections
	closingClients chan chan []byte

	// Client connections registry
	clients map[chan []byte]bool
}

func NewServer() (broker *Broker) {
	// Instantiate a broker
	broker = &Broker{
		Notifier:       make(chan []byte, 1),
		newClients:     make(chan chan []byte),
		closingClients: make(chan chan []byte),
		clients:        make(map[chan []byte]bool),
	}

	// Set it running - listening and broadcasting events
	go broker.listen()

	return
}

func (broker *Broker) ServeHTTP(rw http.ResponseWriter, req *http.Request) {

	// Make sure that the writer supports flushing.
	//
	flusher, ok := rw.(http.Flusher)

	if !ok {
		http.Error(rw, "Streaming unsupported!", http.StatusInternalServerError)
		return
	}

	rw.Header().Set("Content-Type", "text/event-stream")
	rw.Header().Set("Cache-Control", "no-cache")
	rw.Header().Set("Connection", "keep-alive")
	rw.Header().Set("Access-Control-Allow-Origin", "*")

	// Each connection registers its own message channel with the Broker's connections registry
	messageChan := make(chan []byte)

	// Signal the broker that we have a new connection
	broker.newClients <- messageChan

	// Remove this client from the map of connected clients
	// when this handler exits.
	defer func() {
		broker.closingClients <- messageChan
	}()

	// Listen to connection close and un-register messageChan
	// notify := rw.(http.CloseNotifier).CloseNotify()
	notify := req.Context().Done()

	go func() {
		<-notify
		broker.closingClients <- messageChan
	}()

	for {

		// Write to the ResponseWriter
		// Server Sent Events compatible
		fmt.Fprintf(rw, "data: %s\n\n", <-messageChan)

		// Flush the data immediatly instead of buffering it for later.
		flusher.Flush()
	}

}

func (broker *Broker) listen() {
	for {
		select {
		case s := <-broker.newClients:

			// A new client has connected.
			// Register their message channel
			broker.clients[s] = true
			log.Printf("Client added. %d registered clients", len(broker.clients))
		case s := <-broker.closingClients:

			// A client has dettached and we want to
			// stop sending them messages.
			delete(broker.clients, s)
			log.Printf("Removed client. %d registered clients", len(broker.clients))
		case event := <-broker.Notifier:

			// We got a new event from the outside!
			// Send event to all connected clients
			for clientMessageChan, _ := range broker.clients {
				clientMessageChan <- event
			}
		}
	}

}

func main() {

	broker := NewServer()

	go func() {
		for {
			time.Sleep(time.Second * 2)
			eventString := fmt.Sprintf("the time is %v", time.Now())
			log.Println("Receiving event")
			broker.Notifier <- []byte(eventString)
		}
	}()

	log.Fatal("HTTP server error: ", http.ListenAndServe("localhost:3000", broker))

}

Thanks! Any examples of what it would look like to interact with this from a web page?

Very nice. Thanks

looks like closingClients is receiving every message channel twice, because the closing is deferred and is also done on CloseNotify.

Am I reading correctly that the infinite loop is ended by a panic? That is not really obvious.

I'm interested in this browser feature. If I end up trying this, I'll post the improved code here.

@joonazan check out my fork which includes a few fixes:

example like broadcast https://gist.github.com/juusechec/de711709cfb28cacc2b744b0e25a1274

The blog post for this post is at https://robots.thoughtbot.com/writing-a-server-sent-events-server-in-go

FWIW, I think your deferred func at lines 87-89 are duplicating what you do below with the CloseNotifier

Please note, as per the docs, http.CloseNotifier is deprecated.
To test for closed connections, instead of:

<-rw.(http.CloseNotifier).CloseNotify()

use the context.Context contained within the http.Request i.e.

<-req.Context().Done()

Here's a different take on it that simplifies the broker piece a bit rikonor/gist#e53a33.

@ismasan Hi How would one go about to get this to work with gin (https://github.com/gin-gonic/gin)

Thank you very much!

Thanks for this. If I wanted to use your firehose package to provide a NewBroker(), how would the code around line 147 in main() change?

@BertHooyman I'm not sure what you mean? There's already the NewServer() factory method that could be renamed to NewBroker(), which is probably a better name.

@MiraiTunga I haven't used Gin but I'm guessing it will work with regular HTTP handler funcs (ie. ServeHTTP in this gist).

@ismasan > I'm not sure what you mean? There's already the NewServer() factory method that could be renamed to NewBroker(), which is probably a better name.

I was referring to the data type which is different in broker.Notifier <- []byte(eventString) .

Still not sure about your question, but broker.Notifier takes a byte array so you can append to it from whatever source you want, as long as you cast it to a byte array.

Even though it has a lot of pitfalls (like non-being Go idiomatic), it's useful. Thanks.

@maestre3d not a very useful comment, however.

Well, it lacks of semantic naming, poor Go's effectiveness, non-idiomatic. You could use naming convention from Apache Kafka or RabbitMQ for Broker's attributes for example, so it could be easier for anybody to understand.

First, variable naming like the ServeHTTP's function like request/response should be idiomatic. (i.e. r instead req, w instead rw).

Channels must be used carefully, and the shall not be exposed for user modification/aggregation.

Lack of mutex lock's for every broker transaction since this is highly volatile (like repository's transactions).

Another thing, leave concurrency to the user, that listen method could've done inside the ServeHTTP implementation function.

A good addition would be the distributed ID assigning to every new client to send message to specific clients. (I just don't feel the security here sending all the messages to all subscribers/consumers).

Just to finish, that double channel closing is bad by just taking a quick look, your deferred and ctx.Done() functions overlap (and you just remove the channel from the pool, yet you neither close any channel or clear the pool when the broker's closes). I would suggest you to listen to the messages at the very end of the ServeHTTP function like I said and you could listen to request's context.Done() signal/channel with a select statement along with the messages coming to the broker, obviously inside a loop with a return (exit/break from the while/for loop) whenever a ctx.Done() signal is received.

For example, in my case I just required an Event struct with an distributed ID (using sonyflake), Message and Consumer ID along with the Broker just containing a consumer pool (chan type Event almost like yours), a sync mutex and my preferred logger. Thus, I can easily publish any message to either all consumers or to an specific consumer.

I appreciate your contrib, don't get me wrong.
But I don't know why your solution almost took my CPU's usage to 99% while my custom implementation barely uses CPU or memory.
It could be non closing neither the consumer pool or the channel itself.

@maestre3d not a very useful comment, however.

@maestre3d Could you please share your implementation ?

i create my go server implementation here (with gin)
https://gist.github.com/mirzaakhena/0936851fd2812c74da74940eaab746d7

I also include java version for client
https://gist.github.com/mirzaakhena/c75caa8529e648b7fc7b4becd4ca6fe6

@maestre3d Could you please share your implementation ?

@sharkyze I've got a custom implementation but for one of my platform's core, so if something is missing is because it is available with our core.

Here's the implementation, needs more polishing but works great and doesn't consumes your whole CPU and memory. Also, I used sonyflake to generate distributed ID's, but I would suggest you to use UUID v1-v4 instead since sonyflake (Twitter Snowflake) requires external infrastructure dependencies like Apache Zookeeper to work as expected, without it you could have high-cardinality field clashes like event ID.

Here is the complete core so you could know what's actually happening.

PD: Before you use it, I would recommend you to implement required logging, metrics and distributed tracing (observability) using the chain of responsibility pattern.

@maestre3d Could you please share your implementation ?

@sharkyze I've got a custom implementation but for one of my platform's core, so if something is missing is because it is available with our core.

Here's the implementation, needs more polishing but works great and doesn't consumes your whole CPU and memory. Also, I used sonyflake to generate distributed ID's, but I would suggest you to use UUID v1-v4 instead since sonyflake (Twitter Snowflake) requires external infrastructure dependencies like Apache Zookeeper to work as expected, without it you could have high-cardinality field clashes like event ID.

Here is the complete core so you could know what's actually happening.

PD: Before you use it, I would recommend you to implement required logging, metrics and distributed tracing (observability) using the chain of responsibility pattern.

PD 2: Mine's not using any external HTTP packages like Gin, it's using pure go's net/http package, the only external package I'm actually using is sonyflake but that's for traceability purposes and the possibility to send messages to an specific client, but don't worry, it is still pluggable.

Additionally, in my profile you can find an SSE client written in Javascript with React too just in case you wanted an straight-forward JS implementation.

@maestre3d Amazing references, really helpful. Thank you for sharing.

Very minimal setup for SSE

https://dev.to/mirzaakhena/server-sent-events-sse-server-implementation-with-go-4ck2

@ismasan thank you for useful example.
Based on your work and fork of @schmohlio , I've created a demo for Gin framework to show how to get messages by subscription on specific topic. See at https://github.com/romanm-perun/gin-sse-example

Suggestions are welcome!

Thank you for this repo. Because it is easy to understand and learn a lot of things. After reading this I know more about golang's strength
Can I ask

defer func() {
	broker.closingClients <- messageChan
}()

When is this function executed? Because I don't know ServeHTTP finished? or It always runs?

Thank you @phanthaiduong22

This is Go's defer statement.