Add SSE sink (socket.sse module)feat/sse

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

9 files changed, 827 insertions, 13 deletions

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 <filename>
 queue=<printername>' 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 [<url>] [<event-count>]
+
     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 [<url>] [<event-count>]
+-- 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")