Client & server support for text/event-stream, SSE

Author: quinnj

CHANGELOG.md

@@ -5,7 +5,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-<!-- ## [Unreleased] -->
+## [Unreleased]
+### Added
+- Added full Server-Sent Events (SSE) support for both client and server:
+  - **Client-side**: `sse_callback` keyword argument for `HTTP.request` to parse SSE streams
+    directly, invoking a callback with `HTTP.SSEEvent` for each event received.
+  - **Server-side**: `HTTP.sse_stream(response)` creates an `HTTP.SSEStream` that can be written
+    to with `write(stream, HTTP.SSEEvent(...))` for easy SSE response generation.
+  - `HTTP.SSEEvent` struct for representing SSE events with `data`, `event`, `id`, and `retry` fields.
+  - `HTTP.SSEError` exception for SSE-specific errors.
 
 ## [v1.10.1] - 2023-11-28
 ### Changed

README.md

@@ -56,6 +56,15 @@ HTTP.open(:GET, "https://tinyurl.com/bach-cello-suite-1-ogg") do http
 end
 ```
 
+Handle [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events) (SSE) streams by passing an `sse_callback` function to `HTTP.request`:
+
+```julia
+events = HTTP.SSEEvent[]
+HTTP.request("GET", "http://127.0.0.1:8080/events"; sse_callback = event -> push!(events, event))
+```
+
+Each callback receives an `HTTP.SSEEvent` with the parsed `data`, `event`, `id`, `retry`, and `fields` from the stream.
+
 ## Server Examples
 
 [`HTTP.Servers.listen`](https://juliaweb.github.io/HTTP.jl/stable/index.html#HTTP.Servers.listen):

docs/examples/server_sent_events.jl

@@ -38,17 +38,57 @@ loosely following [this tutorial](https://developer.mozilla.org/en-US/docs/Web/A
 ```julia
 using HTTP, JSON
 
+# Using sse_callback for automatic SSE parsing:
+HTTP.request("GET", "http://127.0.0.1:8080/api/events"; sse_callback = (stream, event) -> begin
+    @info "Received event" data=event.data event_type=event.event id=event.id
+end)
+
+# Or using HTTP.open for raw streaming:
 HTTP.open("GET", "http://127.0.0.1:8080/api/events") do io
     while !eof(io)
         println(String(readavailable(io)))
     end
 end
 ```
 
-### Server code:
+### Server code (using HTTP.sse_stream - recommended):
 """
 using HTTP, Sockets, JSON
 
+# Simple SSE server using the HTTP.sse_stream helper
+function simple_sse_server()
+    server = HTTP.serve!(listenany=true) do request
+        response = HTTP.Response(200)
+        # Add CORS headers for browser clients
+        HTTP.setheader(response, "Access-Control-Allow-Origin" => "*")
+
+        # Create SSE stream - automatically sets Content-Type and Cache-Control
+        stream = HTTP.sse_stream(response)
+
+        # Spawn a task to write events asynchronously
+        Threads.@spawn begin
+            try
+                for i in 1:10
+                    # Write a ping event with timestamp
+                    write(stream, HTTP.SSEEvent(string(round(Int, time())); event="ping"))
+
+                    # Occasionally write a data event
+                    if rand(Bool)
+                        write(stream, HTTP.SSEEvent(string(rand())))
+                    end
+                    sleep(1)
+                end
+            finally
+                close(stream)
+            end
+        end
+
+        return response
+    end
+    return server
+end
+
+# More complex example with Router
 const ROUTER = HTTP.Router()
 
 function getItems(req::HTTP.Request)
@@ -62,17 +102,47 @@ function getItems(req::HTTP.Request)
     return HTTP.Response(200, headers, JSON.json(rand(2)))
 end
 
-function events(stream::HTTP.Stream)
+# Using HTTP.sse_stream with a request handler
+function events_handler(req::HTTP.Request)
+    if HTTP.method(req) == "OPTIONS"
+        return HTTP.Response(200, [
+            "Access-Control-Allow-Origin" => "*",
+            "Access-Control-Allow-Methods" => "GET, OPTIONS"
+        ])
+    end
+
+    response = HTTP.Response(200)
+    HTTP.setheader(response, "Access-Control-Allow-Origin" => "*")
+    stream = HTTP.sse_stream(response)
+
+    Threads.@spawn begin
+        try
+            while true
+                write(stream, HTTP.SSEEvent(string(round(Int, time())); event="ping"))
+                if rand(Bool)
+                    write(stream, HTTP.SSEEvent(string(rand())))
+                end
+                sleep(1)
+            end
+        finally
+            close(stream)
+        end
+    end
+
+    return response
+end
+
+# Alternative: manual SSE using stream handler (lower-level approach)
+function events_stream(stream::HTTP.Stream)
     HTTP.setheader(stream, "Access-Control-Allow-Origin" => "*")
     HTTP.setheader(stream, "Access-Control-Allow-Methods" => "GET, OPTIONS")
     HTTP.setheader(stream, "Content-Type" => "text/event-stream")
+    HTTP.setheader(stream, "Cache-Control" => "no-cache")
 
     if HTTP.method(stream.message) == "OPTIONS"
         return nothing
     end
 
-    HTTP.setheader(stream, "Content-Type" => "text/event-stream")
-    HTTP.setheader(stream, "Cache-Control" => "no-cache")
     while true
         write(stream, "event: ping\ndata: $(round(Int, time()))\n\n")
         if rand(Bool)
@@ -83,23 +153,30 @@ function events(stream::HTTP.Stream)
     return nothing
 end
 
-HTTP.register!(ROUTER, "GET", "/api/getItems", HTTP.streamhandler(getItems))
-HTTP.register!(ROUTER, "/api/events", events)
+HTTP.register!(ROUTER, "GET", "/api/getItems", getItems)
+HTTP.register!(ROUTER, "GET", "/api/events", events_handler)
 
-server = HTTP.serve!(ROUTER, "127.0.0.1", 8080; stream=true)
+# Start the server in the normal request-handler mode
+server = HTTP.serve!(ROUTER, "127.0.0.1", 8080)
 
-# Julia usage
-resp = HTTP.get("http://localhost:8080/api/getItems")
+# To run the manual stream-handler variant instead, start a separate server:
+# stream_server = HTTP.serve!(events_stream, "127.0.0.1", 8081; stream=true)
 
-close = Ref(false)
-@async HTTP.open("GET", "http://127.0.0.1:8080/api/events") do io
-    while !eof(io) && !close[]
-        println(String(readavailable(io)))
+# Julia client usage with sse_callback
+stop = Ref(false)
+@async begin
+    try
+        HTTP.request("GET", "http://127.0.0.1:8080/api/events"; sse_callback = (stream, event) -> begin
+            println("Event: ", event.event, " | Data: ", event.data)
+            stop[] && close(stream)
+        end)
+    catch e
+        # Connection closed or stopped
     end
 end
 
 # run the following to stop the streaming client request
-close[] = true
+stop[] = true
 
 # close the server which will stop the HTTP server from listening
 close(server)

docs/src/client.md

@@ -233,6 +233,34 @@ end -> HTTP.Response
 
 Where the `io` argument provided to the function body is an `HTTP.Stream` object, a custom `IO` that represents an open connection that is ready to be written to in order to send the request body, and/or read from to receive the response body. Note that `startread(io)` should be called before calling `readavailable` to ensure the response status line and headers are received and parsed appropriately. Calling `eof(io)` will return true until the response body has been completely received. Note that the returned `HTTP.Response` from `HTTP.open` will _not_ have a `.body` field since the body was read in the function body.
 
+### Server-Sent Events
+
+HTTP.jl can parse [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events) streams directly via the `sse_callback` keyword on `HTTP.request`. When this keyword is supplied, HTTP.jl incrementally parses the incoming bytes as an event stream and invokes the callback with an `HTTP.SSEEvent` struct for every event:
+
+```julia
+using HTTP
+
+events = HTTP.SSEEvent[]
+HTTP.request("GET", "http://127.0.0.1:8080/events"; sse_callback = (stream, event) -> begin
+    @info "event" data=event.data id=event.id event_type=event.event retry_after=event.retry
+    push!(events, event)
+end)
+```
+
+The callback can be `f(event)` or `f(stream, event)`. The two-argument form allows cancelling the request by calling `close(stream)` (for example, in response to a specific event).
+
+Each callback receives a `SSEEvent` with the following fields:
+
+- `data::String`: newline-joined `data:` payload for the event (with the trailing newline removed).
+- `event::Union{Nothing,String}`: the most recent `event:` field, or `nothing` when not provided (equivalent to the default `"message"` type).
+- `id::Union{Nothing,String}`: the last `id:` value observed, automatically persisted between events per the SSE specification.
+- `retry::Union{Nothing,Int}`: the last `retry:` directive in milliseconds, propagated to subsequent events until another `retry:` value is parsed.
+- `fields::Dict{String,String}`: newline-joined string values for every field encountered since the previous event, including custom non-standard fields.
+
+Because HTTP.jl streams the response directly to the callback, the returned `HTTP.Response` will always have `response.body === HTTP.nobody`. The `sse_callback` keyword cannot be combined with `response_stream` or a custom `iofunction`, and parsing errors raise `HTTP.SSEError`. Compressed streams are supported automatically unless `decompress=false` is explicitly set.
+
+For a full end-to-end example, see [`docs/examples/server_sent_events.jl`](https://github.com/JuliaWeb/HTTP.jl/blob/master/docs/examples/server_sent_events.jl).
+
 ### Download
 
 A [`download`](@ref) function is provided for similar functionality to `Downloads.download`.

docs/src/reference.md

@@ -89,6 +89,15 @@ HTTP.WebSockets.isclosed
 HTTP.WebSockets.isok
 ```
 
+### Server-Sent Events
+
+```@docs
+HTTP.SSEEvent
+HTTP.SSEStream
+HTTP.sse_stream
+HTTP.SSEError
+```
+
 ## Utilities
 
 ```@docs

docs/src/server.md

@@ -108,8 +108,84 @@ Lower-level core server functionality that only operates on `HTTP.Stream`. Provi
 
 Nginx-style log formatting is supported via the [`HTTP.@logfmt_str`](@ref) macro and can be passed via the `access_log` keyword argument for [`HTTP.listen`](@ref) or [`HTTP.serve`](@ref).
 
+## Server-Sent Events (SSE)
+
+HTTP.jl provides built-in support for [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events), a standard for pushing real-time updates from server to client over HTTP.
+
+### Creating an SSE Response
+
+Use [`HTTP.sse_stream`](@ref) to create an SSE stream from a response object:
+
+```julia
+using HTTP
+
+HTTP.serve() do request
+    response = HTTP.Response(200)
+    stream = HTTP.sse_stream(response)
+
+    # Spawn a task to write events asynchronously
+    Threads.@spawn begin
+        try
+            for i in 1:5
+                write(stream, HTTP.SSEEvent("Event $i"))
+                sleep(1)
+            end
+        finally
+            close(stream)
+        end
+    end
+
+    return response
+end
+```
+
+The `sse_stream` function:
+1. Creates an `SSEStream` for writing events
+2. Sets the response body to the stream
+3. Adds required headers: `Content-Type: text/event-stream` and `Cache-Control: no-cache`
+4. Uses a bounded internal buffer (configurable via `max_len`, default 16 MiB) to provide backpressure if the client is slow to read
+
+### Writing Events
+
+Write events using `write(stream, HTTP.SSEEvent(...))`:
+
+```julia
+# Simple data-only event
+write(stream, HTTP.SSEEvent("Hello, world!"))
+
+# Event with type (for client-side addEventListener)
+write(stream, HTTP.SSEEvent("User logged in"; event="login"))
+
+# Event with ID (for client reconnection tracking)
+write(stream, HTTP.SSEEvent("Message content"; id="msg-123"))
+
+# Event with retry hint (milliseconds)
+write(stream, HTTP.SSEEvent("Reconnect hint"; retry=5000))
+
+# Event with all fields
+write(stream, HTTP.SSEEvent("Full event"; event="update", id="42", retry=3000))
+
+# Multiline data is automatically handled
+write(stream, HTTP.SSEEvent("Line 1\nLine 2\nLine 3"))
+```
+
+### SSEEvent Fields
+
+The `HTTP.SSEEvent` struct supports:
+- `data::String`: The event payload (required)
+- `event::Union{Nothing,String}`: Event type name (maps to `addEventListener` on client)
+- `id::Union{Nothing,String}`: Event ID for reconnection tracking
+- `retry::Union{Nothing,Int}`: Suggested reconnection delay in milliseconds
+
+### Important Notes
+
+- Always close the stream when done: `close(stream)`
+- Use `Threads.@spawn` or `@async` to write events without blocking the handler
+- The handler must return the response immediately; events are written asynchronously
+- For client-side SSE consumption, see the [Client documentation](client.md#Server-Sent-Events)
+
 ## Serving on the interactive thead pool
 
 Beginning in Julia 1.9, the main server loop is spawned on the [interactive threadpool](https://docs.julialang.org/en/v1.9/manual/multi-threading/#man-threadpools) by default. If users do a Threads.@spawn from a handler, those threaded tasks should run elsewhere and not in the interactive threadpool, keeping the web server responsive.
 
-Note that just having a reserved interactive thread doesn’t guarantee CPU cycles, so users need to properly configure their running Julia session appropriately (i.e. ensuring non-interactive threads available to run tasks, etc).
+Note that just having a reserved interactive thread doesn’t guarantee CPU cycles, so users need to properly configure their running Julia session appropriately (i.e. ensuring non-interactive threads available to run tasks, etc).

src/HTTP.jl

@@ -44,6 +44,7 @@ include("StatusCodes.jl")              ;using .StatusCodes
 include("Messages.jl")                 ;using .Messages
 include("cookies.jl")                  ;using .Cookies
 include("Streams.jl")                  ;using .Streams
+include("SSE.jl")                      ;using .SSE
 
 getrequest(r::Request) = r
 getrequest(s::Stream) = s.message.request
@@ -131,10 +132,16 @@ shorthand for `HTTP.request("GET", ...)`, etc.
 
 Supported optional keyword arguments:
 
- - `query = nothing`, a `Pair` or `Dict` of key => values to be included in the url
- - `response_stream = nothing`, a writeable `IO` stream or any `IO`-like
+- `query = nothing`, a `Pair` or `Dict` of key => values to be included in the url
+- `response_stream = nothing`, a writeable `IO` stream or any `IO`-like
     type `T` for which `write(T, AbstractVector{UInt8})` is defined. The response body
     will be written to this stream instead of returned as a `Vector{UInt8}`.
+- `sse_callback = nothing`, provide a function `f(event)` or `f(stream, event)` to incrementally
+    consume Server-Sent Events responses. When set, HTTP.jl parses the response body as an event
+    stream, invokes the callback for each `HTTP.SSEEvent`, and returns the final `HTTP.Response`
+    with `response.body === HTTP.nobody`. The two-argument form can cancel the request by calling
+    `close(stream)`. This keyword is mutually exclusive with custom `iofunction` or `response_stream`
+    handling.
  - `verbose = 0`, set to `1` or `2` for increasingly verbose logging of the
     request and response process
  - `connect_timeout = 30`, close the connection after this many seconds if it