Skip to content

Instantly share code, notes, and snippets.

@ismasan ismasan/sse.go
Last active Aug 21, 2019

Embed
What would you like to do?
Example SSE server in Golang
// 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()
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))
}
@danesparza

This comment has been minimized.

Copy link

commented Mar 22, 2016

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

@DrGo

This comment has been minimized.

Copy link

commented Apr 12, 2016

Very nice. Thanks

@joonazan

This comment has been minimized.

Copy link

commented Aug 9, 2016

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.

@schmohlio

This comment has been minimized.

Copy link

commented Nov 15, 2016

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

@juusechec

This comment has been minimized.

Copy link

commented Dec 23, 2016

@ismasan

This comment has been minimized.

Copy link
Owner Author

commented Apr 29, 2017

@waylandc

This comment has been minimized.

Copy link

commented Jan 16, 2019

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

@colminator

This comment has been minimized.

Copy link

commented Jan 17, 2019

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()

@rikonor

This comment has been minimized.

Copy link

commented Jan 19, 2019

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

@MiraiTunga

This comment has been minimized.

Copy link

commented Mar 17, 2019

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

@sysscale

This comment has been minimized.

Copy link

commented Mar 29, 2019

Thank you very much!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.