connection

Declared connection classes, their handlers, and the conn value.

Declare a class

local Session = connection "Session" {
    open = function(conn) end,
    frame = function(conn, data) end,
    event = function(conn, event) end,
    timer = { every = "30s", run = function(conn, missed) end },
    close = function(conn) end,
}

Each handler is optional; a key outside these five refuses at declaration. Nothing calls a connection, so there are no methods: handlers run when the wire, an edge or the timer delivers.

Handlers for one connection run one at a time, in arrival order, each under the per-call budget. A tick is never queued behind the inbox, and a connection with a timer never hibernates.

The forward policy

event takes a policy in place of a function, the same hook-or-policy shape publishes uses for topics:

local Account = objects "Account"

local Feed = connection "Feed" {
    open = function(conn) conn:follow(Account(conn.name), "events") end,
    event = "forward",
}

Each delivery then reaches the client as

{ "type": "event", "topic": "...",
  "from": { "id": "...", "class": "...", "name": "..." },
  "data": { } }

built by the platform, with no vm involved, so a hibernated connection stays hibernated while it relays. Declare a function instead to choose your own wire shape, and pay a handler call per event for it.

Declaring neither drops deliveries, which is what you want only when the follow exists to be counted (state:followers) rather than heard.

Upgrading

if request.upgrade and request.path == "/ws" then
    return request:upgrade(Session, { room = "lobby" }, User(name))
end

request:upgrade(Class, seed?, identity) returns the value your fetch handler should return. The seed becomes the initial conn.state; the identity is an object handle the connection speaks as for the rest of its life. Two arguments means no seed.

conn

The event a handler receives is shaped { topic = "...", from = { id, class, name }, data = {...} }.

Opening a wire outward

The same class dials as well as accepts:

local name = Upstream:open("wss://api.example/realtime", { room = "lobby" }, {
    headers = { Authorization = "Bearer " .. key },
    protocols = { "json" },
})

The connection speaks as its own class and the minted name, lives on the node that opened it, and is listed in the console's Connections page beside the inbound ones. There is no socket handle: to send up the wire, publish on a topic the connection follows, and send the payload from its event handler. See wires this project opened for the pattern and who should own the reopen.

Streaming a response

The same class reads an HTTP response as frames:

local name = Model:stream({
    url = "https://api.openai.com/v1/chat/completions",
    headers = { authorization = "Bearer " .. secret "OPENAI_API_KEY" },
    body = { model = "gpt-4o-mini", stream = true, messages = messages },
}, { chat = chat })

How the body becomes frames:

Each frame reaches frame(conn, data) as json when it parses and as a string when it does not, so data: [DONE] arrives as the string "[DONE]". A frame longer than 1 MiB without its delimiter ends the wire. The body's end runs close with by = "peer". A stream has no uplink: conn:send raises on it, and its event handler receives what it follows like any connection.

A wire's liveness

Class(name) is the handle of a wire this project opened, from the name open or stream returned:

if not Upstream(state.store:get("upstream")):alive() then
    state.store:delete("upstream")
end

Example

local Account = objects "Account"
local Auction = objects "Auction"

local Feed = connection "Feed" {
    open = function(conn)
        conn:follow(Account(conn.name), "events")
    end,

    -- The platform envelope, without waking this program.
    event = "forward",

    frame = function(conn, data)
        if data.op == "watch" then
            if conn.state.watching then
                conn:unfollow(Auction(conn.state.watching), "bid")
            end
            conn.state.watching = data.id
            conn:follow(Auction(data.id), "bid")
        end
    end,
}

on "fetch" (function(request)
    if request.upgrade and request.query.user then
        return request:upgrade(Feed, nil, Account(request.query.user))
    end
    return { status_code = 404, body = "" }
end)

Gotchas

• Locals in the fetch handler do not reach the handlers; a session's context travels through the seed into conn.state.
• conn.state dies with the connection on purpose. Anything that must survive belongs on an object one hop back.
• Connection edges are at-most-once and die with the connection.
• The inbox holds 256 items. Deliveries that overflow it sever every edge the connection holds and leave the socket open; client frames that overflow it close the wire.
• The only frame the platform defines is the forwarded event envelope above, and declaring an event handler replaces it. Every other frame on the wire is one your handlers send or interpret, so follow targets come from your code, never from a string the client sent.

Related

• Websockets: the upgrade path and identity.
• Streams: the publisher side.