Server Sent Events | Alchemists

Server Sent Events

Server Sent Events (SSE) allows a server to push data, one way, to a client over a long lived HTTP connection without the client needing to make multiple requests. Even better, the client will automatically reconnect if the connection drops. Here’s a few use cases that SSE is great for:

Real-time notifications.
Live dashboards.
Live edits and rendering of updated content.
Progress updates of long running tasks.

Implementing SSE is fairly trivial in Ruby especially when using Rack 3.0.0 middleware and htmx on the front end so let’s explore further by diving into the code!

Table of Contents

Quick Start
Ruby
- Gems
- Dependencies
- Event Source
- Middleware
- Rack
HTML (htmx)
curl
Falcon
Running
Specs
Conclusion

Quick Start

If you’d prefer to have all of the code and start playing around, you only need these two files:

Ruby

Save as demo and run chmod 755 demo to make it executable.

#! /usr/bin/env ruby
# frozen_string_literal: true

# Save as `demo`, then `chmod 755 demo`, and run as `./demo`.

require "bundler/inline"

gemfile true do
  source "https://rubygems.org"

  gem "amazing_print"
  gem "cogger"
  gem "containable"
  gem "debug"
  gem "infusible"
  gem "initable"
  gem "puma"
  gem "rack"
  gem "rackup"
end

module Container
  extend Containable

  register(:logger) { Cogger.new id: :demo }
end

Dependencies = Infusible[Container]

class EventSource
  include Dependencies[:logger]
  include Initable[%i[req id], kernel: Kernel]

  def call stream
    write stream
  rescue Errno::EPIPE, Errno::ECONNRESET, IOError
    logger.debug { "Event stream disconnected." }
  ensure
    stream.close
  end

  private

  def write stream, count: 1
    kernel.loop do
      stream.write <<~CONTENT
        event: demo
        data: Found: #{id}, Count: #{count}

      CONTENT

      count += 1

      sleep 0.5
    end
  end
end

class Middleware
  include Initable[
    %i[req application],
    %i[keyreq pattern],
    headers: {
      "Cache-Control" => "no-cache",
      "Content-Type" => "text/event-stream",
      "X-Accel-Buffering" => "no"
    },
    event_source: EventSource
  ]

  def call environment
    request = Rack::Request.new environment
    path = request.path

    case path.match pattern
      in id: then [200, headers, event_source.new(id)]
      else application.call environment
    end
  end
end

application = Rack::Builder.new do
  use Middleware, pattern: %r(^/sse/(?<id>.+)$)
  use Rack::Static, urls: ["/"], root: ".", index: "index.html"

  run proc { [200, {"content-type" => "text/plain"}, ["Demonstration"]] }
end

Rackup::Server.start app: application

HTML

Save as index.html.

<!DOCTYPE html>

<html lang="en">
  <head>
    <meta name="viewport" content="width=device-width,initial-scale=1,shrink-to-fit=no">

    <title>Demo</title>

    <meta charset="utf-8">
    <meta name="description" content="A demo site.">
    <meta name="author" content="Alchemists">

    <style type="text/css">
    </style>

    <script src="https://unpkg.com/htmx.org@2.0.10"
            integrity="sha384-H5SrcfygHmAuTDZphMHqBJLc3FhssKjG7w/CeCpFReSfwBWDTKpkzPP8c+cLsK+V"
            crossorigin="anonymous">
    </script>

    <script src="https://unpkg.com/htmx-ext-sse@2.2.4"
            integrity="sha384-A986SAtodyH8eg8x8irJnYUk7i9inVQqYigD6qZ9evobksGNIXfeFvDwLSHcp31N"
            crossorigin="anonymous">
    </script>
  </head>

  <body>
    <p hx-ext="sse" sse-connect="/sse/1" sse-swap="demo">Loading...</p>
  </body>
</html>

Once you’ve saved the above to disk, you should have the following files:

index.html
demo

To launch the application, run: ./demo. The open http://localhost:9292 in your default web browser. The rest of this article will explain how all of this works.

Ruby

With the above, we are using a Bundler Inline script for demonstration purposes. Normally, you’d want to use Rubysmith to split the implementation into separate files but keeping all of this together, for now, allows you to quickly experiment with the code.

Gems

At the top of Bundler Inline script, several gems are installed for you. The following breaks down why each is used:

Ruby
- Amazing Print: For pretty printing of inspected objects.
- Cogger: For logging.
- Containable: For dependency injection containers.
- Debug: For debugging code.
- Infusible: For automatic dependency injection.
- Initable: For automatic initialization.
- Puma: For serving web content.
- Rack: For building modular web applications.
- Rackup: For launching rack middleware.
htmx: For sending HTML fragments over the wire.
- htmx SSE Extension: For processing and rendering SSE.

Dependencies

Next, we register and prepare our dependencies for automatic injection:

module Container
  extend Containable

  register(:logger) { Cogger.new id: :demo }
end

Dependencies = Infusible[Container]

This allows us to leverage dependency injection containers for dependency injection via Containable and Infusible respectively so we can register and inject Cogger for logging purposes.

Event Source

Next, we have our event source:

class EventSource
  include Dependencies[:logger]
  include Initable[%i[req id], kernel: Kernel]

  def call stream
    write stream
  rescue Errno::EPIPE, Errno::ECONNRESET, IOError
    logger.debug { "Event stream disconnected." }
  ensure
    stream.close
  end

  private

  def write stream, count: 1
    kernel.loop do
      stream.write <<~CONTENT
        event: demo
        data: Found: #{id}, Count: #{count}

      CONTENT

      count += 1

      sleep 0.5
    end
  end
end

The logger and kernel are injected as dependencies while the id is required for loading of a record associated with the ID. For demonstration purposes, we only print the ID to avoid setting up a persistence layer since you can do that on your own if you like.

In order to adhere to Rack 3.0.0 specification for event streaming, we only need make our event source callable by implementing the #call method which does the following:

Writes to the given stream.
Gracefully handles errors by logging them.
Ensures the stream is always closed.

You’ll notice the private #write method loops indefinitely on a 0.5 second sleep schedule. In order to make the output interesting, a counter is incremented but, again, you’d normally be loading or processing data to write to the stream instead of updating a simple counter.

When writing to the stream, you always need the following keys:

event: The event name (or identifier). In our case, demo is our name. We’ll look for this even in our htmx client shortly.
data: The actual data we want to return.

Both the event and data data keys must be on separate lines followed by two new lines. Example (with new lines included for demonstration purposes:

event: demo\n
data: Found 1, Count: 5\n
\n

It’s critical that you always adhere to this format or your events will be corrupted. It’s important to note that you can use multiple data keys if necessary. Example:

event: demo
data: Found 1
data: ,
data: Count: 5

The above will be automatically concatenated for you and is identical to the following:

event: demo
data: Found 1, Count: 5

If you need to keep writing to the stream in order to keep the stream alive, so-to-speak, you can drop the data key to use a colon:

event: demo
: A comment.

Some call this a pulse or a heartbeat to prevent the stream from disconnecting especially when your writes to the stream are infrequent. All comments are ignored by your client (basically a no operation). Again, only useful if your updates are infrequent in order to keep the connection open.

Middleware

In order to be compatible with Rack, our middleware must be callable:

class Middleware
  include Initable[
    %i[req application],
    %i[keyreq pattern],
    headers: {
      "Cache-Control" => "no-cache",
      "Content-Type" => "text/event-stream",
      "X-Accel-Buffering" => "no"
    },
    event_source: EventSource
  ]

  def call environment
    request = Rack::Request.new environment
    path = request.path

    case path.match pattern
      in id: then [200, headers, event_source.new(id)]
      else application.call environment
    end
  end
end

We also need to require application as the first positional parameter. Everything else is optional. Each, optional, keyword parameter is:

pattern: This allows us to match the request path based on regular expression pattern. This also makes this middleware reusable for different routes that need shared behavior.
headers: The first two headers are required for being in compliance with Server Sent Events. The X-Accel-Buffering is optional but aids in being compatible with Nginx load balancers.
event_source: This is our EventSource object, as discussed earlier, which aids in swapping in different implementations if desired.

It’s worth pointing out, depending on your situation, that you might want to modify the case statement as follows:

case path.match pattern
  in id:
    environment["puma.mark_as_io_bound"].then { it.call if it }
    environment["rack.session.options"].then { it[:skip] = true if it }
    [200, headers, event_source.new(id)]
  else application.call environment
end

The first line allows you to make use of the Puma feature that allows you to mark your Server Sent Event as I/0 bound so it can exceed the normal thread pool maximum for I/O operations that are usually slower. This pairs well with Puma’s max_io_threads configuration setting.

The second line allows you to skip sending your Rack session/authentication/cookie information with each event since you only need to send this once upon initial connection. Otherwise, you’ll see duplicate sessions pile up with each response which is definitely not what you want when streaming events.

Rack

The last part of our script uses Rack::Builder to add our middleware, serve static assets (especially our index page), and run our demonstration application.

application = Rack::Builder.new do
  use Middleware, pattern: %r(^/sse/(?<id>.+)$)
  use Rack::Static, urls: ["/"], root: ".", index: "index.html"

  run proc { [200, {"content-type" => "text/plain"}, ["Demonstration"]] }
end

Rackup::Server.start app: application

Again — and as mentioned earlier — you’d split all objects out to separate files and use a config.ru file to launch your Rack app but keeping all this logic in one file makes this easier to demonstrate.

That’s it for our Ruby code. Now let’s look at our htmx client.

HTML (htmx)

htmx is great for having a simple client for consuming our Server Sent Events:

<!DOCTYPE html>

<html lang="en">
  <head>
    <meta name="viewport" content="width=device-width,initial-scale=1,shrink-to-fit=no">

    <title>Demo</title>

    <meta charset="utf-8">
    <meta name="description" content="A demo site.">
    <meta name="author" content="Alchemists">

    <style type="text/css">
    </style>

    <script src="https://unpkg.com/htmx.org@2.0.10"
            integrity="sha384-H5SrcfygHmAuTDZphMHqBJLc3FhssKjG7w/CeCpFReSfwBWDTKpkzPP8c+cLsK+V"
            crossorigin="anonymous">
    </script>

    <script src="https://unpkg.com/htmx-ext-sse@2.2.4"
            integrity="sha384-A986SAtodyH8eg8x8irJnYUk7i9inVQqYigD6qZ9evobksGNIXfeFvDwLSHcp31N"
            crossorigin="anonymous">
    </script>
  </head>

  <body>
    <p hx-ext="sse" sse-connect="/sse/1" sse-swap="demo">Loading...</p>
  </body>
</html>

With the above, we require the htmx library and associated htmx SSE Extension extension for consuming and rendering events. You’ll notice that our path (/sse/1) maps to the same path used to configure our middleware and the event name, demo, also maps to the name used in our EventSource implementation. This is what connects the client and server together.

curl

You’re not limited to viewing the output of your SSE via the web browser as you can use curl as follows:

curl -v http://localhost:9292/sse/1

The above should yield output similar to the following:

* Host localhost:9292 was resolved.
* IPv6: ::1
* IPv4: 127.0.0.1
*   Trying [::1]:9292...
* Connected to localhost (::1) port 9292
> GET /sse/1 HTTP/1.1
> Host: localhost:9292
> User-Agent: curl/8.7.1
> Accept: */*
>
* Request completely sent off
< HTTP/1.1 200 OK
< cache-control: no-cache
< content-type: text/event-stream
< x-accel-buffering: no
<
* no chunk, no close, no size. Assume close to signal end
event: demo
data: Found: 1, Count: 1

event: demo
data: Found: 1, Count: 2

event: demo
data: Found: 1, Count: 3

Falcon

If you’d perfer Falcon, instead of Puma, you can definitely do this by requiring the gem and using falcon as the server by changing the very last line of the Bundler Inline script as follows:

Rackup::Server.start app: application, server: :falcon

Running

As mentioned in the Quick Start section, to run the demonstration, you only need to run as follows:

./demo
open http://localhost:9292

💡 You can then launch the same URL in multiple browser tabs to get multiple streams running at once.

Specs

If using Puma, ensure you use two minimum and two maximum threads since you’ll need one thread for normal page requests and one for SSE connections since they are persisted and longer lived. Otherwise, you’ll have random spec failures with fewer threads. Here’s what you should have in your spec helper for your feature specs to get started:

Capybara.server = :puma, {Silent: true, Threads: "2:2"}

The threads value of 2:2 means a minimum of two threads (first number before the colon) and a maximum of two threads (second value after the colon). You can definitely tweak these numbers by making them smaller or larger but do recommend keeping the maximum number of threads two or higher.

Conclusion

As you can see, implementing SSE takes minimal effort. You only need an event stream, middleware, and a client to wire everything up. Even better, you’ve learned how to make all of your objects reusable and configurable should you need to introduce different behavior without complicating or reducing the maintenance of your architecture.

Enjoy and may your applications be lively!