Server-Sent Events in Vapor

Sooner or later every server picks up a job that outlives the request that started it. A build runs. A file gets transcoded. A model streams tokens back a few at a time. And then somebody asks the obvious question: how does the client find out what's happening?

The default answer is polling. The client asks every two seconds, the server answers "not yet" ninety-nine times, and everyone agrees to pretend the latency is fine. The other answer is WebSockets, which work, but you're adding a protocol upgrade, a bidirectional channel you only use in one direction, and a reconnection story you now own.

Server-Sent Events sit in between, and they're a much better fit for this shape of problem than most people give them credit for. It's plain HTTP. One request, one long-lived response, text going one way. Browsers ship a client for it (EventSource) that reconnects on its own and tells the server where it left off. There's no handshake and no framing to think about, because the framing is data: something followed by a blank line.

Vapor can do all of this today. It just doesn't give you anything to do it with, so you end up hand-writing the wire format on top of a body stream. I got tired of writing that code, so I pulled it into a library: vapor-sse.

What it looks like without a library

Here's an SSE endpoint in plain Vapor. Nothing is hidden, which is exactly the problem:

app.get("events") { req -> Response in
    Response(
        status: .ok,
        headers: [
            "Content-Type": "text/event-stream",
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        ],
        body: .init(managedAsyncStream: { writer in
            for i in 0..<10 {
                var buffer = ByteBufferAllocator().buffer(capacity: 32)
                buffer.writeString("data: tick \(i)\n\n")
                try await writer.writeBuffer(buffer)
                try await Task.sleep(for: .seconds(1))
            }
        })
    )
}

That works. It also has four things in it you have to be right about every single time, and only one of them will tell you when you're wrong.

The headers are the obvious one. Miss text/event-stream and the browser hands you a string instead of an event stream. Miss no-cache and a proxy may decide your infinite response is worth keeping.

The wire format is the sneaky one. data: tick 0\n\n is correct, and data: tick 0\n is not, and the difference is that the second one never dispatches an event on the client. Nothing errors. Nothing logs. The connection just sits there looking healthy while your UI shows nothing. Now put a newline inside the payload, which JSON pretty-printing will happily do for you, and you've silently ended the event early.

Then there's the buffer dance, three lines every time you want to say one thing. And finally the encoding: the moment you send anything richer than a string, you're carrying a JSONEncoder around and converting Data to String at each call site.

None of it is hard. It's just the kind of code where being 95% right produces a stream that looks fine and delivers nothing.

The same endpoint with vapor-sse

import Vapor
import SSE

app.get("events") { req in
    req.sse { sse in
        for i in 0..<10 {
            try await sse.send("tick \(i)")
            try await Task.sleep(for: .seconds(1))
        }
    }
}

req.sse sets the headers, opens a chunked body stream, and hands your closure an SSEContext. You send events; the format is the library's problem. The connection stays open for exactly as long as your closure is running, which means the lifetime of the stream is just the lifetime of an async function. Return, and the stream ends cleanly. Throw, and the error propagates out and terminates the stream. There's no separate "close the channel" call to forget.

The wire format, once

The whole spec is a handful of prefixed lines terminated by a blank one, and ServerSentEvent is a direct model of it:

public struct ServerSentEvent: Sendable {
    public var id: String?
    public var event: String?
    public var data: String?
    public var retry: Int?
    public var comment: String?
}

Those five fields are all there is. data is your payload. event names the event so a client can listen for that name specifically. id is what the client echoes back when it reconnects. retry tells the client how long to wait before reconnecting, in milliseconds. comment is a line the client ignores entirely, which turns out to be more useful than it sounds.

Serializing handles the part that bites you by hand:

if let data {
    for line in data.split(separator: "\n", omittingEmptySubsequences: false) {
        result += "data: \(line)\n"
    }
}

Every line of a multi-line payload gets its own data: prefix, which is how the spec says to do it, and the client reassembles them with newlines between. Send a pretty-printed JSON blob or a stack trace and it arrives intact instead of truncated at the first line break.

Sending JSON

Most real payloads aren't strings. Any Encodable value can go straight down the wire:

struct Progress: Codable, Sendable {
    var step: String
    var percent: Int
}

app.get("build", ":id", "progress") { req in
    req.sse { sse in
        try await sse.send(Progress(step: "resolving", percent: 10))
        try await sse.send(Progress(step: "compiling", percent: 60))
        try await sse.send(Progress(step: "done", percent: 100))
    }
}

The default encoder uses .sortedKeys and .withoutEscapingSlashes, so output is compact, single-line, and deterministic. Deterministic matters more than you'd expect: it's what lets you assert on exact stream bytes in a test.

If your model has dates, or you want your own key strategy, pass an encoder in:

let encoder = JSONEncoder()
encoder.dateEncodingStrategy = .iso8601
encoder.keyEncodingStrategy = .convertToSnakeCase

app.get("events") { req in
    req.sse(encoder: encoder) { sse in
        try await sse.send(Progress(step: "resolving", percent: 10))
    }
}

Named events

An SSE stream is one connection, but it doesn't have to be one kind of message. Naming events lets the client subscribe to each kind separately:

app.get("build", ":id", "progress") { req in
    req.sse { sse in
        for await update in buildUpdates(for: req.parameters.get("id")!) {
            switch update {
            case .progress(let p):
                try await sse.send(event: "progress", data: p)
            case .log(let line):
                try await sse.send(event: "log", data: line)
            case .finished(let result):
                try await sse.send(event: "finished", data: result)
                return
            }
        }
    }
}

Note the return on the last case. Ending the function ends the stream, and the client sees a normal end of response rather than a dropped connection.

On the browser side those names map one-to-one onto listeners:

const source = new EventSource("/build/42/progress");

source.addEventListener("progress", (e) => {
  const { step, percent } = JSON.parse(e.data);
  bar.style.width = `${percent}%`;
  label.textContent = step;
});

source.addEventListener("log", (e) => appendLine(e.data));
source.addEventListener("finished", () => source.close());

That last line is worth calling out. EventSource reconnects automatically when a connection ends, forever, unless you close it. Terminating the stream server-side is a signal to the client, not an instruction. If "the work is done" is a real state in your system, send an event that says so and let the client close the connection.

Keeping the connection alive

Idle connections get killed. Load balancers, reverse proxies, and mobile networks all have their own opinion about how long a silent socket gets to live, and the numbers are usually somewhere between thirty and ninety seconds.

This is what comments are for. They're a line the client discards, which makes them free traffic that keeps the socket warm:

app.get("queue", "watch") { req in
    req.sse { sse in
        var lastCount = -1
        while true {
            let count = try await Job.query(on: req.db).filter(\.$state == .pending).count()
            if count != lastCount {
                try await sse.send(event: "depth", data: ["pending": count])
                lastCount = count
            } else {
                try await sse.sendComment("keep-alive")
            }
            try await Task.sleep(for: .seconds(15))
        }
    }
}

SSEContext is Sendable, so you can also hand it to a child task and run a heartbeat concurrently with your real work. That's safe to do. Each event is serialized and handed to Vapor as a single buffer, and Vapor's body stream writer dispatches each of those onto the connection's event loop, so events are written whole. Two tasks sending at once can't splice one event into the middle of another.

What concurrency doesn't buy you is ordering. Two tasks racing to send have no defined order between them, so if you're stamping ids for resume, keep those on one task. Sequential awaits from a single task are ordered, because each send completes before the next one starts.

That loop also answers "what happens when the client goes away?" You don't get a callback. What you get is a write that eventually fails once the connection is gone, and since that's an ordinary thrown error inside an ordinary async function, it propagates out and unwinds your loop the way you'd want. The heartbeat doubles as your disconnect detector.

Reconnection that actually resumes

This is the part of SSE that WebSockets don't give you for free. When EventSource reconnects, it sends the last id it received in a Last-Event-ID header. If you've been putting IDs on your events, you can pick up exactly where the client dropped off:

app.get("build", ":id", "progress") { req in
    let lastID = req.lastEventID.flatMap(Int.init) ?? 0

    return req.sse { sse in
        // Replay anything the client missed.
        for entry in await log.entries(after: lastID) {
            try await sse.send(ServerSentEvent(
                id: String(entry.sequence),
                event: "log",
                data: entry.text
            ))
        }

        // Then continue live, still stamping IDs.
        for await entry in await log.stream() {
            try await sse.send(ServerSentEvent(
                id: String(entry.sequence),
                event: "log",
                data: entry.text
            ))
        }
    }
}

Two details make that work. req.lastEventID is read outside the closure, on the request, because that's where the header lives. And every event carries an id, because the resume point is only as good as the last ID you sent.

ServerSentEvent is also where retry lives, if you want a say in reconnect timing:

try await sse.send(ServerSentEvent(retry: 10_000, comment: "backing off"))

Useful when you're shedding load and would rather clients came back in ten seconds than in the three the browser defaults to.

Testing it

Because the format is deterministic, tests are exact string comparisons. Here's one from the library's own suite:

@Test func sseNamedEventWithJSON() async throws {
    struct Update: Codable, Sendable {
        var status: String
    }

    try await withApp { app in
        app.get("events") { req in
            req.sse { sse in
                try await sse.send(event: "status", data: Update(status: "ok"))
            }
        }

        try await app.testing().test(.GET, "events") { res async in
            #expect(res.body.string == "event: status\ndata: {\"status\":\"ok\"}\n\n")
        }
    }
}

No parsing, no fuzzy matching, no waiting on a real client. app.testing() collects the full body, so a bounded stream is testable the same way any other route is. This is the payoff for .sortedKeys in the default encoder: the bytes are the same on every run and every machine.

For a route that streams forever, give it a termination condition in tests, either a bounded loop or a cancellation you trigger. Asserting on a stream that never ends works about as well as you'd expect.

Two things that will bite you in production

Proxy buffering. nginx buffers proxied responses by default, which for a normal response is a good idea and for SSE means your events arrive in a clump when the buffer fills or the connection closes. It looks exactly like your server never sent anything. Response is a class and its headers are mutable, so you can opt out per route:

app.get("events") { req -> Response in
    let response = req.sse { sse in
        try await sse.send("hello")
    }
    response.headers.add(name: "X-Accel-Buffering", value: "no")
    return response
}

Check what's actually in front of your app. Cloudflare, ALB, and nginx all have opinions here, and they're not the same opinions.

Connection limits. Every open stream is a connection held for its entire lifetime. Six of them per origin in older browsers over HTTP/1.1, and on the server side, a few thousand idle streams is a real resource conversation rather than a rounding error. HTTP/2 fixes the browser-side limit through multiplexing, which is a good reason to make sure it's on.

Consuming it from Swift

Browsers get EventSource. On Apple platforms, URLSession.bytes(for:) gets you most of the way there in about fifteen lines:

var request = URLRequest(url: url)
request.setValue("text/event-stream", forHTTPHeaderField: "Accept")

let (bytes, _) = try await URLSession.shared.bytes(for: request)

var data = ""
for try await line in bytes.lines {
    if line.isEmpty {
        // Blank line: the event is complete.
        if !data.isEmpty { handle(data) }
        data = ""
    } else if line.hasPrefix("data:") {
        let value = line.dropFirst(5).drop(while: { $0 == " " })
        data += data.isEmpty ? value : "\n" + value
    }
}

That's the minimum viable client, handling only data. A full one tracks event and id too and reconnects with Last-Event-ID. The reason it's this short is the same reason SSE is worth reaching for: it's a text format over a normal HTTP response, so you can parse it with the tools you already have.

Try it

.package(url: "https://github.com/kylebrowning/vapor-sse.git", from: "1.0.0")

It needs Vapor 4.121.3 or newer and Swift 6, and Vapor is its only dependency. The whole library is four files and around two hundred lines. There's nothing clever in it, which is the point: the SSE spec is small, and the code that implements it should be small enough to read in one sitting.

Version 1.0.0 is tagged for Vapor 4, and a port to Vapor 5 is in progress for a 2.0 release. If you have a case that doesn't fit, that's the most useful thing you could send me right now. It's MIT licensed, and issues and pull requests are welcome at github.com/kylebrowning/vapor-sse.

Next time you're about to reach for a WebSocket to push progress at a browser, check whether anything is actually flowing back the other way. If the answer is no, this is a smaller tool that already does the job.