std/sse
import std/sse to use it. Source: src/std/sse.iyi at v0.18.0.
**The server.** Inside Server.serve's block, SSE.stream { |events| ... } answers the request: a 200 with Content-Type: text/event-stream and Cache-Control: no-cache (a caller's own Cache-Control replaces it; its own Content-Type is refused), whose body the block writes as it goes through Response.stream. events.send(data, event: nil, id: nil, retry: nil) writes one event - its event:, id: and retry: lines, then its data a data: line per line, then the blank line that dispatches it - as one chunk, sent at once; events.comment(text) writes : lines, which a client reads past, and which are what keeps a quiet stream's proxies from dropping it (§9.2.7). A line break in data is three things on the way in (CRLF, CR, LF) and arrives as LF, since the format has no way to say which it was. What would break the stream is refused before it is written: a line break in an event type, a line break or a NUL in an id (a client ignores an id with a NUL, §9.2.6), a negative retry, and text that is not UTF-8, which is the only encoding the format has (§9.2.5). A client that went is not a panic: the write that finds it answers the SocketError, every write after it answers the same, and gone? says so, so the block's loop ends on a value. A client is only found gone when something is written to it; a comment is the cheapest thing that can be.
**The client.** SSE.open(url) { |event| ... } is EventSource without a browser: a GET through HTTP.open with Accept: text/event-stream and Cache-Control: no-cache (the request's no-store cache mode, as Fetch writes it), and Last-Event-ID when the last event ID is not empty. The block gets each Event - its type (message when the stream named none), its data, and the last event ID - and answers true to go on or false to stop, which closes the connection and answers nil. An answer that is not a 200, or whose Content-Type is not text/event-stream (parameters allowed), fails the connection as the standard has it: no reconnection, and an SSEError naming what came. A connection that drops or ends is reconnected after the reconnection time - retry milliseconds, 3000 unless the caller says, until the stream's own retry: field sets it - with the last event ID in Last-Event-ID, as a browser does. A browser goes on for ever; here, once reconnects + 1 connections in a row (5 by default, so six) - the first or a reconnection - have brought no event, open answers the last one's failure (SocketError, HTTPError, or an SSEError for a stream that ended cleanly); a connection that brings one starts the count again. There is no backoff past the reconnection time, which §9.2.3 leaves to the user agent.
Two decisions where the standard's text and a browser differ or are silent. The last event ID buffer of a new connection's stream starts from the last event ID, not empty: §9.2.6 initializes each stream's buffers to the empty string, which would make the first event of a resumed stream that carries no id: reset the ID and lose the place, and Chromium's EventSource hands the ID to the new parser ("The new parser takes over the event ID", event_source.cc). And an ID that holds a control character other than a tab is kept and handed to the block but not sent: std/http refuses such a byte in any header value, and a server that sent one should not make its client panic.
**The parser.** EventParser is §9.2.6's interpretation, fed bytes as they arrive in pieces of any size: UTF-8 decoded with ill-formed bytes as U+FFFD (std/unicode's clean, the maximal-subpart rule the Encoding Standard's decoder follows), one leading BOM dropped, lines ended by CRLF, LF or CR - a CR at the end of one piece and an LF at the start of the next being one CRLF -, a line that starts with : read past, a line without a colon a field with an empty value, one space after the colon dropped, field names compared exactly. data lines are joined with LF, the last LF removed; an id with a NUL is ignored; a retry of ASCII digits alone sets the reconnection time (saturating at 2147483647 ms, about 24 days, past what sleep takes); a blank line sets the last event ID and dispatches the event when its data is not empty, and what a stream ends with before its blank line is discarded. bench/std_sse_exercise.sh holds it to the standard's own examples and to a Python transcription of the algorithm over a generated corpus fed in pieces of one to seven bytes, and runs the client against a Python server that drops the connection mid-stream, and the server under Python's and curl's reading.
Not here: a limit on a line or an event's size (a browser has none either; a stream is as long as the server makes it), withCredentials and CORS, which are a browser's, and a client that reconnects with backoff.
Eventstruct
One dispatched event: its type (message unless the stream named one with event:), its data, and the last event ID as it stood when it was dispatched, which an event without an id: field inherits.
pub struct Event
def data : String
def event : String
def id : String
def initialize(event : String, data : String, id : String)
endEventParserclass
§9.2.6's interpretation of an event stream, fed its bytes as they come: feed(bytes) { |event| ... } yields each event a blank line dispatches, and keeps what a piece ended inside - a line, an event, the CR of a CRLF - for the next. finish discards it, as the end of a stream does. last_event_id is the source's last event ID string, set at each blank line; retry the reconnection time the stream set, nil until it sets one; dispatched the events yielded so far.
pub class EventParser
# A `retry` value: ASCII digits and nothing else, base ten, saturating
# at `MAX_RETRY`; nil for anything else, the empty value included.
protected def self.digits(text : String) : Int32 | ::Nil
def dispatched : Int64
def feed(text : String, & : (Event -> Nil)) : Nil
# The end of the stream: a line or an event not yet ended is discarded,
# not dispatched. The last event ID and the reconnection time stay.
def finish : Nil
# A parser for a stream whose last event ID is *last_event_id* so far:
# a reconnection's, which carries the one before it (see the module's
# header for why).
def initialize(last_event_id : String = "")
def last_event_id : String
def retry : Int32 | ::Nil
endEventWriterclass
What SSE.stream's block writes with: an event a send, a comment a comment, each one chunk sent at once. The first write that fails - the client went - is kept, and it and every write after it answer it; gone? says whether one has.
pub class EventWriter
# A comment, a `:` line per line of *text*: read past by every client,
# and a write that finds a client gone.
def comment(text : String = "") : Nil | Cancelled | SocketError
# Whether a write found the client gone.
def gone? : Bool
def initialize(writer : BodyWriter)
# *text* cut at each CRLF, CR and LF: one line for a text without one,
# an empty last line for a text that ends with one.
protected def self.lines(text : String) : Array(String)
# One event: `event:`, `id:` and `retry:` when given, *data* a `data:`
# line per line (CRLF, CR and LF each end one, and each arrives as LF),
# then the blank line that dispatches it. An empty *data* is an event
# whose data is empty; an empty *id* resets the client's last event ID.
def send(data : String, event : String | ::Nil = nil, id : String | ::Nil = nil, retry : Int32 | ::Nil = nil) : Nil | Cancelled | SocketError
protected def self.utf8(text : String, what : String) : Nil
endSSEstruct
pub struct SSE
# Whether a `Content-Type` is `text/event-stream`, in any case, with any
# parameters.
def self.event_stream?(type : String) : Bool
# One connection's answer read: true when the block stopped, false when
# the stream ended, or why it could not be read.
protected def self.follow(response : Response, parser : EventParser, block : Proc(Event, Bool)) : Bool | SSEError | Cancelled | SocketError | HTTPError
def initialize
# The event stream at *url*, each event handed to *block*, which answers
# true to go on and false to stop (nil, the connection closed). A
# connection that drops or ends is made again after the reconnection
# time - *retry* ms until the stream sets its own - with the last event
# ID (*last_event_id* to start with, to resume a stream) in
# `Last-Event-ID`. Once *reconnects* + 1 connections in a row have
# brought no event, the last one's failure is the answer. An answer
# that is not a 200 `text/event-stream` is an `SSEError` at once.
# *headers* go with every request; `Accept` and `Last-Event-ID` are the
# stream's own and are refused there.
def self.open(url : String, headers : Hash(String, String) = {} of String => String, last_event_id : String | ::Nil = nil, reconnects : Int32 = RECONNECTS, retry : Int32 = RETRY, &block : (Event -> Bool)) : Nil | Cancelled | SocketError | HTTPError | SSEError
# The answer to a request for an event stream: a 200, `text/event-stream`,
# `Cache-Control: no-cache` unless *headers* names its own, whose body
# *block* writes through an `EventWriter` as it goes. The stream ends
# when the block returns. A `Content-Type` in *headers* is refused.
def self.stream(headers : Hash(String, String) = {} of String => String, &block : (EventWriter -> Nil)) : Response
endSSEErrorstruct
What the other end did that fails an event stream (§9.2.3), as a value: an answer that is not a 200, one that is not text/event-stream, or a stream that kept ending with no event until the client gave up. status is the answer's. message says it is SSE's.
pub struct SSEError
def initialize(reason : String, status : Int32 = 200)
def reason : String
def status : Int32
def to_s : String
endError for SSEErrorimpl
impl Error for SSEError
def message : String
end
Server-sent events (the WHATWG HTML Living Standard, §9.2): a stream of text events over one long HTTP response,
text/event-stream, for both ends of std/http.