From cd6b41a13af403d32e08ff416e60982b5cb718dc Mon Sep 17 00:00:00 2001 From: Thijs Schreijer Date: Mon, 31 Aug 2026 22:27:42 +0200 Subject: Add SSE sink (socket.sse module) Implements SSE by using the replacement body sink. Replaces the sink with an SSE parser if the response is a text stream. Example application included --- .github/workflows/build.yml | 3 + luasocket-scm-3.rockspec | 1 + samples/README | 9 ++ samples/sse.lua | 63 ++++++++ src/makefile | 3 +- src/sse.lua | 193 +++++++++++++++++++++++ test/httpfixture.lua | 47 ++++-- test/ssehttptest.lua | 151 ++++++++++++++++++ test/ssetest.lua | 370 ++++++++++++++++++++++++++++++++++++++++++++ 9 files changed, 827 insertions(+), 13 deletions(-) create mode 100644 samples/sse.lua create mode 100644 src/sse.lua create mode 100644 test/ssehttptest.lua create mode 100644 test/ssetest.lua diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index a81649a..d2e27a4 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -43,6 +43,8 @@ jobs: lua testclnt.lua lua testsrvr.lua > /dev/null & lua httpfixturetest.lua + lua testsrvr.lua > /dev/null & + lua ssehttptest.lua lua stufftest.lua lua excepttest.lua lua test_bind.lua @@ -51,6 +53,7 @@ jobs: lua mimetest.lua lua urltest.lua lua headerstest.lua + lua ssetest.lua lua test_socket_error.lua lua test_tcp_receive_zero.lua lua test_udp_receive_zero.lua diff --git a/luasocket-scm-3.rockspec b/luasocket-scm-3.rockspec index 475f594..b89684c 100644 --- a/luasocket-scm-3.rockspec +++ b/luasocket-scm-3.rockspec @@ -67,6 +67,7 @@ local function make_plat(plat) ["socket.ftp"] = "src/ftp.lua", ["socket.headers"] = "src/headers.lua", ["socket.smtp"] = "src/smtp.lua", + ["socket.sse"] = "src/sse.lua", ltn12 = "src/ltn12.lua", socket = "src/socket.lua", mbox = "src/mbox.lua", diff --git a/samples/README b/samples/README index 4ee06b6..5fff925 100644 --- a/samples/README +++ b/samples/README @@ -104,6 +104,15 @@ Unix machines. It uses the lp.lua implementation, in the samples directory. Just run 'lua lpr.lua queue=' and the file will print! + sse.lua -- Server-Sent Events (SSE) client + +This little program uses socket.sse to connect to a text/event-stream +URL and print each event received, stopping after a fixed number of +them (5 by default). With no arguments it connects to a public live +test feed, so you can try it with no setup at all. Just run + + lua sse.lua [] [] + cddb.lua -- CDDB client This is the first try on a simple CDDB client. Not really diff --git a/samples/sse.lua b/samples/sse.lua new file mode 100644 index 0000000..16eece9 --- /dev/null +++ b/samples/sse.lua @@ -0,0 +1,63 @@ +----------------------------------------------------------------------------- +-- Server-Sent Events (SSE) demo client +-- LuaSocket sample files +-- +-- Usage: lua sse.lua [] [] +-- Both arguments are positional and optional; with none given, connects to +-- a public live test feed and stops after 5 events. +----------------------------------------------------------------------------- +local http = require("socket.http") +local sse = require("socket.sse") + +-- default target: Wikimedia's public, continuously-streaming recent-changes +-- feed -- a real, live text/event-stream endpoint, handy for trying out +-- socket.sse without standing up a server of your own +local DEFAULT_URL = "https://stream.wikimedia.org/v2/stream/recentchange" +local DEFAULT_LIMIT = 5 + +-- shortens a long data payload for readable terminal output +local function preview(text, limit) + limit = limit or 100 + if #text > limit then return string.sub(text, 1, limit) .. "..." end + return text +end + +-- builds a message callback that prints each received Message, then stops +-- the stream once "limit" of them have been printed. This module parses a +-- single request/response and never reconnects on its own (see +-- docs/adr/0001-sse-single-shot-no-auto-reconnect.md), so a caller who wants +-- to stop early just does what we do here: return an error from the +-- callback, which ends the request the same way any sink error would +local function makeprinter(limit) + local count = 0 + return function(message) + count = count + 1 + io.write(string.format("[%d] event=%s id=%s\n data=%s\n", + count, message.event, tostring(message.id), preview(message.data))) + if count >= limit then + return nil, string.format("stopped after %d events", limit) + end + return 1 + end +end + +-- main program +arg = arg or {} +local url = arg[1] or DEFAULT_URL +local limit = tonumber(arg[2]) or DEFAULT_LIMIT + +io.write("connecting to ", url, " (stopping after ", limit, " events)\n") +local ok, code = http.request{ + url = url, + headers_callback = sse.responseheaders(makeprinter(limit)) +} + +if ok then + -- the server closed the connection on its own before we hit our limit + io.write("connection closed by server, code=", tostring(code), "\n") +else + -- once the callback above returns an error, http.request reports it the + -- same way it reports any failure: (nil, message) -- so "code" here is + -- really our own "stopped after N events" message, not a real error + io.write(code, "\n") +end diff --git a/src/makefile b/src/makefile index 310fe4e..a6ba91e 100644 --- a/src/makefile +++ b/src/makefile @@ -375,7 +375,8 @@ TO_SOCKET_LDIR= \ tp.lua \ ftp.lua \ headers.lua \ - smtp.lua + smtp.lua \ + sse.lua TO_TOP_LDIR= \ ltn12.lua \ diff --git a/src/sse.lua b/src/sse.lua new file mode 100644 index 0000000..fa77427 --- /dev/null +++ b/src/sse.lua @@ -0,0 +1,193 @@ +----------------------------------------------------------------------------- +-- Server-Sent Events (SSE) parsing support for the Lua language. +-- LuaSocket toolkit. +----------------------------------------------------------------------------- + +----------------------------------------------------------------------------- +-- Declare module and import dependencies +----------------------------------------------------------------------------- +local socket = require("socket") +local string = require("string") +socket.sse = {} +local _M = socket.sse + +----------------------------------------------------------------------------- +-- Program constants +----------------------------------------------------------------------------- +-- maximum size of a single SSE line +_M.MAXLINESIZE = 8192 +-- maximum total size of all lines making up a single event/comment block +_M.MAXEVENTSIZE = 65536 + +----------------------------------------------------------------------------- +-- Strips a single leading space, per the SSE field-value trimming rule +----------------------------------------------------------------------------- +local function striponeleadingspace(value) + if string.sub(value, 1, 1) == " " then return string.sub(value, 2) end + return value +end + +----------------------------------------------------------------------------- +-- Parses one SSE field line ("name: value", "name:value" or "name") into +-- its name/value pair +----------------------------------------------------------------------------- +local function parsefield(line) + local colon = string.find(line, ":", 1, true) + if not colon then return line, "" end + local name = string.sub(line, 1, colon - 1) + local value = striponeleadingspace(string.sub(line, colon + 1)) + return name, value +end + +----------------------------------------------------------------------------- +-- Builds a Parser sink: an ltn12 sink that consumes raw SSE response bytes +-- and dispatches parsed Messages (and, if enabled, Comments) to msgsink, a +-- Message sink. config is an optional table: +-- comments: boolean, dispatch Comments to msgsink when true (default false) +-- context: caller-owned table that last_event_id/retry get written into +-- as parsing progresses +----------------------------------------------------------------------------- +function _M.parser(msgsink, config) + config = config or {} + local comments = config.comments + local context = config.context or {} + + local linebuffer = "" + local eventsize = 0 + + -- id persists across Messages that omit their own id: line, per spec, + -- so it lives outside reset(); retry is a stream-level reconnection + -- hint with no per-Message meaning at all, so it is tracked only on + -- context.retry, never as event-local state + local id + local eventtype, data, hasdata + + local function reset() + eventtype = nil + data = nil + hasdata = false + eventsize = 0 + end + reset() + + -- dispatches the current Message, if any data was accumulated for it, + -- then resets event-local state for the next one + local function dispatch() + if not hasdata then + reset() + return 1 + end + local message = { + event = eventtype or "message", + data = data, + id = id, + } + reset() + local ok, err = msgsink(message) + if not ok then return nil, err end + return 1 + end + + local function processline(line) + if line == "" then return dispatch() end + if string.sub(line, 1, 1) == ":" then + if comments then + local text = striponeleadingspace(string.sub(line, 2)) + local ok, err = msgsink({ comment = text }) + if not ok then return nil, err end + end + return 1 + end + local name, value = parsefield(line) + if name == "event" then + eventtype = value + elseif name == "data" then + data = hasdata and (data .. "\n" .. value) or value + hasdata = true + elseif name == "id" then + if not string.find(value, "\0", 1, true) then + id = value + context.last_event_id = id + end + elseif name == "retry" then + if string.find(value, "^%d+$") then + context.retry = tonumber(value) + end + end + return 1 + end + + return function(chunk, err) + if not chunk then + if err then return nil, err end + return 1 + end + linebuffer = linebuffer .. chunk + while true do + local nl = string.find(linebuffer, "\n", 1, true) + if not nl then + if #linebuffer > _M.MAXLINESIZE then return nil, "oversized" end + break + end + local line = string.sub(linebuffer, 1, nl - 1) + linebuffer = string.sub(linebuffer, nl + 1) + if #line > _M.MAXLINESIZE then return nil, "oversized" end + if string.sub(line, -1) == "\r" then line = string.sub(line, 1, -2) end + eventsize = eventsize + #line + 1 + if eventsize > _M.MAXEVENTSIZE then return nil, "oversized" end + local ok, procerr = processline(line) + if not ok then return nil, procerr end + end + return 1 + end +end + +----------------------------------------------------------------------------- +-- The media type this module activates on, ignoring Content-Type parameters +----------------------------------------------------------------------------- +_M.EVENTSTREAMTYPE = "text/event-stream" + +----------------------------------------------------------------------------- +-- Extracts the media type portion of a Content-Type header value, dropping +-- any trailing parameters (e.g. "; charset=utf-8") and normalizing case +----------------------------------------------------------------------------- +local function mediatype(contenttype) + local mt = string.match(contenttype or "", "^%s*([^;%s]*)") + return string.lower(mt or "") +end + +----------------------------------------------------------------------------- +-- Wraps a plain function(message) ... end callback into a Message sink, so +-- it can be used anywhere one is expected. Follows the same contract as any +-- ltn12 sink: the callback must return a truthy value to signal success; +-- any falsy return is an error, propagated as-is (even with no err message) +-- rather than swallowed. A raw sink passed in here already speaks that +-- contract, so wrapping it is a no-op. +----------------------------------------------------------------------------- +function _M.callbacksink(callback) + return function(message) + local ok, err = callback(message) + if not ok then return nil, err end + return 1 + end +end + +----------------------------------------------------------------------------- +-- Builds a function suitable for reqt.headers_callback: given (code, +-- headers, status), checks headers["content-type"] for the +-- text/event-stream media type (ignoring trailing parameters) and, only on +-- a match, offers a sink chaining the Parser sink (see _M.parser) to +-- msgsink -- a Message sink or a plain callback, per _M.callbacksink -- as +-- the request's sink. On no match, declines by returning true with no +-- sink, leaving http.request's own sink handling untouched. +----------------------------------------------------------------------------- +function _M.responseheaders(msgsink, config) + return function(code, headers, status) + if not headers or mediatype(headers["content-type"]) ~= _M.EVENTSTREAMTYPE then + return true + end + return true, _M.parser(_M.callbacksink(msgsink), config) + end +end + +return _M diff --git a/test/httpfixture.lua b/test/httpfixture.lua index 3ea0737..14c7548 100644 --- a/test/httpfixture.lua +++ b/test/httpfixture.lua @@ -49,18 +49,12 @@ local function quote(s) return "\"" .. s .. "\"" end --- Scripts the server to accept one connection (into the shared `data` --- global, mirroring test/testclnt.lua's reconnect()), write a sequence of --- raw byte chunks to it, then close it. `chunks` is an array of either --- plain strings, or {body, delay = seconds} tables -- the delay (via --- socket.sleep) is applied before sending that chunk, so callers can prove --- incremental/partial delivery instead of one atomic send. -function M.accept_and_send(remote, chunks) - local parts = { - "if data then data:close() data = nil end", - "data = server:accept()", - "data:setoption(\"tcp-nodelay\", true)", - } +-- Appends the script parts for one accept-send-close cycle (see +-- accept_and_send_sequence) onto `parts`. +local function append_connection(parts, chunks) + parts[#parts + 1] = "if data then data:close() data = nil end" + parts[#parts + 1] = "data = server:accept()" + parts[#parts + 1] = "data:setoption(\"tcp-nodelay\", true)" for _, chunk in ipairs(chunks) do local body, delay if type(chunk) == "table" then @@ -74,6 +68,35 @@ function M.accept_and_send(remote, chunks) parts[#parts + 1] = "data:send(" .. quote(body) .. ")" end parts[#parts + 1] = "data:close() data = nil" +end + +-- Scripts the server to accept one connection (into the shared `data` +-- global, mirroring test/testclnt.lua's reconnect()), write a sequence of +-- raw byte chunks to it, then close it. `chunks` is an array of either +-- plain strings, or {body, delay = seconds} tables -- the delay (via +-- socket.sleep) is applied before sending that chunk, so callers can prove +-- incremental/partial delivery instead of one atomic send. +function M.accept_and_send(remote, chunks) + local parts = {} + append_connection(parts, chunks) + remote(table.concat(parts, "\n")) +end + +-- Like accept_and_send, but scripts several accept-send-close cycles as one +-- server-side script sent over a single remote() round trip. Needed for a +-- client call that opens more than one connection in sequence within a +-- single call of its own (e.g. socket.http.request following a redirect): +-- queuing a second accept_and_send for that connection ahead of time would +-- deadlock, since its remote() call can't get an ack until the server +-- finishes the first accept_and_send's blocking accept() -- which itself +-- can't complete until the client makes the very call that's stuck waiting +-- on that ack. `connections` is an array of `chunks` arrays, one per +-- connection, handled in order. +function M.accept_and_send_sequence(remote, connections) + local parts = {} + for _, chunks in ipairs(connections) do + append_connection(parts, chunks) + end remote(table.concat(parts, "\n")) end diff --git a/test/ssehttptest.lua b/test/ssehttptest.lua new file mode 100644 index 0000000..b4ba6bc --- /dev/null +++ b/test/ssehttptest.lua @@ -0,0 +1,151 @@ +-- End-to-end coverage proving socket.http.request, the headers_callback +-- hook, and socket.sse cooperate correctly over a real socket -- using the +-- httpfixture.lua harness (see httpfixturetest.lua) instead of Apache or a +-- third-party host. Also covers the two headers_callback interactions that +-- only matter once other request-handling logic (redirects, +-- shouldreceivebody) is in the path. +local socket = require("socket") +local http = require("socket.http") +local ltn12 = require("ltn12") +local sse = require("socket.sse") +local fixture = require("httpfixture") + +dofile("testsupport.lua") + +local url = "http://" .. fixture.host .. ":" .. fixture.port .. "/" + +local control, remote = fixture.connect() + +io.write("testing a live SSE stream dispatches Messages incrementally, not batched until close: ") +do + local received, elapsed = {}, {} + local t0 + local factory = sse.responseheaders(function(message) + table.insert(received, message) + table.insert(elapsed, socket.gettime() - t0) + return true + end) + + -- Framed as HTTP chunked-transfer-encoding, one SSE Message per HTTP + -- chunk: the "until-closed"/"default" body source reads in fixed + -- socket.BLOCKSIZE (2048-byte) gulps, which would silently buffer both + -- of these small Messages into one read and defeat this test; the + -- chunked source instead reads exactly each chunk's declared size, so + -- Message 1 surfaces as soon as its chunk arrives, independent of + -- Message 2's delayed chunk. + local function httpchunk(piece) + return string.format("%x\r\n%s\r\n", #piece, piece) + end + + -- generous relative to the artificial delay so a loaded CI runner's + -- scheduling jitter can't turn correct incremental behavior into a + -- flaky failure; the delay is deliberately long enough that "arrived + -- before it" and "arrived after it" stay unambiguous even with slack + local delay = 1.0 + local headersblock = fixture.response("200 OK", { + "Content-Type: text/event-stream", + "Transfer-Encoding: chunked", + "Connection: close", + }, "") + fixture.accept_and_send(remote, { + headersblock, + httpchunk("data: first\n\n"), + { httpchunk("data: second\n\n"), delay = delay }, + "0\r\n\r\n", + }) + + t0 = socket.gettime() + local ok, code = assert(http.request{ url = url .. "sse", headers_callback = factory }) + assert(ok, "request failed") + assert(code == 200, "status code mismatch: " .. tostring(code)) + assert(#received == 2, "expected two messages, got " .. #received) + assert(received[1].data == "first", "wrong data for message 1: " .. tostring(received[1].data)) + assert(received[2].data == "second", "wrong data for message 2: " .. tostring(received[2].data)) + -- message 1 must have been dispatched well before the server even sent + -- (let alone finished delaying) message 2 -- if delivery were batched + -- until the connection closed instead, both messages would only appear + -- once the full delay had elapsed + assert(elapsed[1] < delay / 2, + "message 1 dispatched too late (" .. elapsed[1] .. "s) to have arrived before message 2's delayed send") + assert(elapsed[2] >= delay / 2, + "message 2 dispatched suspiciously early (" .. elapsed[2] .. "s); the server's artificial delay may not have been exercised") +end +print("ok") + +io.write("testing headers_callback is not invoked for a redirected (3xx) response: ") +do + local calls = {} + local function responseheaders(code) + table.insert(calls, code) + return true + end + + local redirect = fixture.response("302 Found", { + "Location: /final", + "Content-Length: 0", + "Connection: close", + }, "") + local finalbody = "landed" + local final = fixture.response("200 OK", { + "Content-Length: " .. #finalbody, + "Connection: close", + }, finalbody) + fixture.accept_and_send_sequence(remote, { { redirect }, { final } }) + + local target = {} + local ok, code = assert(http.request{ + url = url .. "redirect-me", + sink = ltn12.sink.table(target), + headers_callback = responseheaders, + }) + assert(ok, "request failed") + assert(table.concat(target) == finalbody, "expected the redirected response's body") + assert(code == 200, "expected the final response's status code, got " .. tostring(code)) + assert(#calls == 1, "expected headers_callback to be invoked exactly once, got " .. #calls) + assert(calls[1] == 200, "headers_callback must only ever see the final response, got " .. tostring(calls[1])) +end +print("ok") + +-- Shared by the three shouldreceivebody variants below (204, 304, HEAD): +-- scripts `response`, issues a request built from `reqtextra`, and asserts +-- headers_callback still fires with `expectedcode` while the sink it offers +-- is never actually invoked, since shouldreceivebody skips the body. +local function assertsinkswapskipped(reqtextra, response, expectedcode) + local sinkcalls, invokedwith = 0, nil + local function responseheaders(code) + invokedwith = code + return true, function() sinkcalls = sinkcalls + 1; return 1 end + end + fixture.accept_and_send(remote, { response }) + + local reqt = { url = url .. "skip-body", headers_callback = responseheaders } + for k, v in pairs(reqtextra or {}) do reqt[k] = v end + local ok, code = assert(http.request(reqt)) + assert(ok, "request failed") + assert(code == expectedcode, "status code mismatch: " .. tostring(code)) + assert(invokedwith == expectedcode, "expected headers_callback to still be invoked, got " .. tostring(invokedwith)) + assert(sinkcalls == 0, "the offered sink must never run when shouldreceivebody skips the body") +end + +io.write("testing the headers_callback sink-swap offer is inert when a 204 response skips the body: ") +assertsinkswapskipped(nil, fixture.response("204 No Content", { "Connection: close" }, ""), 204) +print("ok") + +io.write("testing the headers_callback sink-swap offer is inert when a 304 response skips the body: ") +assertsinkswapskipped(nil, fixture.response("304 Not Modified", { "Connection: close" }, ""), 304) +print("ok") + +io.write("testing the headers_callback sink-swap offer is inert when a HEAD request skips the body: ") +do + local body = "should never be read" + local response = fixture.response("200 OK", { + "Content-Length: " .. #body, + "Connection: close", + }, body) + assertsinkswapskipped({ method = "HEAD" }, response, 200) +end +print("ok") + +remote("os.exit()") + +print("the library passed all tests") diff --git a/test/ssetest.lua b/test/ssetest.lua new file mode 100644 index 0000000..03206ba --- /dev/null +++ b/test/ssetest.lua @@ -0,0 +1,370 @@ +local socket = require("socket") +local sse = require("socket.sse") + +dofile("testsupport.lua") + +-- collects everything a parser sink dispatches to it (Messages and, when +-- enabled, Comments) into an ordered list; the "collect" name signals it +-- passes through unchanged rather than transforming +local function collect() + local list = {} + local snk = function(item, err) + if err then return nil, err end + table.insert(list, item) + return 1 + end + return snk, list +end + +-- feeds a parser sink the given raw chunks, then signals end of stream +local function feed(parser, ...) + for _, chunk in ipairs({...}) do + local ok, err = parser(chunk) + if not ok then return nil, err end + end + return parser(nil) +end + +-------------------------------- +io.write("testing event/data/id parsing on a single Message, retry line consumed but not attached: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local ok = feed(parser, + "event: greeting\r\n" .. + "data: hello\r\n" .. + "id: 1\r\n" .. + "retry: 3000\r\n" .. + "\r\n") + assert(ok, "parser returned error") + assert(#list == 1, "expected exactly one message") + assert(list[1].event == "greeting", "wrong event") + assert(list[1].data == "hello", "wrong data") + assert(list[1].id == "1", "wrong id") + assert(list[1].retry == nil, "Message must not carry a retry field; retry only lives on the context table") + print("ok") +end + +-------------------------------- +io.write("testing multiple data: lines are joined with \\n: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local ok = feed(parser, "data: line one\ndata: line two\ndata: line three\n\n") + assert(ok, "parser returned error") + assert(#list == 1, "expected exactly one message") + assert(list[1].data == "line one\nline two\nline three", "data not joined correctly: " .. list[1].data) + print("ok") +end + +-------------------------------- +io.write("testing event defaults to 'message' when omitted: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local ok = feed(parser, "data: no event here\n\n") + assert(ok, "parser returned error") + assert(#list == 1, "expected exactly one message") + assert(list[1].event == "message", "expected default event type, got " .. tostring(list[1].event)) + print("ok") +end + +-------------------------------- +io.write("testing id persists forward onto later Messages: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local ok = feed(parser, + "id: abc\ndata: first\n\n" .. + "data: second\n\n" .. + "id: def\ndata: third\n\n" .. + "data: fourth\n\n") + assert(ok, "parser returned error") + assert(#list == 4, "expected four messages, got " .. #list) + assert(list[1].id == "abc", "message 1 id") + assert(list[2].id == "abc", "message 2 id should persist from message 1") + assert(list[3].id == "def", "message 3 id") + assert(list[4].id == "def", "message 4 id should persist from message 3") + print("ok") +end + +-------------------------------- +io.write("testing comment lines are silently absorbed when comments is off: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local ok = feed(parser, ": this is a comment\ndata: real message\n\n") + assert(ok, "parser returned error") + assert(#list == 1, "expected only the message, comment should be absorbed") + assert(list[1].data == "real message", "wrong data") + print("ok") +end + +-------------------------------- +io.write("testing comment lines are dispatched distinctly when comments is on: ") +do + local snk, list = collect() + local parser = sse.parser(snk, { comments = true }) + local ok = feed(parser, ": keep-alive\ndata: real message\n\n") + assert(ok, "parser returned error") + assert(#list == 2, "expected comment and message, got " .. #list) + assert(list[1].comment == "keep-alive", "wrong comment text: " .. tostring(list[1].comment)) + assert(list[1].event == nil, "comment must not look like a message") + assert(list[2].data == "real message", "wrong data") + print("ok") +end + +-------------------------------- +io.write("testing context table gets last_event_id/retry written and retains them after parsing ends: ") +do + local snk = collect() + local context = {} + local parser = sse.parser(snk, { context = context }) + local ok = feed(parser, "id: xyz\nretry: 5000\ndata: hi\n\n") + assert(ok, "parser returned error") + assert(context.last_event_id == "xyz", "context.last_event_id not written") + assert(context.retry == 5000, "context.retry not written") + print("ok") +end + +-------------------------------- +io.write("testing context.retry persists across later events that don't repeat it, and no Message ever carries a retry field: ") +do + local snk, list = collect() + local context = {} + local parser = sse.parser(snk, { context = context }) + local ok = feed(parser, + "retry: 2000\ndata: first\n\n" .. + "data: second\n\n") + assert(ok, "parser returned error") + assert(#list == 2, "expected two messages, got " .. #list) + assert(list[1].retry == nil, "message 1 must not carry a retry field") + assert(list[2].retry == nil, "message 2 must not carry a retry field") + assert(context.retry == 2000, "context.retry should still reflect the most recent retry hint after a later event that didn't repeat it") + print("ok") +end + +-------------------------------- +io.write("testing an oversized line produces a distinct error and aborts: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local huge = string.rep("x", sse.MAXLINESIZE + 1) + local ok, err = feed(parser, "data: " .. huge .. "\n\n") + assert(not ok, "expected parser to fail on oversized line") + assert(err == "oversized", "expected 'oversized' error, got " .. tostring(err)) + assert(#list == 0, "no message should have been dispatched") + print("ok") +end + +-------------------------------- +io.write("testing an oversized event (many lines under MAXEVENTSIZE total) produces a distinct error: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local line = "data: " .. string.rep("y", 100) .. "\n" + local lines = string.rep(line, math.ceil(sse.MAXEVENTSIZE / #line) + 1) + local ok, err = feed(parser, lines) + assert(not ok, "expected parser to fail on oversized event") + assert(err == "oversized", "expected 'oversized' error, got " .. tostring(err)) + assert(#list == 0, "no message should have been dispatched") + print("ok") +end + +-------------------------------- +io.write("testing a field split across two raw-byte chunks is parsed correctly: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + -- split mid field-name, mid value, and mid line-terminator + local ok = feed(parser, "eve", "nt: greet", "ing\ndata: hel", "lo\r", "\n\r\n") + assert(ok, "parser returned error") + assert(#list == 1, "expected exactly one message, got " .. #list) + assert(list[1].event == "greeting", "wrong event: " .. tostring(list[1].event)) + assert(list[1].data == "hello", "wrong data: " .. tostring(list[1].data)) + print("ok") +end + +-------------------------------- +io.write("testing a Message-sink error aborts the parser sink chain: ") +do + local calls = 0 + local snk = function(message) + calls = calls + 1 + if calls == 1 then return 1 end + return nil, "sink refused message" + end + local parser = sse.parser(snk) + local ok, err = feed(parser, "data: first\n\ndata: second\n\n") + assert(not ok, "expected parser to propagate message sink error") + assert(err == "sink refused message", "unexpected error: " .. tostring(err)) + assert(calls == 2, "expected the sink to be invoked twice") + print("ok") +end + +-------------------------------- +io.write("testing a Message with no data: line is not dispatched: ") +do + local snk, list = collect() + local parser = sse.parser(snk) + local ok = feed(parser, "event: ping\nid: 1\n\ndata: real\n\n") + assert(ok, "parser returned error") + assert(#list == 1, "the dataless block should not have produced a message") + assert(list[1].data == "real", "wrong data") + print("ok") +end + +-------------------------------- +io.write("testing callbacksink lets a plain callback act as a message sink: ") +do + local received = {} + local callback = function(message) table.insert(received, message); return true end + local snk = sse.callbacksink(callback) + local parser = sse.parser(snk) + local ok = feed(parser, "data: first\n\ndata: second\n\n") + assert(ok, "parser returned error") + assert(#received == 2, "expected two messages, got " .. #received) + assert(received[1].data == "first", "wrong data") + assert(received[2].data == "second", "wrong data") + print("ok") +end + +-------------------------------- +io.write("testing callbacksink propagates an error returned by the callback: ") +do + local calls = 0 + local callback = function(message) + calls = calls + 1 + if message.data == "bad" then return nil, "callback refused" end + return true + end + local snk = sse.callbacksink(callback) + local parser = sse.parser(snk) + local ok, err = feed(parser, "data: good\n\ndata: bad\n\ndata: unreachable\n\n") + assert(not ok, "expected callback error to abort the chain") + assert(err == "callback refused", "unexpected error: " .. tostring(err)) + assert(calls == 2, "expected the callback to stop being invoked after it errors, got " .. calls) + print("ok") +end + +-------------------------------- +io.write("testing callbacksink does not swallow a sink that fails with a falsy ok and no error message: ") +do + local calls = 0 + local rawsink = function(message) + calls = calls + 1 + if message.data == "bad" then return false end + return 1 + end + local snk = sse.callbacksink(rawsink) + local parser = sse.parser(snk) + local ok, err = feed(parser, "data: good\n\ndata: bad\n\ndata: unreachable\n\n") + assert(not ok, "expected the falsy ok to abort the chain even without an error message") + assert(err == nil, "no error message was given, so none should be invented") + assert(calls == 2, "expected the sink to stop being invoked once it fails, got " .. calls) + print("ok") +end + +-------------------------------- +io.write("testing callbacksink treats a callback that returns nothing as an error, not a default success: ") +do + local snk = sse.callbacksink(function(message) end) + local parser = sse.parser(snk) + local ok = feed(parser, "data: hi\n\n") + assert(not ok, "a callback must explicitly return truthy to signal success, matching a raw sink's contract") + print("ok") +end + +-------------------------------- +io.write("testing responseheaders forwards its config (context/comments) to the parser: ") +do + local received = {} + local context = {} + local factory = sse.responseheaders( + function(message) table.insert(received, message); return true end, + { comments = true, context = context }) + local ok, sink = factory(200, { ["content-type"] = "text/event-stream" }, "HTTP/1.1 200 OK") + assert(ok, "expected success") + local fed = feed(sink, ": keep-alive\nid: xyz\ndata: hello\n\n") + assert(fed, "sink chain returned an error") + assert(#received == 2, "expected the comment and the message, got " .. #received) + assert(received[1].comment == "keep-alive", "comments config was not forwarded to the parser") + assert(context.last_event_id == "xyz", "context table was not forwarded to the parser") + print("ok") +end + +-------------------------------- +io.write("testing responseheaders installs the sink chain on a matching Content-Type: ") +do + local received = {} + local factory = sse.responseheaders(function(message) table.insert(received, message); return true end) + local reqt = {} + reqt.headers_callback = factory + local ok, sink = reqt.headers_callback(200, { ["content-type"] = "text/event-stream" }, "HTTP/1.1 200 OK") + assert(ok, "expected success on a matching content-type") + assert(type(sink) == "function", "expected a sink function on a matching content-type") + reqt.sink = sink + local fed = feed(reqt.sink, "data: hello\n\n") + assert(fed, "sink chain returned an error") + assert(#received == 1, "expected one message dispatched through the installed sink") + assert(received[1].data == "hello", "wrong data") + print("ok") +end + +-------------------------------- +io.write("testing responseheaders also accepts a raw ltn12-style message sink directly: ") +do + local received = {} + -- a raw sink following full ltn12 protocol: truthy on success, (nil, err) on failure + local rawsink = function(message) + table.insert(received, message) + return 1 + end + local factory = sse.responseheaders(rawsink) + local ok, sink = factory(200, { ["content-type"] = "text/event-stream" }, "HTTP/1.1 200 OK") + assert(ok, "expected success") + local fed = feed(sink, "data: raw\n\n") + assert(fed, "sink chain returned an error") + assert(#received == 1, "expected one message") + assert(received[1].data == "raw", "wrong data") + print("ok") +end + +-------------------------------- +io.write("testing responseheaders matches a Content-Type with trailing parameters: ") +do + local factory = sse.responseheaders(function() end) + local ok, sink = factory(200, { ["content-type"] = "text/event-stream; charset=utf-8" }, "HTTP/1.1 200 OK") + assert(ok, "expected success") + assert(type(sink) == "function", "expected a sink for a parameterized but matching content-type") + print("ok") +end + +-------------------------------- +io.write("testing responseheaders declines cleanly on a non-matching Content-Type: ") +do + local calledback = false + local factory = sse.responseheaders(function() calledback = true end) + local reqt = { sink = "original sink placeholder" } + local ok, sink = factory(200, { ["content-type"] = "text/html" }, "HTTP/1.1 200 OK") + assert(ok, "a non-matching content-type must still succeed (just decline the sink swap)") + assert(sink == nil, "a non-matching content-type must not offer a sink swap") + -- mirror what http.lua does with the factory's return values, to prove + -- the caller's existing sink survives untouched + if sink then reqt.sink = sink end + assert(reqt.sink == "original sink placeholder", "existing sink must be left untouched") + assert(not calledback, "message callback must not fire for a non-matching content-type") + print("ok") +end + +-------------------------------- +io.write("testing responseheaders declines cleanly when Content-Type is missing: ") +do + local factory = sse.responseheaders(function() end) + local ok, sink = factory(200, {}, "HTTP/1.1 200 OK") + assert(ok, "missing content-type must still succeed (just decline the sink swap)") + assert(sink == nil, "missing content-type must not offer a sink swap") + print("ok") +end + +print("the library passed all tests") -- cgit v1.2.3-55-g6feb