# HG changeset patch # User MrJuneJune # Date 1785887351 25200 # Node ID 609d3c6aff4effe30b78f058d85d74d821647085 # Parent 30c2196d03d4f118334ff90812f23fb55dfd7ddd [seobeo] Add persistent SSE streams Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> diff -r 30c2196d03d4 -r 609d3c6aff4e .claude/skills/zenbu-seobeo-networking/SKILL.md --- a/.claude/skills/zenbu-seobeo-networking/SKILL.md Tue Aug 04 16:49:01 2026 -0700 +++ b/.claude/skills/zenbu-seobeo-networking/SKILL.md Tue Aug 04 16:49:11 2026 -0700 @@ -64,6 +64,27 @@ Seobeo_WebSocket_Server_Register("/chat", Chat_Handler, NULL); ``` +Server-Sent Events: + +```c +void Events( + Seobeo_SSE_Stream *p_stream, + Seobeo_Request_Entry *p_request, + Dowa_Arena *p_arena) +{ + Seobeo_SSE_Send_Data(p_stream, "ready"); +} + +Seobeo_Router_Register_SSE("/events", Events); +``` + +SSE handlers run once during connection setup and must return promptly. The +stream may be retained for later thread-safe sends only through balanced +`Seobeo_SSE_Retain` / `Seobeo_SSE_Release` calls; request/response arenas and +request-map pointers expire when the handler returns. Check +`Seobeo_SSE_Is_Open`, handle bounded backpressure, and use comments for +heartbeats. + Background work: ```c @@ -98,6 +119,7 @@ bazel test //seobeo:seobeo_client_test bazel test //seobeo:seobeo_websocket_test bazel test //seobeo:seobeo_websocket_server_test +bazel test //seobeo/tests:seobeo_sse_test ``` For API changes, also build downstream users: diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/BUILD --- a/seobeo/BUILD Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/BUILD Tue Aug 04 16:49:11 2026 -0700 @@ -36,6 +36,7 @@ name = "seobeo_min_macos", srcs = [ "s_network.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "os/s_macos_edge.c", @@ -55,6 +56,7 @@ name = "seobeo_min_linux", srcs = [ "s_network.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "os/s_linux_edge.c", @@ -86,6 +88,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "os/s_macos_edge.c", @@ -106,6 +109,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "os/s_linux_edge.c", @@ -137,6 +141,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_websocket_common.c", @@ -160,6 +165,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_websocket_common.c", @@ -193,6 +199,7 @@ name = "seobeo_tcp_client_macos", srcs = [ "s_network.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", @@ -214,6 +221,7 @@ name = "seobeo_tcp_client_linux", srcs = [ "s_network.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", @@ -246,6 +254,7 @@ name = "seobeo_tcp_client_ws_macos", srcs = [ "s_network.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", @@ -269,6 +278,7 @@ name = "seobeo_tcp_client_ws_linux", srcs = [ "s_network.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", @@ -304,6 +314,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", @@ -331,6 +342,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", @@ -386,6 +398,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_websocket_common.c", @@ -409,6 +422,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_websocket_common.c", @@ -443,6 +457,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", @@ -470,6 +485,7 @@ srcs = [ "s_network.c", "s_web.c", + "s_sse.c", "s_logging.c", "s_ssl.c", "s_http_client.c", diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/README.md --- a/seobeo/README.md Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/README.md Tue Aug 04 16:49:11 2026 -0700 @@ -8,6 +8,7 @@ - SSL/TLS support - Async networking with libuv - Joinable/detached tasks and bounded worker pools +- Server-Sent Events routes and event framing - Snapshot testing utilities ## Files @@ -18,6 +19,7 @@ | `seobeo_internal.h` | Internal declarations | | `s_http_client.c` | HTTP client implementation | | `s_network.c` | Network utilities | +| `s_sse.c` | Server-Sent Events framing and stream lifecycle | | `s_ssl.c` | SSL/TLS handling | | `s_logging.c` | Logging utilities | | `s_worker.c` | Thread and worker-pool implementation | @@ -88,6 +90,42 @@ discarded by a non-draining shutdown. User callbacks never run while the pool mutex is held. +## Server-Sent Events + +Register an SSE route separately from ordinary response and stream handlers: + +```c +static void Build_Events( + Seobeo_SSE_Stream *p_stream, + Seobeo_Request_Entry *p_request, + Dowa_Arena *p_arena) +{ + (void)p_request; + (void)p_arena; + Seobeo_SSE_Event event = { + .event = "build", + .id = "42", + .data = "started\ncompiling", + .retry_ms = 2000, + }; + Seobeo_SSE_Send(p_stream, &event); + Seobeo_SSE_Send_Comment(p_stream, "heartbeat"); +} + +Seobeo_Router_Register_SSE("/events/builds", Build_Events); +``` + +SSE handlers run once when a client connects and must return promptly so edge +workers remain available. Call `Seobeo_SSE_Retain` before storing a stream for +later work and pair it with `Seobeo_SSE_Release`; use +`Seobeo_SSE_Is_Open` before sending. Do not retain the request map or arena +after the handler returns. A return value of `1` means the record was retained +behind socket backpressure and `Seobeo_SSE_Flush` should be retried later. +`SEOBEO_SSE_BACKPRESSURE` means the bounded pending queue is full and the new +record was not accepted. Events are bounded by the handle write-buffer +capacity so one record is queued atomically; larger records return +`SEOBEO_SSE_EVENT_TOO_LARGE`. + ## Dependencies - libuv (via //third_party/libuv) diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/examples/BUILD --- a/seobeo/examples/BUILD Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/examples/BUILD Tue Aug 04 16:49:11 2026 -0700 @@ -1,6 +1,13 @@ load("@rules_cc//cc:cc_binary.bzl", "cc_binary") cc_binary( + name = "sse_server_example", + srcs = ["sse_server_example.c"], + deps = ["//seobeo:seobeo"], + visibility = ["//visibility:public"], +) + +cc_binary( name = "websocket_server_example", srcs = ["websocket_server_example.c"], deps = ["//seobeo:seobeo"], diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/examples/sse_server_example.c --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/seobeo/examples/sse_server_example.c Tue Aug 04 16:49:11 2026 -0700 @@ -0,0 +1,67 @@ +#include "seobeo/seobeo.h" + +#include + +static void Send_Later(void *p_context) +{ + Seobeo_SSE_Stream *p_stream = p_context; + usleep(100000); + if (Seobeo_SSE_Is_Open(p_stream)) + Seobeo_SSE_Send_Data(p_stream, "later"); +} + +static void Release_Stream(void *p_context) +{ + Seobeo_SSE_Release(p_context); +} + +static void Events( + Seobeo_SSE_Stream *p_stream, + Seobeo_Request_Entry *p_request, + Dowa_Arena *p_arena) +{ + (void)p_request; + (void)p_arena; + Seobeo_SSE_Send_Data(p_stream, "connected"); + if (Seobeo_SSE_Retain(p_stream)) + { + Seobeo_Worker_Result result = Seobeo_Thread_Start_Detached( + Send_Later, + p_stream, + Release_Stream); + if (result != SEOBEO_WORKER_OK) + Seobeo_SSE_Release(p_stream); + } +} + +static Seobeo_Request_Entry *Health( + Seobeo_Request_Entry *p_request, + Dowa_Arena *p_arena) +{ + (void)p_request; + Seobeo_Request_Entry *p_response = NULL; + Dowa_HashMap_Push_Arena(p_response, "status", "200", p_arena); + Dowa_HashMap_Push_Arena( + p_response, + "content-type", + "text/plain", + p_arena); + Dowa_HashMap_Push_Arena(p_response, "body", "ok", p_arena); + return p_response; +} + +int main(int argc, char **argv) +{ + const char *port = argc > 1 ? argv[1] : "18081"; + Seobeo_Router_Init(); + Seobeo_Router_Register_SSE("/events", Events); + Seobeo_Router_Register("GET", "/health", Health); + int result = Seobeo_Web_Server_Start_On( + "127.0.0.1", + "seobeo/examples", + port, + SEOBEO_MODE_EDGE, + 1); + Seobeo_Router_Destroy(); + return result; +} diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/os/s_linux_edge.c --- a/seobeo/os/s_linux_edge.c Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/os/s_linux_edge.c Tue Aug 04 16:49:11 2026 -0700 @@ -74,7 +74,7 @@ configure_keep_alive(p_cli_handle->socket); struct epoll_event client_ev = { - .events = EPOLLIN | EPOLLET, + .events = EPOLLIN | EPOLLET | EPOLLRDHUP, .data.ptr = p_cli_handle }; @@ -90,20 +90,46 @@ { Seobeo_Handle *p_client_handle = p_handle; - // Connection error or hangup - clean up - if (events[i].events & (EPOLLERR | EPOLLHUP)) + boolean peer_closed = + (events[i].events & (EPOLLHUP | EPOLLRDHUP)) != 0; + // Preserve readable bytes on a half-close so the final request can + // still receive its response. + if ((events[i].events & EPOLLERR) || + (peer_closed && !(events[i].events & EPOLLIN))) { epoll_ctl(epfd, EPOLL_CTL_DEL, p_client_handle->socket, NULL); Seobeo_Handle_Destroy(p_client_handle); continue; } + if (p_client_handle->is_sse) + { + int32 read_result = Seobeo_Handle_Read(p_client_handle); + if (read_result < 0) + { + epoll_ctl(epfd, EPOLL_CTL_DEL, p_client_handle->socket, NULL); + Seobeo_Handle_Destroy(p_client_handle); + } + else if (p_client_handle->read_buffer_len) + { + Seobeo_Handle_Consume( + p_client_handle, + p_client_handle->read_buffer_len); + } + continue; + } + // Handle requests (loop for pipelined requests) boolean keep_alive = TRUE; while (keep_alive) { keep_alive = Seobeo_Web_ClientHandle_Request(p_client_handle, args->cache, TRUE); + if (p_client_handle->is_sse) + break; + if (peer_closed) + keep_alive = FALSE; + // No more data in buffer, wait for next epoll event if (keep_alive && p_client_handle->read_buffer_len == 0) break; diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/os/s_macos_edge.c --- a/seobeo/os/s_macos_edge.c Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/os/s_macos_edge.c Tue Aug 04 16:49:11 2026 -0700 @@ -51,6 +51,31 @@ kevent(kq, &client_kev, 1, NULL, 0, NULL); } } else { + if (h->is_sse) + { + if (evlist[i].flags & EV_EOF) + { + Seobeo_Handle_Destroy(h); + continue; + } + int32 read_result = Seobeo_Handle_Read(h); + if (read_result < 0) + { + Seobeo_Handle_Destroy(h); + continue; + } + if (h->read_buffer_len) + Seobeo_Handle_Consume(h, h->read_buffer_len); + struct kevent sse_kev = { + .ident = h->socket, + .filter = EVFILT_READ, + .flags = EV_ADD | EV_ONESHOT, + .udata = h + }; + kevent(kq, &sse_kev, 1, NULL, 0, NULL); + continue; + } + // Remove from kqueue first struct kevent del_kev = { .ident = h->socket, @@ -59,9 +84,22 @@ }; kevent(kq, &del_kev, 1, NULL, 0, NULL); - // Handle request (this function destroys the handle internally) - // TODO: Add keep-alive support for macOS like Linux version + // macOS handles ordinary requests once, while SSE takes ownership. Seobeo_Web_ClientHandle_Request(h, args->cache, FALSE); + if (h->is_sse) + { + struct kevent sse_kev = { + .ident = h->socket, + .filter = EVFILT_READ, + .flags = EV_ADD | EV_ONESHOT, + .udata = h + }; + kevent(kq, &sse_kev, 1, NULL, 0, NULL); + } + else + { + Seobeo_Handle_Destroy(h); + } } } } diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/s_network.c --- a/seobeo/s_network.c Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/s_network.c Tue Aug 04 16:49:11 2026 -0700 @@ -146,6 +146,7 @@ p_handle->write_buffer_len = 0; p_handle->destroyed = FALSE; + p_handle->is_sse = FALSE; if (!p_handle->host || !p_handle->port || @@ -224,6 +225,7 @@ p_handle->write_buffer_len = 0; p_handle->destroyed = FALSE; + p_handle->is_sse = FALSE; return p_handle; } @@ -277,6 +279,7 @@ p_client_handle->text_copy = NULL; p_client_handle->file_name = NULL; p_client_handle->destroyed = FALSE; + p_client_handle->is_sse = FALSE; return p_client_handle; } @@ -292,6 +295,8 @@ return; } + Seobeo_SSE_Server_Detach_Handle(p_handle); + if (p_handle->host) Dowa_Free(p_handle->host); if (p_handle->port) Dowa_Free(p_handle->port); @@ -311,6 +316,8 @@ int32 Seobeo_Handle_Flush(Seobeo_Handle *p_handle) { + if (!p_handle || !p_handle->write_buffer) + return -1; uint32 total = p_handle->write_buffer_len; uint32 sent = 0; @@ -321,8 +328,30 @@ if (p_handle->ssl) { int n = Seobeo_SSL_Write(p_handle, p_handle->write_buffer + sent, total - sent); - if (n < 0) return -1; - if (n == 0) return 0; // would block + if (n < 0) + { + if (sent) + { + memmove( + p_handle->write_buffer, + p_handle->write_buffer + sent, + total - sent); + p_handle->write_buffer_len = total - sent; + } + return -1; + } + if (n == 0) + { + if (sent) + { + memmove( + p_handle->write_buffer, + p_handle->write_buffer + sent, + total - sent); + p_handle->write_buffer_len = total - sent; + } + return 1; + } sent += (uint32)n; }else { @@ -334,7 +363,26 @@ ); if (n < 0) { if (errno == EINTR) continue; - if (errno == EAGAIN) return 1; + if (errno == EAGAIN || errno == EWOULDBLOCK) + { + if (sent) + { + memmove( + p_handle->write_buffer, + p_handle->write_buffer + sent, + total - sent); + p_handle->write_buffer_len = total - sent; + } + return 1; + } + if (sent) + { + memmove( + p_handle->write_buffer, + p_handle->write_buffer + sent, + total - sent); + p_handle->write_buffer_len = total - sent; + } return -1; } sent += (uint32)n; diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/s_sse.c --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/seobeo/s_sse.c Tue Aug 04 16:49:11 2026 -0700 @@ -0,0 +1,450 @@ +#include "seobeo/seobeo.h" + +#include +#include +#include + +static const char *SEOBEO_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" + "X-Accel-Buffering: no\r\n" + "\r\n"; + +static pthread_mutex_t g_sse_mutex = PTHREAD_MUTEX_INITIALIZER; +static Seobeo_SSE_Stream **g_sse_streams = NULL; + +typedef struct { + char *data; + uint32 size; +} Seobeo_SSE_Pending_Frame; + +#define SEOBEO_SSE_MAX_PENDING 32 + +static boolean sse_is_open_unlocked(const Seobeo_SSE_Stream *p_stream) +{ + return p_stream && + p_stream->started && + !p_stream->closed && + p_stream->p_handle && + p_stream->p_handle->socket >= 0 && + !atomic_load(&p_stream->p_handle->destroyed); +} + +static void sse_clear_pending_unlocked(Seobeo_SSE_Stream *p_stream) +{ + Seobeo_SSE_Pending_Frame *p_frames = p_stream->p_pending_frames; + size_t count = Dowa_Array_Length(p_frames); + for (size_t i = p_stream->pending_offset; i < count; i++) + Dowa_Free(p_frames[i].data); + Dowa_Array_Free(p_frames); + p_stream->p_pending_frames = NULL; + p_stream->pending_offset = 0; +} + +static void sse_release_unlocked(Seobeo_SSE_Stream *p_stream) +{ + if (!p_stream) + return; + if (atomic_fetch_sub(&p_stream->ref_count, 1) != 1) + return; + sse_clear_pending_unlocked(p_stream); + Dowa_Free(p_stream->path); + Dowa_Free(p_stream); +} + +static int32 sse_drain_pending_unlocked(Seobeo_SSE_Stream *p_stream) +{ + if (!sse_is_open_unlocked(p_stream)) + return -1; + int32 flush_result = Seobeo_Handle_Flush(p_stream->p_handle); + if (flush_result != 0) + return flush_result; + + Seobeo_SSE_Pending_Frame *p_frames = p_stream->p_pending_frames; + size_t count = Dowa_Array_Length(p_frames); + while (p_stream->pending_offset < count) + { + Seobeo_SSE_Pending_Frame *p_frame = + &p_frames[p_stream->pending_offset]; + int32 queue_result = Seobeo_Handle_Queue( + p_stream->p_handle, + (const uint8 *)p_frame->data, + p_frame->size); + if (queue_result != 0) + return queue_result; + Dowa_Free(p_frame->data); + p_stream->pending_offset++; + flush_result = Seobeo_Handle_Flush(p_stream->p_handle); + if (flush_result != 0) + return flush_result; + } + + Dowa_Array_Free(p_frames); + p_stream->p_pending_frames = NULL; + p_stream->pending_offset = 0; + return 0; +} + +static int32 sse_enqueue_frame_unlocked( + Seobeo_SSE_Stream *p_stream, + char *frame, + uint32 frame_size) +{ + Seobeo_SSE_Pending_Frame *p_frames = p_stream->p_pending_frames; + size_t pending = Dowa_Array_Length(p_frames) - p_stream->pending_offset; + if (pending >= SEOBEO_SSE_MAX_PENDING) + return SEOBEO_SSE_BACKPRESSURE; + Seobeo_SSE_Pending_Frame pending_frame = { + .data = frame, + .size = frame_size, + }; + Dowa_Array_Push(p_frames, pending_frame); + p_stream->p_pending_frames = p_frames; + return sse_drain_pending_unlocked(p_stream); +} + +static boolean sse_field_is_safe(const char *value) +{ + return !value || (strchr(value, '\r') == NULL && strchr(value, '\n') == NULL); +} + +static size_t sse_multiline_size(const char *prefix, const char *value) +{ + const char *line = value ? value : ""; + size_t prefix_length = strlen(prefix); + size_t total = 0; + + while (TRUE) + { + const char *end = line; + while (*end && *end != '\r' && *end != '\n') + end++; + size_t line_length = (size_t)(end - line); + total += prefix_length + (line_length ? 1 : 0) + line_length + 1; + if (*end == '\0') + break; + if (*end == '\r' && end[1] == '\n') + end++; + line = end + 1; + } + return total; +} + +static char *sse_append_multiline( + char *output, + const char *prefix, + const char *value) +{ + const char *line = value ? value : ""; + size_t prefix_length = strlen(prefix); + + while (TRUE) + { + const char *end = line; + while (*end && *end != '\r' && *end != '\n') + end++; + size_t line_length = (size_t)(end - line); + memcpy(output, prefix, prefix_length); + output += prefix_length; + if (line_length) + *output++ = ' '; + memcpy(output, line, line_length); + output += line_length; + *output++ = '\n'; + if (*end == '\0') + break; + if (*end == '\r' && end[1] == '\n') + end++; + line = end + 1; + } + return output; +} + +static int32 sse_send_frame( + Seobeo_SSE_Stream *p_stream, + const char *frame, + size_t frame_size) +{ + if (!sse_is_open_unlocked(p_stream) || !frame) + return -1; + if (frame_size > p_stream->p_handle->write_buffer_capacity) + return SEOBEO_SSE_EVENT_TOO_LARGE; + + int32 flush_result = Seobeo_Handle_Flush(p_stream->p_handle); + if (flush_result != 0) + return flush_result; + + int32 queue_result = Seobeo_Handle_Queue( + p_stream->p_handle, + (const uint8 *)frame, + (uint32)frame_size); + if (queue_result != 0) + return queue_result; + return Seobeo_Handle_Flush(p_stream->p_handle); +} + +int32 Seobeo_SSE_Start( + Seobeo_SSE_Stream *p_stream, + Seobeo_Handle *p_handle) +{ + if (!p_stream || !p_handle || atomic_load(&p_handle->destroyed)) + return -1; + p_stream->p_handle = p_handle; + p_stream->started = TRUE; + p_stream->closed = FALSE; + return sse_send_frame( + p_stream, + SEOBEO_SSE_HEADERS, + strlen(SEOBEO_SSE_HEADERS)); +} + +static int32 sse_send_event_unlocked( + Seobeo_SSE_Stream *p_stream, + const Seobeo_SSE_Event *p_event) +{ + if (!sse_is_open_unlocked(p_stream)) + return -1; + if (!p_event || !sse_field_is_safe(p_event->event) || + !sse_field_is_safe(p_event->id) || + p_event->retry_ms < SEOBEO_SSE_RETRY_NONE) + { + return SEOBEO_SSE_INVALID_EVENT; + } + + size_t frame_size = 1 + sse_multiline_size("data:", p_event->data); + if (p_event->id) + frame_size += strlen("id: ") + strlen(p_event->id) + 1; + if (p_event->event) + frame_size += strlen("event: ") + strlen(p_event->event) + 1; + + char retry[32] = {0}; + if (p_event->retry_ms != SEOBEO_SSE_RETRY_NONE) + { + int written = snprintf( + retry, + sizeof(retry), + "retry: %d\n", + p_event->retry_ms); + if (written < 0 || (size_t)written >= sizeof(retry)) + return SEOBEO_SSE_INVALID_EVENT; + frame_size += (size_t)written; + } + + if (frame_size > p_stream->p_handle->write_buffer_capacity) + return SEOBEO_SSE_EVENT_TOO_LARGE; + char *frame = malloc(frame_size); + if (!frame) + return -1; + + char *output = frame; + if (p_event->id) + output += sprintf(output, "id: %s\n", p_event->id); + if (p_event->event) + output += sprintf(output, "event: %s\n", p_event->event); + if (retry[0]) + { + size_t retry_length = strlen(retry); + memcpy(output, retry, retry_length); + output += retry_length; + } + output = sse_append_multiline(output, "data:", p_event->data); + *output++ = '\n'; + + int32 result = sse_enqueue_frame_unlocked( + p_stream, + frame, + (uint32)(output - frame)); + if (result == SEOBEO_SSE_BACKPRESSURE) + Dowa_Free(frame); + return result; +} + +int32 Seobeo_SSE_Send( + Seobeo_SSE_Stream *p_stream, + const Seobeo_SSE_Event *p_event) +{ + pthread_mutex_lock(&g_sse_mutex); + int32 result = sse_send_event_unlocked(p_stream, p_event); + pthread_mutex_unlock(&g_sse_mutex); + return result; +} + +int32 Seobeo_SSE_Send_Data( + Seobeo_SSE_Stream *p_stream, + const char *data) +{ + Seobeo_SSE_Event event = { + .data = data, + .retry_ms = SEOBEO_SSE_RETRY_NONE, + }; + return Seobeo_SSE_Send(p_stream, &event); +} + +int32 Seobeo_SSE_Send_Comment( + Seobeo_SSE_Stream *p_stream, + const char *comment) +{ + pthread_mutex_lock(&g_sse_mutex); + if (!sse_is_open_unlocked(p_stream)) + { + pthread_mutex_unlock(&g_sse_mutex); + return -1; + } + size_t frame_size = sse_multiline_size(":", comment) + 1; + if (frame_size > p_stream->p_handle->write_buffer_capacity) + { + pthread_mutex_unlock(&g_sse_mutex); + return SEOBEO_SSE_EVENT_TOO_LARGE; + } + char *frame = malloc(frame_size); + if (!frame) + { + pthread_mutex_unlock(&g_sse_mutex); + return -1; + } + char *output = sse_append_multiline(frame, ":", comment); + *output++ = '\n'; + int32 result = sse_enqueue_frame_unlocked( + p_stream, + frame, + (uint32)(output - frame)); + if (result == SEOBEO_SSE_BACKPRESSURE) + Dowa_Free(frame); + pthread_mutex_unlock(&g_sse_mutex); + return result; +} + +int32 Seobeo_SSE_Flush(Seobeo_SSE_Stream *p_stream) +{ + pthread_mutex_lock(&g_sse_mutex); + int32 result = sse_is_open_unlocked(p_stream) + ? sse_drain_pending_unlocked(p_stream) + : -1; + pthread_mutex_unlock(&g_sse_mutex); + return result; +} + +boolean Seobeo_SSE_Is_Open(const Seobeo_SSE_Stream *p_stream) +{ + pthread_mutex_lock(&g_sse_mutex); + boolean open = sse_is_open_unlocked(p_stream); + pthread_mutex_unlock(&g_sse_mutex); + return open; +} + +boolean Seobeo_SSE_Retain(Seobeo_SSE_Stream *p_stream) +{ + if (!p_stream) + return FALSE; + pthread_mutex_lock(&g_sse_mutex); + unsigned int references = atomic_load(&p_stream->ref_count); + if (references == 0) + { + pthread_mutex_unlock(&g_sse_mutex); + return FALSE; + } + atomic_fetch_add(&p_stream->ref_count, 1); + pthread_mutex_unlock(&g_sse_mutex); + return TRUE; +} + +void Seobeo_SSE_Release(Seobeo_SSE_Stream *p_stream) +{ + if (!p_stream) + return; + pthread_mutex_lock(&g_sse_mutex); + sse_release_unlocked(p_stream); + pthread_mutex_unlock(&g_sse_mutex); +} + +void Seobeo_SSE_Close(Seobeo_SSE_Stream *p_stream) +{ + if (!p_stream) + return; + pthread_mutex_lock(&g_sse_mutex); + if (p_stream->started && !p_stream->closed && p_stream->p_handle) + { + sse_drain_pending_unlocked(p_stream); + if (p_stream->managed) + shutdown(p_stream->p_handle->socket, SHUT_RDWR); + } + sse_clear_pending_unlocked(p_stream); + p_stream->closed = TRUE; + pthread_mutex_unlock(&g_sse_mutex); +} + +Seobeo_SSE_Stream *Seobeo_SSE_Server_Attach( + Seobeo_Handle *p_handle, + const char *path) +{ + if (!p_handle || !path) + return NULL; + Seobeo_SSE_Stream *p_stream = malloc(sizeof(*p_stream)); + if (!p_stream) + return NULL; + memset(p_stream, 0, sizeof(*p_stream)); + p_stream->path = strdup(path); + if (!p_stream->path) + { + Dowa_Free(p_stream); + return NULL; + } + atomic_init(&p_stream->ref_count, 1); + if (Seobeo_SSE_Start(p_stream, p_handle) < 0) + { + Dowa_Free(p_stream->path); + Dowa_Free(p_stream); + return NULL; + } + p_stream->managed = TRUE; + p_handle->is_sse = TRUE; + + pthread_mutex_lock(&g_sse_mutex); + Dowa_Array_Push(g_sse_streams, p_stream); + pthread_mutex_unlock(&g_sse_mutex); + return p_stream; +} + +void Seobeo_SSE_Server_Detach_Handle(Seobeo_Handle *p_handle) +{ + if (!p_handle) + return; + p_handle->is_sse = FALSE; + pthread_mutex_lock(&g_sse_mutex); + size_t count = Dowa_Array_Length(g_sse_streams); + for (size_t i = 0; i < count; i++) + { + Seobeo_SSE_Stream *p_stream = g_sse_streams[i]; + if (p_stream->p_handle != p_handle) + continue; + p_stream->closed = TRUE; + p_stream->p_handle = NULL; + sse_clear_pending_unlocked(p_stream); + g_sse_streams[i] = Dowa_Array_Pop(g_sse_streams); + sse_release_unlocked(p_stream); + break; + } + pthread_mutex_unlock(&g_sse_mutex); +} + +void Seobeo_SSE_Server_Destroy(void) +{ + pthread_mutex_lock(&g_sse_mutex); + size_t count = Dowa_Array_Length(g_sse_streams); + for (size_t i = 0; i < count; i++) + { + Seobeo_SSE_Stream *p_stream = g_sse_streams[i]; + if (!p_stream) + continue; + p_stream->closed = TRUE; + if (p_stream->p_handle) + p_stream->p_handle->is_sse = FALSE; + p_stream->p_handle = NULL; + sse_clear_pending_unlocked(p_stream); + sse_release_unlocked(p_stream); + } + Dowa_Array_Free(g_sse_streams); + pthread_mutex_unlock(&g_sse_mutex); +} diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/s_web.c --- a/seobeo/s_web.c Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/s_web.c Tue Aug 04 16:49:11 2026 -0700 @@ -136,7 +136,7 @@ } token = strtok(NULL, ", \t"); } - free(copy); + Dowa_Free(copy); return found; } @@ -243,6 +243,29 @@ #endif // --- Try to match streaming route first --- + Seobeo_SSE_Handler sse_handler = Seobeo_Router_Find_SSE_Handler( + method, + path, + &p_req_map, + p_request_arena); + if (sse_handler != NULL) + { + Seobeo_SSE_Stream *p_stream = Seobeo_SSE_Server_Attach( + p_cli_handle, + path); + if (p_stream) + { + sse_handler(p_stream, p_req_map, p_response_arena); + should_keep_alive = Seobeo_SSE_Is_Open(p_stream); + } + else + { + should_keep_alive = FALSE; + } + goto clean_up_arenas; + } + + // --- Try to match generic streaming route --- Seobeo_Stream_Handler stream_handler = Seobeo_Router_Find_Stream_Handler(method, path, &p_req_map, p_request_arena); if (stream_handler != NULL) { @@ -356,7 +379,7 @@ (const uint8*)file_content, (uint32)body_size); Seobeo_Handle_Flush(p_cli_handle); - free(file_content); + Dowa_Free(file_content); } else { @@ -690,6 +713,7 @@ char *path_pattern; // "/v1/users/:id/posts/:post_id" Seobeo_Route_Handler handler; Seobeo_Stream_Handler stream_handler; // For streaming responses + Seobeo_SSE_Handler sse_handler; // Pre-parsed path segments for efficient matching char **path_segments; // ["v1", "users", ":id", "posts", ":post_id"] @@ -712,6 +736,7 @@ route.path_pattern = strdup(path_pattern); route.handler = handler; route.stream_handler = NULL; + route.sse_handler = NULL; route.path_segments = Dowa_String_Split(path_pattern, "/", strlen(path_pattern), 1, NULL); route.segment_count = Dowa_Array_Length(route.path_segments); route.is_param = (boolean*)malloc(sizeof(boolean) * route.segment_count); @@ -730,6 +755,7 @@ route.path_pattern = strdup(path_pattern); route.handler = NULL; route.stream_handler = handler; + route.sse_handler = NULL; route.path_segments = Dowa_String_Split(path_pattern, "/", strlen(path_pattern), 1, NULL); route.segment_count = Dowa_Array_Length(route.path_segments); route.is_param = (boolean*)malloc(sizeof(boolean) * route.segment_count); @@ -740,6 +766,33 @@ Dowa_Array_Push(g_routes, route); } +void Seobeo_Router_Register_SSE( + const char *path_pattern, + Seobeo_SSE_Handler handler) +{ + Seobeo_Route route = {0}; + + route.method = strdup("GET"); + route.path_pattern = strdup(path_pattern); + route.handler = NULL; + route.stream_handler = NULL; + route.sse_handler = handler; + route.path_segments = Dowa_String_Split( + path_pattern, + "/", + strlen(path_pattern), + 1, + NULL); + route.segment_count = Dowa_Array_Length(route.path_segments); + route.is_param = (boolean*)malloc( + sizeof(boolean) * route.segment_count); + + for (size_t i = 0; i < route.segment_count; i++) + route.is_param[i] = (route.path_segments[i][0] == ':'); + + Dowa_Array_Push(g_routes, route); +} + static boolean match_route_and_extract( Seobeo_Route *route, const char *request_path, @@ -759,20 +812,8 @@ for (size_t i = 0; i < route->segment_count; i++) { - // parameters - if (route->is_param[i]) + if (!route->is_param[i]) { - char *param_name = route->path_segments[i]; // e.g., ":id" - char *param_value = request_segments[i]; // e.g., "123" - - // Should Copy to arena - char *key = Dowa_String_Copy_Arena(param_name, p_arena); - char *value = Dowa_String_Copy_Arena(param_value, p_arena); - Dowa_HashMap_Push_Arena(*pp_request_map, key, value, p_arena); - } - else - { - // Does not match. if (strcmp(route->path_segments[i], request_segments[i]) != 0) { Dowa_Arena_Free(p_temp_arena); @@ -780,7 +821,16 @@ } } } - + + for (size_t i = 0; i < route->segment_count; i++) + { + if (!route->is_param[i]) + continue; + char *key = Dowa_String_Copy_Arena(route->path_segments[i], p_arena); + char *value = Dowa_String_Copy_Arena(request_segments[i], p_arena); + Dowa_HashMap_Push_Arena(*pp_request_map, key, value, p_arena); + } + Dowa_Arena_Free(p_temp_arena); return TRUE; } @@ -835,6 +885,34 @@ return NULL; } +Seobeo_SSE_Handler Seobeo_Router_Find_SSE_Handler( + const char *method, + const char *path, + Seobeo_Request_Entry **pp_request_map, + Dowa_Arena *p_arena) +{ + if (g_routes == NULL || method == NULL || path == NULL) + return NULL; + + size_t route_count = Dowa_Array_Length(g_routes); + for (size_t i = 0; i < route_count; i++) + { + Seobeo_Route *route = &g_routes[i]; + if (strcmp(route->method, method) != 0) + continue; + if (route->sse_handler && + match_route_and_extract( + route, + path, + pp_request_map, + p_arena)) + { + return route->sse_handler; + } + } + return NULL; +} + void Seobeo_Router_Send_Response( Seobeo_Handle *p_handle, Seobeo_Request_Entry *p_response_map, @@ -958,6 +1036,7 @@ void Seobeo_Router_Destroy() { + Seobeo_SSE_Server_Destroy(); if (g_routes == NULL) return; @@ -965,10 +1044,10 @@ for (size_t i = 0; i < route_count; i++) { Seobeo_Route *route = &g_routes[i]; - if (route->method) free(route->method); - if (route->path_pattern) free(route->path_pattern); + if (route->method) Dowa_Free(route->method); + if (route->path_pattern) Dowa_Free(route->path_pattern); if (route->path_segments) Dowa_Array_Free(route->path_segments); - if (route->is_param) free(route->is_param); + if (route->is_param) Dowa_Free(route->is_param); } Dowa_Array_Free(g_routes); g_routes = NULL; diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/seobeo.h --- a/seobeo/seobeo.h Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/seobeo.h Tue Aug 04 16:49:11 2026 -0700 @@ -27,6 +27,10 @@ #define INITIAL_BUFFER_CAPACITY 4096 +#define SEOBEO_SSE_RETRY_NONE -1 +#define SEOBEO_SSE_EVENT_TOO_LARGE -2 +#define SEOBEO_SSE_INVALID_EVENT -3 +#define SEOBEO_SSE_BACKPRESSURE -4 // HTTP STATUS CODE #define HTTP_OK 200 @@ -342,6 +346,8 @@ extern void Seobeo_Router_Register(const char *method, const char *path_pattern, Seobeo_Route_Handler handler); /* Register a streaming route handler. Handler receives client handle for direct streaming. */ extern void Seobeo_Router_Register_Stream(const char *method, const char *path_pattern, Seobeo_Stream_Handler handler); +/* Register a GET route that owns an SSE stream until its handler returns. */ +extern void Seobeo_Router_Register_SSE(const char *path_pattern, Seobeo_SSE_Handler handler); /* Clean up router resources */ extern void Seobeo_Router_Destroy(); /* Find matching route handler (internal use) */ @@ -353,16 +359,36 @@ extern char *Seobeo_Web_LoadFile(const char *file_path, size_t *p_file_size); /* Being a proxy and keeping the client open */ extern Seobeo_Stream_Handler Seobeo_Router_Find_Stream_Handler(const char *method, const char *path, Seobeo_Request_Entry **pp_request_map, Dowa_Arena *p_arena); +extern Seobeo_SSE_Handler Seobeo_Router_Find_SSE_Handler(const char *method, const char *path, Seobeo_Request_Entry **pp_request_map, Dowa_Arena *p_arena); + +/* Start an SSE response and send its required HTTP headers. */ +extern int32 Seobeo_SSE_Start(Seobeo_SSE_Stream *p_stream, Seobeo_Handle *p_handle); +/* Send one SSE record. retry_ms may be SEOBEO_SSE_RETRY_NONE. */ +extern int32 Seobeo_SSE_Send(Seobeo_SSE_Stream *p_stream, const Seobeo_SSE_Event *p_event); +/* Send a data-only SSE record. */ +extern int32 Seobeo_SSE_Send_Data(Seobeo_SSE_Stream *p_stream, const char *data); +/* Send a comment/heartbeat record. */ +extern int32 Seobeo_SSE_Send_Comment(Seobeo_SSE_Stream *p_stream, const char *comment); +/* Flush queued SSE bytes. Returns 1 when the socket would block. */ +extern int32 Seobeo_SSE_Flush(Seobeo_SSE_Stream *p_stream); +/* Return TRUE while the stream can still accept records. */ +extern boolean Seobeo_SSE_Is_Open(const Seobeo_SSE_Stream *p_stream); +/* Retain a stream before storing it beyond the connection callback. */ +extern boolean Seobeo_SSE_Retain(Seobeo_SSE_Stream *p_stream); +/* Release a previously retained stream. */ +extern void Seobeo_SSE_Release(Seobeo_SSE_Stream *p_stream); +/* Stop accepting records. The router retains ownership of the handle. */ +extern void Seobeo_SSE_Close(Seobeo_SSE_Stream *p_stream); // --- Helper functions --- // /* Destroy handle. It will handle all NULL poointers. */ extern void Seobeo_Handle_Destroy(Seobeo_Handle *p_handle); /* Write to socket from write_buffer in the handle. */ -extern int Seobeo_Handle_Flush(Seobeo_Handle *p_handle); +extern int32 Seobeo_Handle_Flush(Seobeo_Handle *p_handle); /* Write to socket with given data source, if the data source is bigger than the write buffer for handle then we just directly write from the data source. */ -extern int Seobeo_Handle_Queue(Seobeo_Handle *p_handle, const uint8_t *data, uint32_t data_size); +extern int32 Seobeo_Handle_Queue(Seobeo_Handle *p_handle, const uint8 *data, uint32 data_size); /* Read to socket from read_buffer in the handle. */ -extern int Seobeo_Handle_Read(Seobeo_Handle *p_handle); +extern int32 Seobeo_Handle_Read(Seobeo_Handle *p_handle); /* Move to read_buffer to front and we already consumed the given amount in the handle. */ extern void Seobeo_Handle_Consume(Seobeo_Handle *p_handle, uint32 consumed); /* Assign IP4 or IP6 to sockaddr. TODO: Maybe create my own struct for this? */ diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/seobeo_internal.h --- a/seobeo/seobeo_internal.h Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/seobeo_internal.h Tue Aug 04 16:49:11 2026 -0700 @@ -52,6 +52,7 @@ // Keep-alive support time_t last_activity; // Timestamp of last request/response boolean keep_alive; // Whether this connection should be kept alive + boolean is_sse; // SSE takeover; edge loop only watches disconnects } Seobeo_Handle; // Cached file entry with content and size (for binary file support) @@ -77,6 +78,24 @@ // HTTP request map type: maps header names to header values typedef Dowa_KV(char*, char*) Seobeo_Request_Entry; +typedef struct { + Seobeo_Handle *p_handle; + char *path; + void *p_pending_frames; + size_t pending_offset; + atomic_uint ref_count; + boolean started; + boolean closed; + boolean managed; +} Seobeo_SSE_Stream; + +typedef struct { + const char *event; + const char *id; + const char *data; + int32 retry_ms; +} Seobeo_SSE_Event; + // --- Router Types --- // // Forward declaration typedef struct Seobeo_Route_Struct Seobeo_Route; @@ -94,6 +113,18 @@ Dowa_Arena *p_arena ); +typedef void (*Seobeo_SSE_Handler)( + Seobeo_SSE_Stream *p_stream, + Seobeo_Request_Entry *p_request_map, + Dowa_Arena *p_arena +); + +extern Seobeo_SSE_Stream *Seobeo_SSE_Server_Attach( + Seobeo_Handle *p_handle, + const char *path); +extern void Seobeo_SSE_Server_Detach_Handle(Seobeo_Handle *p_handle); +extern void Seobeo_SSE_Server_Destroy(void); + // --- Parse Header into Dowa Map ---// extern int Seobeo_Web_Header_Parse(Seobeo_Handle *p_handle, Seobeo_Request_Entry **pp_map, Dowa_Arena *p_arena); diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/tests/BUILD --- a/seobeo/tests/BUILD Tue Aug 04 16:49:01 2026 -0700 +++ b/seobeo/tests/BUILD Tue Aug 04 16:49:11 2026 -0700 @@ -84,6 +84,27 @@ ) cc_test( + name = "seobeo_sse_server_test", + srcs = ["seobeo_sse_server_test.c"], + deps = ["//seobeo:seobeo"], + data = ["//seobeo/examples:sse_server_example"], + args = ["$(location //seobeo/examples:sse_server_example)"], + size = "medium", + timeout = "moderate", + target_compatible_with = ["@platforms//os:linux"], + visibility = ["//visibility:public"], +) + +cc_test( + name = "seobeo_sse_test", + srcs = ["seobeo_sse_test.c"], + deps = ["//seobeo:seobeo"], + size = "small", + timeout = "short", + visibility = ["//visibility:public"], +) + +cc_test( name = "seobeo_server_bind_test", srcs = ["seobeo_server_bind_test.c"], deps = ["//seobeo:seobeo"], diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/tests/seobeo_sse_server_test.c --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/seobeo/tests/seobeo_sse_server_test.c Tue Aug 04 16:49:11 2026 -0700 @@ -0,0 +1,168 @@ +#include "seobeo/seobeo.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +static int reserve_port(void) +{ + int socket_fd = socket(AF_INET, SOCK_STREAM, 0); + assert(socket_fd >= 0); + struct sockaddr_in address = { + .sin_family = AF_INET, + .sin_addr.s_addr = htonl(INADDR_LOOPBACK), + .sin_port = 0, + }; + assert(bind( + socket_fd, + (struct sockaddr *)&address, + sizeof(address)) == 0); + socklen_t length = sizeof(address); + assert(getsockname( + socket_fd, + (struct sockaddr *)&address, + &length) == 0); + int port = ntohs(address.sin_port); + close(socket_fd); + return port; +} + +static int connect_to(int port) +{ + int socket_fd = socket(AF_INET, SOCK_STREAM, 0); + if (socket_fd < 0) + return -1; + struct timeval timeout = { + .tv_sec = 3, + }; + setsockopt( + socket_fd, + SOL_SOCKET, + SO_RCVTIMEO, + &timeout, + sizeof(timeout)); + struct sockaddr_in address = { + .sin_family = AF_INET, + .sin_addr.s_addr = htonl(INADDR_LOOPBACK), + .sin_port = htons((uint16)port), + }; + if (connect( + socket_fd, + (struct sockaddr *)&address, + sizeof(address)) != 0) + { + close(socket_fd); + return -1; + } + return socket_fd; +} + +static void send_request(int socket_fd, const char *path) +{ + char request[512]; + int length = snprintf( + request, + sizeof(request), + "GET %s HTTP/1.1\r\n" + "Host: 127.0.0.1\r\n" + "Connection: keep-alive\r\n" + "\r\n", + path); + assert(length > 0 && (size_t)length < sizeof(request)); + assert(send(socket_fd, request, (size_t)length, 0) == length); +} + +static size_t receive_until( + int socket_fd, + char *buffer, + size_t capacity, + const char *needle) +{ + size_t used = 0; + while (used + 1 < capacity) + { + ssize_t amount = recv( + socket_fd, + buffer + used, + capacity - used - 1, + 0); + assert(amount > 0); + used += (size_t)amount; + buffer[used] = '\0'; + if (strstr(buffer, needle)) + return used; + } + assert(FALSE && "response exceeded test buffer"); + return 0; +} + +static pid_t start_server( + const char *binary, + const char *port) +{ + pid_t pid = fork(); + assert(pid >= 0); + if (pid == 0) + { + execl(binary, binary, port, NULL); + _exit(127); + } + + int numeric_port = atoi(port); + for (int attempt = 0; attempt < 100; attempt++) + { + int socket_fd = connect_to(numeric_port); + if (socket_fd >= 0) + { + close(socket_fd); + return pid; + } + usleep(20000); + } + kill(pid, SIGTERM); + waitpid(pid, NULL, 0); + assert(FALSE && "SSE test server did not start"); + return -1; +} + +int main(int argc, char **argv) +{ + assert(argc == 2); + int port = reserve_port(); + char port_text[16]; + snprintf(port_text, sizeof(port_text), "%d", port); + pid_t server = start_server(argv[1], port_text); + + int event_socket = connect_to(port); + assert(event_socket >= 0); + send_request(event_socket, "/events"); + char events[4096] = {0}; + receive_until( + event_socket, + events, + sizeof(events), + "data: later\n\n"); + assert(strstr(events, "HTTP/1.1 200 OK\r\n")); + assert(strstr(events, "Content-Type: text/event-stream\r\n")); + assert(strstr(events, "data: connected\n\n")); + + int health_socket = connect_to(port); + assert(health_socket >= 0); + send_request(health_socket, "/health"); + assert(shutdown(health_socket, SHUT_WR) == 0); + char health[1024] = {0}; + receive_until(health_socket, health, sizeof(health), "\r\n\r\nok"); + assert(strstr(health, "HTTP/1.1 200 OK\r\n")); + + close(health_socket); + close(event_socket); + kill(server, SIGTERM); + waitpid(server, NULL, 0); + printf("Seobeo SSE server integration test passed\n"); + return 0; +} diff -r 30c2196d03d4 -r 609d3c6aff4e seobeo/tests/seobeo_sse_test.c --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/seobeo/tests/seobeo_sse_test.c Tue Aug 04 16:49:11 2026 -0700 @@ -0,0 +1,211 @@ +#include "seobeo/seobeo.h" + +#include +#include +#include +#include +#include +#include +#include + +static void read_expected(int socket_fd, const char *expected) +{ + size_t expected_length = strlen(expected); + size_t received = 0; + char buffer[8192] = {0}; + assert(expected_length < sizeof(buffer)); + + while (received < expected_length) + { + ssize_t amount = recv( + socket_fd, + buffer + received, + expected_length - received, + 0); + assert(amount > 0); + received += (size_t)amount; + } + assert(received == expected_length); + assert(memcmp(buffer, expected, expected_length) == 0); +} + +static void route_handler( + Seobeo_SSE_Stream *p_stream, + Seobeo_Request_Entry *p_request, + Dowa_Arena *p_arena) +{ + (void)p_stream; + (void)p_request; + (void)p_arena; +} + +static void initialize_handle( + Seobeo_Handle *p_handle, + int socket_fd, + uint8 *write_buffer, + uint32 write_capacity) +{ + memset(p_handle, 0, sizeof(*p_handle)); + p_handle->socket = socket_fd; + p_handle->connected = TRUE; + p_handle->write_buffer = write_buffer; + p_handle->write_buffer_capacity = write_capacity; + atomic_init(&p_handle->destroyed, FALSE); +} + +int main(void) +{ + const char *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" + "X-Accel-Buffering: no\r\n" + "\r\n"; + int sockets[2]; + assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0); + + uint8 write_buffer[4096] = {0}; + Seobeo_Handle handle; + initialize_handle( + &handle, + sockets[0], + write_buffer, + (uint32)sizeof(write_buffer)); + Seobeo_SSE_Stream stream = {0}; + assert(Seobeo_SSE_Start(&stream, &handle) == 0); + assert(Seobeo_SSE_Is_Open(&stream)); + read_expected(sockets[1], headers); + + Seobeo_SSE_Event event = { + .event = "deployment", + .id = "42", + .data = "started\r\nrunning", + .retry_ms = 1500, + }; + assert(Seobeo_SSE_Send(&stream, &event) == 0); + read_expected( + sockets[1], + "id: 42\n" + "event: deployment\n" + "retry: 1500\n" + "data: started\n" + "data: running\n" + "\n"); + + assert(Seobeo_SSE_Send_Comment( + &stream, + "heartbeat\nstill here") == 0); + read_expected( + sockets[1], + ": heartbeat\n" + ": still here\n" + "\n"); + + int flags = fcntl(sockets[0], F_GETFL, 0); + assert(flags >= 0); + assert(fcntl(sockets[0], F_SETFL, flags | O_NONBLOCK) == 0); + char filler[4096]; + memset(filler, 'z', sizeof(filler)); + while (send(sockets[0], filler, sizeof(filler), 0) > 0) + { + } + assert(errno == EAGAIN || errno == EWOULDBLOCK); + memcpy(handle.write_buffer, "older", 5); + handle.write_buffer_len = 5; + assert(Seobeo_SSE_Send_Data(&stream, "queued") == 1); + while (recv(sockets[1], filler, sizeof(filler), MSG_DONTWAIT) > 0) + { + } + assert(errno == EAGAIN || errno == EWOULDBLOCK); + assert(Seobeo_SSE_Flush(&stream) == 0); + read_expected(sockets[1], "olderdata: queued\n\n"); + + event.event = "bad\nevent"; + assert(Seobeo_SSE_Send(&stream, &event) == SEOBEO_SSE_INVALID_EVENT); + event.event = NULL; + event.id = "bad\rid"; + assert(Seobeo_SSE_Send(&stream, &event) == SEOBEO_SSE_INVALID_EVENT); + + char oversized[4097]; + memset(oversized, 'x', sizeof(oversized) - 1); + oversized[sizeof(oversized) - 1] = '\0'; + assert(Seobeo_SSE_Send_Data( + &stream, + oversized) == SEOBEO_SSE_EVENT_TOO_LARGE); + + Seobeo_SSE_Close(&stream); + assert(!Seobeo_SSE_Is_Open(&stream)); + assert(Seobeo_SSE_Send_Data(&stream, "closed") < 0); + close(sockets[0]); + close(sockets[1]); + + assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0); + initialize_handle( + &handle, + sockets[0], + write_buffer, + (uint32)sizeof(write_buffer)); + memset(&stream, 0, sizeof(stream)); + assert(Seobeo_SSE_Start(&stream, &handle) == 0); + read_expected(sockets[1], headers); + close(sockets[1]); + assert(Seobeo_SSE_Send_Data(&stream, "disconnect") < 0); + close(sockets[0]); + + assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0); + initialize_handle( + &handle, + sockets[0], + write_buffer, + (uint32)sizeof(write_buffer)); + Seobeo_SSE_Stream *p_managed = Seobeo_SSE_Server_Attach( + &handle, + "/events/builds"); + assert(p_managed); + assert(handle.is_sse); + assert(Seobeo_SSE_Retain(p_managed)); + read_expected(sockets[1], headers); + assert(Seobeo_SSE_Send_Data(p_managed, "after-handler") == 0); + read_expected(sockets[1], "data: after-handler\n\n"); + Seobeo_SSE_Server_Detach_Handle(&handle); + assert(!handle.is_sse); + assert(!Seobeo_SSE_Is_Open(p_managed)); + Seobeo_SSE_Release(p_managed); + Seobeo_SSE_Server_Destroy(); + close(sockets[0]); + close(sockets[1]); + + Seobeo_Router_Init(); + Seobeo_Router_Register_SSE("/events/:topic", route_handler); + Seobeo_Router_Register_SSE("/events/:topic/fixed", route_handler); + Dowa_Arena *p_arena = Dowa_Arena_Create(4096); + Seobeo_Request_Entry *p_request = NULL; + Seobeo_SSE_Handler handler = Seobeo_Router_Find_SSE_Handler( + "GET", + "/events/builds", + &p_request, + p_arena); + assert(handler == route_handler); + Seobeo_Request_Entry *p_topic = Dowa_HashMap_Get_Ptr( + p_request, + ":topic"); + assert(p_topic && strcmp(p_topic->value, "builds") == 0); + assert(Seobeo_Router_Find_SSE_Handler( + "POST", + "/events/builds", + &p_request, + p_arena) == NULL); + Seobeo_Request_Entry *p_unmatched = NULL; + assert(Seobeo_Router_Find_SSE_Handler( + "GET", + "/events/builds/other", + &p_unmatched, + p_arena) == NULL); + assert(Dowa_HashMap_Get_Ptr(p_unmatched, ":topic") == NULL); + Dowa_Arena_Free(p_arena); + Seobeo_Router_Destroy(); + + printf("Seobeo SSE tests passed\n"); + return 0; +}