Feature: dynamic http server

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pollnet"
-version = "1.0.4"
+version = "1.1.0"
 authors = ["probable-basilisk <[email protected]>"]
 edition = "2021"
 

bindings/c/pollnet.h

@@ -216,6 +216,27 @@ sockethandle_t pollnet_serve_static_http(pollnet_ctx* ctx, const char* addr, con
  */
 sockethandle_t pollnet_serve_http(pollnet_ctx* ctx, const char* addr);
 
+/*
+ * Open a dynamic HTTP server on the given TCP address and port. The server
+ * acts like a TCP/WS server in that the server socket itself returns individual
+ * client sockets, each corresponding to a single HTTP request.
+ *
+ * A client socket in turn will emit three messages: 
+ *  1. request method+path,
+ *  2. request headers
+ *  3. request body
+ *
+ * And expects three responses:
+ *  1. response code (e.g., 200)
+ *  2. response headers
+ *  3. response body
+ *
+ * If keep_alive is true, then a single client socket can keep repeating this pattern
+ * to serve multiple requests on the same TCP socket; otherwise, the socket will be closed
+ * after the first request-response round.
+ */
+sockethandle_t pollnet_serve_dynamic_http(pollnet_ctx* ctx, const char* addr, bool keep_alive);
+
 /*
  * Add a virtual file to an HTTP server socket: a request to the path
  * will return the provided data. If the server also serves physical files,

bindings/luajit/pollnet.lua

@@ -37,6 +37,7 @@ uint32_t pollnet_get_error(pollnet_ctx* ctx, sockethandle_t handle, char* dest,
 sockethandle_t pollnet_get_connected_client_handle(pollnet_ctx* ctx, sockethandle_t handle);
 sockethandle_t pollnet_listen_ws(pollnet_ctx* ctx, const char* addr);
 sockethandle_t pollnet_serve_static_http(pollnet_ctx* ctx, const char* addr, const char* serve_dir);
+sockethandle_t pollnet_serve_dynamic_http(pollnet_ctx* ctx, const char* addr, bool keep_alive);
 sockethandle_t pollnet_serve_http(pollnet_ctx* ctx, const char* addr);
 void pollnet_add_virtual_file(pollnet_ctx* ctx, sockethandle_t handle, const char* filename, const char* filedata, uint32_t filesize);
 void pollnet_remove_virtual_file(pollnet_ctx* ctx, sockethandle_t handle, const char* filename);
@@ -136,11 +137,41 @@ local function format_headers(headers)
   table.sort(keys)
   local frags = {}
   for idx, name in ipairs(keys) do
-    frags[idx] = ("%s:%s"):format(name, headers[name])
+    local val = headers[name]
+    if type(val) == 'string' then
+      table.insert(frags, ("%s:%s"):format(name, val))
+    else -- assume table representing a duplicated header
+      for _, subval in ipairs(val) do
+        table.insert(frags, ("%s:%s"):format(name, subval))
+      end
+    end
   end
   return table.concat(frags, "\n")
 end
 
+local function parse_headers(headers_str)
+  local headers = {}
+  for line in headers_str:gmatch("[^\n]+") do
+    local key, val = line:match("^([^:]*):(.*)$")
+    if key then headers[key:lower()] = val end
+  end
+  return headers
+end
+
+local function parse_query(s)
+  local queries = {[1]=s}
+  for k, v in s:gmatch("([^=&]+)=([^&]+)") do
+    queries[k] = v
+  end
+  return queries
+end
+
+local function parse_method(s)
+  local method, path, query = s:match("^(%w+) ([^?]+)%??(.*)$")
+  local queries = parse_query(query)
+  return method, path, queries
+end
+
 function socket_mt:http_get(url, headers, ret_body_only)
   headers = format_headers(headers or "")
   ret_body_only = not not ret_body_only
@@ -199,14 +230,21 @@ function socket_mt:remove_virtual_file(filename)
   pollnet.pollnet_remove_virtual_file(_ctx, self._socket, filename)
 end
 
-function socket_mt:listen_ws(addr)
+function socket_mt:listen_ws(addr, callback)
+  if callback then self:on_connection(callback) end
   return self:_open(pollnet.pollnet_listen_ws, addr)
 end
 
-function socket_mt:listen_tcp(addr)
+function socket_mt:listen_tcp(addr, callback)
+  if callback then self:on_connection(callback) end
   return self:_open(pollnet.pollnet_listen_tcp, addr)
 end
 
+function socket_mt:serve_dynamic_http(addr, keep_alive, callback)
+  if callback then self:on_connection(callback) end
+  return self:_open(pollnet.pollnet_serve_dynamic_http, addr, keep_alive or false)
+end
+
 function socket_mt:on_connection(f)
   self._on_connection = f
   return self
@@ -272,6 +310,33 @@ function socket_mt:poll()
   end
 end
 
+function socket_mt:await()
+  local yield_count = 0
+  while true do
+    if self.timeout and (yield_count > self.timeout) then
+      return false, "timeout"
+    end
+    local happy, msg = self:poll()
+    if not happy then 
+      self:close()
+      return false, "error: " .. tostring(msg)
+    end
+    if msg then return msg end
+    yield_count = yield_count + 1
+    coroutine.yield()
+  end
+end
+
+function socket_mt:await_n(count)
+  local parts = {}
+  for idx = 1, count do
+    local part, err = self:await()
+    if not part then return false, err end
+    parts[idx] = part
+  end
+  return parts
+end
+
 function socket_mt:last_message()
   return self._last_message
 end
@@ -286,65 +351,143 @@ function socket_mt:send(msg)
   pollnet.pollnet_send(_ctx, self._socket, msg)
 end
 
+function socket_mt:send_binary(msg)
+  assert(self._socket)
+  assert(type(msg) == 'string', "Argument to send must be a string")
+  pollnet.pollnet_send_binary(_ctx, self._socket, msg, #msg)
+end
+
 function socket_mt:close()
   if not self._socket then return end
   pollnet.pollnet_close(_ctx, self._socket)
   self._socket = nil
 end
 
-local function open_ws(url)
-  return Socket():open_ws(url)
+local function get_nanoid()
+  local _id_scratch = ffi.new("int8_t[?]", 128)
+  local msg_size = pollnet.pollnet_get_nanoid(_id_scratch, 128)
+  return ffi.string(_id_scratch, msg_size)
 end
 
-local function listen_ws(addr)
-  return Socket():listen_ws(addr)
+local function sleep_ms(ms)
+  pollnet.pollnet_sleep_ms(ms)
 end
 
-local function open_tcp(addr)
-  return Socket():open_tcp(addr)
+local reactor_mt = {}
+local function Reactor()
+  local ret = setmetatable({}, {__index = reactor_mt})
+  ret:init()
+  return ret
 end
 
-local function listen_tcp(addr)
-  return Socket():listen_tcp(addr)
+function reactor_mt:init()
+  self.threads = {}
 end
 
-local function serve_http(addr, dir)
-  return Socket():serve_http(addr, dir)
+function reactor_mt:run(thread_body)
+  local thread = coroutine.create(function()
+    thread_body(self)
+  end)
+  self.threads[thread] = true
 end
 
-local function http_get(url, headers, return_body_only)
-  return Socket():http_get(url, headers, return_body_only)
+function reactor_mt:run_server(server_sock, client_body)
+  server_sock:on_connection(function(client_sock, addr)
+    self:run(function()
+      client_body(client_sock, addr)
+    end)
+  end)
+  self:run(function()
+    while true do server_sock:await() end
+  end)
 end
 
-local function http_post(url, headers, body, return_body_only)
-  return Socket():http_post(url, headers, body, return_body_only)
+function reactor_mt:log(...)
+  print(...)
 end
 
-local function get_nanoid()
-  local _id_scratch = ffi.new("int8_t[?]", 128)
-  local msg_size = pollnet.pollnet_get_nanoid(_id_scratch, 128)
-  return ffi.string(_id_scratch, msg_size)
+function reactor_mt:update()
+  local live_count = 0
+  local cur_threads = self.threads
+  self.threads = {}
+  for thread, _ in pairs(cur_threads) do
+    if coroutine.status(thread) == "dead" then
+      cur_threads[thread] = nil
+    else
+      live_count = live_count + 1
+      local happy, err = coroutine.resume(thread)
+      if not happy then self:log("Error", err) end
+    end
+  end
+  for thread, _ in pairs(self.threads) do
+    live_count = live_count + 1
+    cur_threads[thread] = true
+  end
+  self.threads = cur_threads
+  return live_count
 end
 
-local function sleep_ms(ms)
-  pollnet.pollnet_sleep_ms(ms)
+local function invoke_handler(handler, req, expose_errors)
+  local happy, res = pcall(handler, req)
+  if happy then 
+    return res 
+  else
+    return {
+      status = "500", 
+      body = (expose_errors and tostring(res)) or "Internal Error"
+    }
+  end
+end
+
+local function wrap_req_handler(handler, expose_errors)
+  return function(req_sock, addr)
+    while true do
+      local raw_req = req_sock:await_n(3)
+      if not raw_req then break end
+      local method, path, query = parse_method(raw_req[1])
+      local headers = parse_headers(raw_req[2])
+      local reply = invoke_handler(handler, {
+        addr = addr,
+        method = method,
+        path = path,
+        query = query,
+        headers = headers,
+        body = raw_req[3],
+        raw = raw_req
+      }, expose_errors)
+      req_sock:send(reply.status or "404")
+      req_sock:send(format_headers(reply.headers or {}))
+      req_sock:send_binary(reply.body or "")
+    end
+    req_sock:close()
+  end
 end
 
-return {
+local exports = {
   VERSION = POLLNET_VERSION,
   init = init_ctx,
   init_hack_static = init_ctx_hack_static,
   shutdown = shutdown_ctx, 
-  open_ws = open_ws, 
-  listen_ws = listen_ws,
-  open_tcp = open_tcp,
-  listen_tcp = listen_tcp,
-  serve_http = serve_http,
-  http_get = http_get,
-  http_post = http_post,
   Socket = Socket,
+  Reactor = Reactor,
   pollnet = pollnet,
   nanoid = get_nanoid,
   sleep_ms = sleep_ms,
-  format_headers = format_headers
-}
+  format_headers = format_headers,
+  parse_headers = parse_headers,
+  parse_method = parse_method,
+  wrap_req_handler = wrap_req_handler
+}
+
+local fnames = {
+  "open_ws", "listen_ws", "open_tcp", "listen_tcp",
+  "serve_http", "serve_dynamic_http", "http_get", "http_post"
+}
+for _, name in ipairs(fnames) do
+  exports[name] = function(...)
+    local sock = Socket()
+    return sock[name](sock, ...)
+  end
+end
+
+return exports

examples/dyn_http_server.lua

@@ -0,0 +1,56 @@
+local pollnet = require("pollnet")
+
+local FAVICON = {
+  status = "200",
+  headers = {
+    ['content-type'] = "image/svg+xml"
+  },
+  -- Note! SVG documents must have *no whitespace* at front!
+  body = [[<?xml version="1.0" standalone="no"?>
+    <svg viewBox="0 0 200 200" xmlns="http://www.w3.org/2000/svg">
+    <path fill="#FF0066" d="M47.3,-48.8C61.8,-32.9,74.4,-16.4,68.5,-5.8C62.7,4.8,38.5,9.5,24,17.4C9.5,25.3,4.8,36.3,-4.3,40.6C-13.3,44.8,-26.5,42.3,-32.6,34.4C-38.6,26.5,-37.5,13.3,-35.9,1.6C-34.3,-10.1,-32.3,-20.2,-26.3,-36.1C-20.2,-52,-10.1,-73.7,3.2,-76.8C16.4,-80,32.9,-64.6,47.3,-48.8Z" transform="translate(100 100)" />
+    </svg>]]
+}
+
+local function tag(name)
+  return function(children)
+    return ("<%s>%s</%s>"):format(name, table.concat(children), name)
+  end
+end
+
+local counts = {}
+
+local handler = pollnet.wrap_req_handler(function(req)
+  print("METHOD:", req.method)
+  print("PATH:", req.path)
+  print("-- HEADERS --")
+  local header_items = {}
+  counts[req.path] = (counts[req.path] or 0) + 1
+  for k, v in pairs(req.headers) do
+    print(k, "->", v)
+    table.insert(header_items, tag"li"{k, ": ", v})
+  end
+  if req.path == "/favicon.ico" then return FAVICON end
+  local body = tag"html"{tag"body"{
+    tag"p"{req.method, " ", req.path},
+    tag"p"{"This path has been requested ", counts[req.path], " times"},
+    tag"ul"(header_items)
+  }}
+  return {
+    status = "200",
+    headers = {
+      ['Set-Cookie'] = {"session=1", "foobar=hello"}
+    },
+    body = body
+  }
+end, true)
+
+local reactor = pollnet.Reactor()
+local HOST_ADDR = "0.0.0.0:8080"
+reactor:run_server(pollnet.serve_dynamic_http(HOST_ADDR, true), handler)
+
+print("Serving on", HOST_ADDR)
+
+while reactor:update() > 0 do
+  pollnet.sleep_ms(50)
+end