Server-Sent Events (SSE)

Server-sent events can be used in several ways. The first example below is per-connection, per-handler SSE. For more complex broadcasting logic, see the second example using r3labs/sse.

Caution

SSE connections are long-lived, so the server’s write timeout must be disabled. Both examples set s.WriteTimeout = 0 via BeforeServeFunc.

Using SSE

Server

The handler writes the SSE headers, then emits an event every second until the client disconnects. http.NewResponseController(w).Flush() pushes each event to the client immediately.

package main

import (
  "context"
  "fmt"
  "log"
  "net/http"
  "os"
  "os/signal"
  "syscall"
  "time"

  "github.com/labstack/echo/v5"
  "github.com/labstack/echo/v5/middleware"
)

func main() {
  e := echo.New()

  e.Use(middleware.RequestLogger())
  e.Use(middleware.Recover())
  e.File("/", "./index.html")

  e.GET("/sse", func(c *echo.Context) error {
    log.Printf("SSE client connected, ip: %v", c.RealIP())

    w := c.Response()
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")

    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    count := uint64(0)
    for {
      select {
      case <-c.Request().Context().Done():
        log.Printf("SSE client disconnected, ip: %v", c.RealIP())
        return nil
      case <-ticker.C:
        count++
        event := Event{
          Data: []byte(fmt.Sprintf("count: %d, time: %s\n\n", count, time.Now().Format(time.RFC3339Nano))),
        }
        if err := event.MarshalTo(w); err != nil {
          return err
        }
        if err := http.NewResponseController(w).Flush(); err != nil {
          return err
        }
      }
    }
  })

  sc := echo.StartConfig{
    Address: ":8080",
    BeforeServeFunc: func(s *http.Server) error {
      s.WriteTimeout = 0 // IMPORTANT: disable for SSE
      return nil
    },
  }
  ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) // start shutdown process on ctrl+c
  defer cancel()

  if err := sc.Start(ctx, e); err != nil {
    e.Logger.Error("failed to start server", "error", err)
  }
}

Event structure and Marshal method

package main

import (
  "bytes"
  "fmt"
  "io"
)

// Event represents Server-Sent Event.
// SSE explanation: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
type Event struct {
  // ID is used to set the EventSource object's last event ID value.
  ID []byte
  // Data field is for the message. When the EventSource receives multiple consecutive lines
  // that begin with data:, it concatenates them, inserting a newline character between each one.
  // Trailing newlines are removed.
  Data []byte
  // Event is a string identifying the type of event described. If this is specified, an event
  // will be dispatched on the browser to the listener for the specified event name; the website
  // source code should use addEventListener() to listen for named events. The onmessage handler
  // is called if no event name is specified for a message.
  Event []byte
  // Retry is the reconnection time. If the connection to the server is lost, the browser will
  // wait for the specified time before attempting to reconnect. This must be an integer, specifying
  // the reconnection time in milliseconds. If a non-integer value is specified, the field is ignored.
  Retry []byte
  // Comment line can be used to prevent connections from timing out; a server can send a comment
  // periodically to keep the connection alive.
  Comment []byte
}

// MarshalTo marshals Event to given Writer
func (ev *Event) MarshalTo(w io.Writer) error {
  // Marshalling part is taken from: https://github.com/r3labs/sse/blob/c6d5381ee3ca63828b321c16baa008fd6c0b4564/http.go#L16
  if len(ev.Data) == 0 && len(ev.Comment) == 0 {
    return nil
  }

  if len(ev.Data) > 0 {
    if _, err := fmt.Fprintf(w, "id: %s\n", ev.ID); err != nil {
      return err
    }

    sd := bytes.Split(ev.Data, []byte("\n"))
    for i := range sd {
      if _, err := fmt.Fprintf(w, "data: %s\n", sd[i]); err != nil {
        return err
      }
    }

    if len(ev.Event) > 0 {
      if _, err := fmt.Fprintf(w, "event: %s\n", ev.Event); err != nil {
        return err
      }
    }

    if len(ev.Retry) > 0 {
      if _, err := fmt.Fprintf(w, "retry: %s\n", ev.Retry); err != nil {
        return err
      }
    }
  }

  if len(ev.Comment) > 0 {
    if _, err := fmt.Fprintf(w, ": %s\n", ev.Comment); err != nil {
      return err
    }
  }

  if _, err := fmt.Fprint(w, "\n"); err != nil {
    return err
  }

  return nil
}

HTML serving SSE

<!DOCTYPE html>
<html>
<body>

<h1>Getting server-sent updates</h1>
<div id="result"></div>

<script>
    // Example taken from: https://www.w3schools.com/html/html5_serversentevents.asp
    if (typeof (EventSource) !== "undefined") {
        const source = new EventSource("/sse");
        source.onmessage = function (event) {
            document.getElementById("result").innerHTML += event.data + "<br>";
        };
    } else {
        document.getElementById("result").innerHTML = "Sorry, your browser does not support server-sent events...";
    }
</script>

</body>
</html>

Broadcasting with r3labs/sse

When you need to broadcast a single stream of events to many subscribers, the r3labs/sse library handles stream and subscriber management for you.

Server

package main

import (
  "context"
  "net/http"
  "os"
  "os/signal"
  "syscall"
  "time"

  "github.com/labstack/echo/v5"
  "github.com/labstack/echo/v5/middleware"
  "github.com/r3labs/sse/v2"
)

func main() {
  e := echo.New()

  server := sse.New()             // create SSE broadcaster server
  server.AutoReplay = false       // do not replay messages for each new subscriber that connects
  _ = server.CreateStream("time") // EventSource in "index.html" connecting to stream named "time"

  go func(s *sse.Server) {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()

    for {
      select {
      case <-ticker.C:
        s.Publish("time", &sse.Event{
          Data: []byte("time: " + time.Now().Format(time.RFC3339Nano)),
        })
      }
    }
  }(server)

  e.Use(middleware.RequestLogger())
  e.Use(middleware.Recover())
  e.File("/", "./index.html")

  //e.GET("/sse", echo.WrapHandler(server))

  e.GET("/sse", func(c *echo.Context) error { // longer variant with disconnect logic
    e.Logger.Info("New client connected", "ip", c.RealIP())
    go func() {
      <-c.Request().Context().Done() // Received Browser Disconnection
      e.Logger.Info("Client disconnected", "ip", c.RealIP())
    }()

    server.ServeHTTP(c.Response(), c.Request())
    return nil
  })

  sc := echo.StartConfig{
    Address: ":8080",
    BeforeServeFunc: func(s *http.Server) error {
      s.WriteTimeout = 0 // IMPORTANT: disable for SSE
      return nil
    },
  }
  ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) // start shutdown process on ctrl+c
  defer cancel()

  if err := sc.Start(ctx, e); err != nil {
    e.Logger.Error("failed to start server", "error", err)
  }
}

HTML serving SSE

<!DOCTYPE html>
<html>
<body>

<h1>Getting server-sent updates</h1>
<div id="result"></div>

<script>
    // Example taken from: https://www.w3schools.com/html/html5_serversentevents.asp
    if (typeof (EventSource) !== "undefined") {
        const source = new EventSource("/sse?stream=time");
        source.onmessage = function (event) {
            document.getElementById("result").innerHTML += event.data + "<br>";
        };
    } else {
        document.getElementById("result").innerHTML = "Sorry, your browser does not support server-sent events...";
    }
</script>

</body>
</html>