require 'set'
require 'json'
class Broadcaster
def initialize
@mutex = Mutex.new
@connections = Set.new
end
def subscribe(stream)
@mutex.synchronize { @connections << stream }
lambda { @mutex.synchronize { @connections.delete(stream) } }
end
def publish(event:, data:, id: nil)
frame = format_sse(event: event, data: data, id: id)
dead = []
@mutex.synchronize do
@connections.each do |stream|
begin
stream << frame
rescue IOError, Errno::EPIPE
dead << stream
end
end
dead.each { |s| @connections.delete(s) }
end
end
def size
@mutex.synchronize { @connections.size }
end
private
def format_sse(event:, data:, id: nil)
lines = []
lines << "id: #{id}" if id
lines << "event: #{event}"
payload = data.is_a?(String) ? data : JSON.generate(data)
payload.each_line { |line| lines << "data: #{line.chomp}" }
lines.join("\n") + "\n\n"
end
end
require 'sinatra'
require 'json'
require_relative 'broadcaster'
set :server, :puma
BROADCAST = Broadcaster.new
get '/events' do
content_type 'text/event-stream'
headers 'X-Accel-Buffering' => 'no', 'Cache-Control' => 'no-cache'
stream(:keep_open) do |out|
out << "retry: 3000\n\n"
unsubscribe = BROADCAST.subscribe(out)
pinger = Thread.new do
loop do
sleep 15
begin
out << ": keep-alive\n\n"
rescue IOError, Errno::EPIPE
break
end
end
end
out.callback do
unsubscribe.call
pinger.kill
end
end
end
post '/publish' do
body = JSON.parse(request.body.read)
BROADCAST.publish(
event: body.fetch('event', 'message'),
data: body.fetch('data'),
id: body['id']
)
status 202
{ subscribers: BROADCAST.size }.to_json
end
const source = new EventSource('/events');
source.addEventListener('message', (e) => {
const payload = JSON.parse(e.data);
console.log('message', payload, 'lastId=', e.lastEventId);
});
source.addEventListener('notice', (e) => {
renderNotice(JSON.parse(e.data));
});
source.onopen = () => console.log('SSE connected');
source.onerror = (err) => {
// EventSource auto-reconnects using Last-Event-ID; only log for now.
console.warn('SSE dropped, browser will retry', err);
};
function renderNotice(notice) {
const el = document.createElement('div');
el.className = 'notice';
el.textContent = notice.text;
document.getElementById('feed').prepend(el);
}
Server-Sent Events (SSE) is a one-way streaming protocol where the server holds an HTTP connection open and pushes text/event-stream frames to the browser's EventSource. Unlike WebSockets it needs no protocol upgrade, rides over plain HTTP, and reconnects automatically — which makes it a good fit for dashboards, notifications, and log tailing. The challenge in a threaded Rack server is keeping that connection alive without leaking threads, and delivering messages to the right subscribers.
The Broadcaster tab is a tiny in-process pub/sub hub. It keeps a Set of open connection objects guarded by a Mutex, because Sinatra serves each request on its own thread and the subscriber set is shared mutable state. subscribe registers a connection and returns a teardown lambda; publish serializes a payload once as an SSE frame and fans it out, dropping any connection that raises on write (a client that has gone away). Formatting lives in format_sse, which emits the event:, data:, and optional id: fields followed by the blank line that terminates a frame.
The app.rb tab wires this into Sinatra using stream(:keep_open). That variant hands back a stream object and does NOT close it when the block returns, so the response body stays open for out-of-band writes. The route sets Content-Type: text/event-stream, disables buffering with X-Accel-Buffering: no (important behind nginx), and registers the stream with the Broadcaster. A background Thread sends a comment line : every fifteen seconds — a heartbeat that keeps proxies and load balancers from reaping an idle connection. The crucial detail is stream.callback: when the client disconnects, Sinatra invokes it, letting the code unsubscribe and kill the ping thread so nothing leaks. A separate POST /publish route lets any producer inject an event, and retry: on the initial frame tells the browser how long to wait before reconnecting.
The sse-client.js tab shows the consumer: new EventSource('/events') with addEventListener handlers per event name, plus an onerror hook. Because EventSource reconnects on its own and replays from Last-Event-ID, the server should treat id: as a resumable cursor. The main trade-off is that this hub is single-process; scaling across workers requires an external bus like Redis pub/sub in place of the in-memory Set.
Related snips
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
class Comment < ApplicationRecord
belongs_to :article
belongs_to :author, class_name: "User"
validates :body, presence: true, length: { maximum: 2_000 }Live comments with model broadcasts + turbo_stream_from
Share this code
Here's the card — post it anywhere.
