santoku-websocket

one WebSocket API over a pluggable transport: callbacks, text and binary sends, queueing until open, and reconnect, with native, OpenResty and browser backends

santoku-websocket is one WebSocket API for every runtime santoku runs on. The core is plain Lua and takes a transport backend, the way santoku-http takes a socket backend: require("santoku.websocket")(backend). santoku-socket's santoku.socket.websocket is the native backend, santoku-resty's santoku.resty.websocket is the OpenResty client and server, and santoku-web's santoku.web.websocket is the browser's. Code written against the core runs on each of them. Under OpenResty a conn is a cosocket, so only code running in the request that opened or accepted it can send on it, and the shared-room card below shows the pattern for that. The core adds text and binary sends, queueing until open, and reconnect with backoff.

Connect, send and close

connect(url, opts, done) returns a connection at once and calls done(true, conn) when the handshake completes, or done(false, err). conn.send(data, kind) takes kind "text" (the default) or "binary" and returns true, or false and a reason; sends made before open queue and go out in order once it opens. Messages arrive as opts.message(data, kind), and the end of the connection as opts.closed(code, reason). conn.close(code, reason) starts the closing handshake. On the native backend nothing arrives until you call conn.wait(fn, ms) or conn.step(ms), the santoku-socket tab's WebSocket client card. This example runs the core over a small inline backend that echoes every send.

local arr = require("santoku.array")
local backend = {
  connect = function (url, opts, done)
    return done(true, {
      send = function (data, binary)
        opts.message(data, binary and "binary" or "text")
        return true
      end,
      close = function (code, reason)
        opts.closed(code, reason)
        return true
      end,
    })
  end,
}
local ws = require("santoku.websocket")(backend)
local got = {}
local conn = ws.connect("ws://example.test/echo", {
  message = function (data, kind) arr.push(got, kind .. " " .. #data .. " bytes") end,
  closed = function (code, reason) arr.push(got, "closed " .. code .. " " .. reason) end,
}, function (ok) print("open:", ok) end)
print("send:", conn.send("hello"))
print("send:", conn.send("\1\2\3", "binary"))
conn.close(1000, "done")
for i = 1, #got do
  print(got[i])
end
return #got

Reconnect with backoff

opts.reconnect = true, or a table { retries, backoff_ms }, makes the core retry a failed or dropped connection: 3 retries and 1000 ms by default, the wait growing with each attempt. opts.retrying(attempt, code, reason) reports each retry, and sends made meanwhile queue until the connection is back. Retrying needs a backend with after(ms, fn): the browser and OpenResty backends have one, and the native backend has no timer, so reconnect there raises. This backend refuses twice, and its after runs the retry at once instead of waiting.

local tries = 0
local backend = {
  after = function (ms, fn)
    print("retry after", ms, "ms")
    return fn()
  end,
  connect = function (url, opts, done)
    tries = tries + 1
    if tries < 3 then
      return done(false, "refused")
    end
    return done(true, {
      send = function () return true end,
      close = function () return true end,
    })
  end,
}
local ws = require("santoku.websocket")(backend)
ws.connect("ws://example.test/", {
  reconnect = { retries = 3, backoff_ms = 250 },
  retrying = function (attempt, code, reason) print("attempt", attempt, code, reason) end,
}, function (ok) print("open:", ok, "after", tries, "tries") end)
return tries

A backend per runtime

Pick the backend for where the code runs. Native code drives its own loop: the native conn adds step(ms) and wait(fn, ms) from santoku.socket.websocket, and has TLS options (verify, cafile, sslname). Under OpenResty the backend runs the receive loop in a light thread, so callbacks fire while your handler sleeps or does other cosocket work. In the browser the backend sets binaryType to arraybuffer, so binary messages arrive as Lua strings; browsers can't set request headers, so opts.headers fails there.

-- native, with santoku-socket
local ws = require("santoku.websocket")(require("santoku.socket.websocket"))

-- OpenResty, with santoku-resty
local ws = require("santoku.websocket")(require("santoku.resty.websocket"))

-- browser, with santoku-web
local ws = require("santoku.websocket")(require("santoku.web.websocket"))

ws.connect("wss://example.com/feed", {
  protocols = { "chat" },
  message = function (data, kind) print(kind, data) end,
  closed = function (code, reason) print("closed", code, reason) end,
}, function (ok, conn)
  if ok then
    conn.send("hello")
  end
end)

Illustrative: runs outside the browser.

A server under OpenResty

accept(opts, done) upgrades the current request and gives the same conn and events as connect, so one handler serves both sides of a test. Only the OpenResty backend can accept. done runs in the request's own coroutine, then accept spawns the receive loop as a light thread and returns. The request stays open until that thread ends, and the conn's close sends the close frame with your code.

location = /echo {
  content_by_lua_block {
    local ws = require("santoku.websocket")(require("santoku.resty.websocket"))
    local conn
    ws.accept({
      message = function (data, kind) conn.send(data, kind) end,
    }, function (ok, c) conn = c end)
  }
}

Illustrative: runs outside the browser.

A shared room under OpenResty

A conn from accept can only send from its own request. A broadcast that calls another member's conn.send from the sender's message callback raises resty/websocket/protocol.lua: bad request in the sender's receive thread. That thread dies, the sender stops reading, and neither client sees an error. Give each member a queue and an ngx.semaphore instead. broadcast pushes to every queue and posts, and one light thread per connection, spawned in done, waits on its semaphore and does all the sending for that socket. The room lives in a module table, so it's per worker: set workers = "1" in the descriptor's nginx block, or move fan-out to a store every worker reads, such as a lua_shared_dict queue that each connection polls.

-- server/lib/my_api/room.lua
local semaphore = require("ngx.semaphore")
local arr = require("santoku.array")
local members, seq = {}, 0

local function deliver (m, text)
  arr.push(m.queue, text)
  m.sema:post(1)
end

return {
  join = function ()
    seq = seq + 1
    local m = { id = "u" .. seq, queue = {}, sema = semaphore.new() }
    members[m.id] = m
    return m
  end,
  leave = function (m)
    members[m.id] = nil
    m.done = true
    m.sema:post(1)
  end,
  broadcast = function (text)
    for _, m in pairs(members) do
      deliver(m, text)
    end
  end,
  drain = function (m, send)
    while not m.done do
      m.sema:wait(60)
      local queue = m.queue
      m.queue = {}
      for i = 1, #queue do
        if m.done or not send(queue[i]) then
          return
        end
      end
    end
  end,
}

-- server/lib/my_api/web/ws.lua, behind location = /ws
local ws = require("santoku.websocket")(require("santoku.resty.websocket"))
local room = require("my_api.room")
local member
ws.accept({
  message = function (data, kind)
    if kind == "text" then
      room.broadcast(member.id .. ": " .. data)
    end
  end,
  closed = function () room.leave(member) end,
}, function (ok, conn)
  if not ok then
    return ngx.exit(400)
  end
  member = room.join()
  ngx.thread.spawn(room.drain, member, conn.send)
end)

Illustrative: runs outside the browser.

Writing a backend

A backend is a table with connect(url, opts, done), plus after(ms, fn) for reconnect and accept(opts, done) for a server. connect calls done(true, t) or done(false, err), and reports traffic through opts.message(data, kind), opts.closed(code, reason), opts.ping and opts.pong. The transport t has send(data, binary), close(code, reason) and optionally ping(data); fields such as protocol, headers, tls, step and wait pass through to the conn the core returns. The two examples above are complete backends.

local backend = {
  connect = function (url, opts, done) end,
  accept = function (opts, done) end,
  after = function (ms, fn) end,
}

Illustrative: runs outside the browser.