runtime: native SSE streaming — http_sse_open/send/close
Add Server-Sent Events support to the El runtime. El v2 handlers can now
hold HTTP connections open and push events in real time.
New builtins in el_seed.c:
__http_conn_fd() — retrieve raw fd from thread-local set by worker
__http_sse_open(fd) — send SSE headers (text/event-stream), keep-alive
__http_sse_send(fd, data) — write "data: <data>\n\n" frame
__http_sse_close(fd) — close the connection fd
http_worker_v2 in legacy/el_runtime.c now:
- stashes the fd via el_seed_set_http_conn_fd() before calling the handler
- detects the "__sse__" sentinel return value to skip http_send_response
and skip close(fd) — SSE handler took ownership of the fd
- clears the thread-local after the handler returns
El wrappers added to runtime/http.el:
http_conn_fd() http_sse_open(fd) http_sse_send(fd, data)
http_sse_close(fd) http_sse_sentinel()
This commit is contained in:
@@ -554,6 +554,94 @@ el_val_t __http_response(el_val_t status, el_val_t headers_json, el_val_t body)
|
||||
return http_response(status, headers_json, body);
|
||||
}
|
||||
|
||||
/* ── HTTP SSE — Server-Sent Events streaming ─────────────────────────────── */
|
||||
/*
|
||||
* Thread-local file descriptor stashed by http_worker_v2 before calling the
|
||||
* El handler. El SSE builtins read this to get the raw socket fd.
|
||||
*
|
||||
* Lifecycle:
|
||||
* http_worker_v2 sets _tl_http_conn_fd = fd (via el_seed_set_http_conn_fd)
|
||||
* El handler calls __http_conn_fd() → receives that fd
|
||||
* El handler calls __http_sse_open(fd) → sends SSE headers, keeps fd open
|
||||
* El handler calls __http_sse_send(fd, data) → writes "data: ...\n\n"
|
||||
* El handler calls __http_sse_close(fd) → closes the fd
|
||||
* El handler returns "__sse__" sentinel → http_worker_v2 does NOT close fd
|
||||
*
|
||||
* The -1 value means no current connection (guard against misuse outside
|
||||
* a handler context).
|
||||
*/
|
||||
static __thread int _tl_http_conn_fd = -1;
|
||||
|
||||
/* Called by el_runtime.c's http_worker_v2 — not part of the El ABI. */
|
||||
void el_seed_set_http_conn_fd(int fd) {
|
||||
_tl_http_conn_fd = fd;
|
||||
}
|
||||
|
||||
/* __http_conn_fd() — returns the raw fd for the current HTTP connection.
|
||||
* Valid only inside an http_serve_v2 handler before it returns. */
|
||||
el_val_t __http_conn_fd(void) {
|
||||
return EL_INT(_tl_http_conn_fd);
|
||||
}
|
||||
|
||||
/* __http_sse_open(fd) — sends SSE response headers on fd, keeping it open.
|
||||
* Returns 1 on success, 0 on write failure. */
|
||||
el_val_t __http_sse_open(el_val_t conn_id) {
|
||||
int fd = (int)(int64_t)conn_id;
|
||||
if (fd < 0) return 0;
|
||||
static const char sse_headers[] =
|
||||
"HTTP/1.1 200 OK\r\n"
|
||||
"Content-Type: text/event-stream\r\n"
|
||||
"Cache-Control: no-cache\r\n"
|
||||
"Connection: keep-alive\r\n"
|
||||
"Access-Control-Allow-Origin: *\r\n"
|
||||
"\r\n";
|
||||
size_t n = sizeof(sse_headers) - 1; /* exclude NUL */
|
||||
size_t sent = 0;
|
||||
while (sent < n) {
|
||||
ssize_t w = write(fd, sse_headers + sent, n - sent);
|
||||
if (w <= 0) return 0;
|
||||
sent += (size_t)w;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* __http_sse_send(fd, data) — writes one SSE event frame: "data: <data>\n\n".
|
||||
* data must not contain newlines. Returns 1 on success, 0 on client disconnect. */
|
||||
el_val_t __http_sse_send(el_val_t conn_id, el_val_t data) {
|
||||
int fd = (int)(int64_t)conn_id;
|
||||
if (fd < 0) return 0;
|
||||
const char* s = EL_CSTR(data);
|
||||
if (!s) s = "";
|
||||
/* Build "data: <s>\n\n" in a single buffer for one write call. */
|
||||
size_t prefix_len = 6; /* "data: " */
|
||||
size_t slen = strlen(s);
|
||||
size_t total = prefix_len + slen + 2; /* + "\n\n" */
|
||||
char* buf = malloc(total + 1);
|
||||
if (!buf) return 0;
|
||||
memcpy(buf, "data: ", 6);
|
||||
memcpy(buf + 6, s, slen);
|
||||
buf[6 + slen] = '\n';
|
||||
buf[6 + slen + 1] = '\n';
|
||||
buf[total] = '\0';
|
||||
size_t sent = 0;
|
||||
int ok = 1;
|
||||
while (sent < total) {
|
||||
ssize_t w = write(fd, buf + sent, total - sent);
|
||||
if (w <= 0) { ok = 0; break; }
|
||||
sent += (size_t)w;
|
||||
}
|
||||
free(buf);
|
||||
return ok ? 1 : 0;
|
||||
}
|
||||
|
||||
/* __http_sse_close(fd) — closes the SSE connection fd. */
|
||||
el_val_t __http_sse_close(el_val_t conn_id) {
|
||||
int fd = (int)(int64_t)conn_id;
|
||||
if (fd < 0) return 0;
|
||||
close(fd);
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* ── Threading ───────────────────────────────────────────────────────────── */
|
||||
/*
|
||||
* Design:
|
||||
|
||||
Reference in New Issue
Block a user