Mercurial

comparison seobeo/tests/seobeo_sse_test.c @ 257:609d3c6aff4e

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[seobeo] Add persistent SSE streams Co-authored-by: Copilot <[email protected]>
author MrJuneJune <me@mrjunejune.com>
date Tue, 04 Aug 2026 16:49:11 -0700
parents
children 1f9877b637e9
comparison
equal deleted inserted replaced
256:30c2196d03d4 257:609d3c6aff4e
1 #include "seobeo/seobeo.h"
2
3 #include <assert.h>
4 #include <fcntl.h>
5 #include <stdatomic.h>
6 #include <stdio.h>
7 #include <string.h>
8 #include <sys/socket.h>
9 #include <unistd.h>
10
11 static void read_expected(int socket_fd, const char *expected)
12 {
13 size_t expected_length = strlen(expected);
14 size_t received = 0;
15 char buffer[8192] = {0};
16 assert(expected_length < sizeof(buffer));
17
18 while (received < expected_length)
19 {
20 ssize_t amount = recv(
21 socket_fd,
22 buffer + received,
23 expected_length - received,
24 0);
25 assert(amount > 0);
26 received += (size_t)amount;
27 }
28 assert(received == expected_length);
29 assert(memcmp(buffer, expected, expected_length) == 0);
30 }
31
32 static void route_handler(
33 Seobeo_SSE_Stream *p_stream,
34 Seobeo_Request_Entry *p_request,
35 Dowa_Arena *p_arena)
36 {
37 (void)p_stream;
38 (void)p_request;
39 (void)p_arena;
40 }
41
42 static void initialize_handle(
43 Seobeo_Handle *p_handle,
44 int socket_fd,
45 uint8 *write_buffer,
46 uint32 write_capacity)
47 {
48 memset(p_handle, 0, sizeof(*p_handle));
49 p_handle->socket = socket_fd;
50 p_handle->connected = TRUE;
51 p_handle->write_buffer = write_buffer;
52 p_handle->write_buffer_capacity = write_capacity;
53 atomic_init(&p_handle->destroyed, FALSE);
54 }
55
56 int main(void)
57 {
58 const char *headers =
59 "HTTP/1.1 200 OK\r\n"
60 "Content-Type: text/event-stream\r\n"
61 "Cache-Control: no-cache\r\n"
62 "Connection: keep-alive\r\n"
63 "X-Accel-Buffering: no\r\n"
64 "\r\n";
65 int sockets[2];
66 assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0);
67
68 uint8 write_buffer[4096] = {0};
69 Seobeo_Handle handle;
70 initialize_handle(
71 &handle,
72 sockets[0],
73 write_buffer,
74 (uint32)sizeof(write_buffer));
75 Seobeo_SSE_Stream stream = {0};
76 assert(Seobeo_SSE_Start(&stream, &handle) == 0);
77 assert(Seobeo_SSE_Is_Open(&stream));
78 read_expected(sockets[1], headers);
79
80 Seobeo_SSE_Event event = {
81 .event = "deployment",
82 .id = "42",
83 .data = "started\r\nrunning",
84 .retry_ms = 1500,
85 };
86 assert(Seobeo_SSE_Send(&stream, &event) == 0);
87 read_expected(
88 sockets[1],
89 "id: 42\n"
90 "event: deployment\n"
91 "retry: 1500\n"
92 "data: started\n"
93 "data: running\n"
94 "\n");
95
96 assert(Seobeo_SSE_Send_Comment(
97 &stream,
98 "heartbeat\nstill here") == 0);
99 read_expected(
100 sockets[1],
101 ": heartbeat\n"
102 ": still here\n"
103 "\n");
104
105 int flags = fcntl(sockets[0], F_GETFL, 0);
106 assert(flags >= 0);
107 assert(fcntl(sockets[0], F_SETFL, flags | O_NONBLOCK) == 0);
108 char filler[4096];
109 memset(filler, 'z', sizeof(filler));
110 while (send(sockets[0], filler, sizeof(filler), 0) > 0)
111 {
112 }
113 assert(errno == EAGAIN || errno == EWOULDBLOCK);
114 memcpy(handle.write_buffer, "older", 5);
115 handle.write_buffer_len = 5;
116 assert(Seobeo_SSE_Send_Data(&stream, "queued") == 1);
117 while (recv(sockets[1], filler, sizeof(filler), MSG_DONTWAIT) > 0)
118 {
119 }
120 assert(errno == EAGAIN || errno == EWOULDBLOCK);
121 assert(Seobeo_SSE_Flush(&stream) == 0);
122 read_expected(sockets[1], "olderdata: queued\n\n");
123
124 event.event = "bad\nevent";
125 assert(Seobeo_SSE_Send(&stream, &event) == SEOBEO_SSE_INVALID_EVENT);
126 event.event = NULL;
127 event.id = "bad\rid";
128 assert(Seobeo_SSE_Send(&stream, &event) == SEOBEO_SSE_INVALID_EVENT);
129
130 char oversized[4097];
131 memset(oversized, 'x', sizeof(oversized) - 1);
132 oversized[sizeof(oversized) - 1] = '\0';
133 assert(Seobeo_SSE_Send_Data(
134 &stream,
135 oversized) == SEOBEO_SSE_EVENT_TOO_LARGE);
136
137 Seobeo_SSE_Close(&stream);
138 assert(!Seobeo_SSE_Is_Open(&stream));
139 assert(Seobeo_SSE_Send_Data(&stream, "closed") < 0);
140 close(sockets[0]);
141 close(sockets[1]);
142
143 assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0);
144 initialize_handle(
145 &handle,
146 sockets[0],
147 write_buffer,
148 (uint32)sizeof(write_buffer));
149 memset(&stream, 0, sizeof(stream));
150 assert(Seobeo_SSE_Start(&stream, &handle) == 0);
151 read_expected(sockets[1], headers);
152 close(sockets[1]);
153 assert(Seobeo_SSE_Send_Data(&stream, "disconnect") < 0);
154 close(sockets[0]);
155
156 assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0);
157 initialize_handle(
158 &handle,
159 sockets[0],
160 write_buffer,
161 (uint32)sizeof(write_buffer));
162 Seobeo_SSE_Stream *p_managed = Seobeo_SSE_Server_Attach(
163 &handle,
164 "/events/builds");
165 assert(p_managed);
166 assert(handle.is_sse);
167 assert(Seobeo_SSE_Retain(p_managed));
168 read_expected(sockets[1], headers);
169 assert(Seobeo_SSE_Send_Data(p_managed, "after-handler") == 0);
170 read_expected(sockets[1], "data: after-handler\n\n");
171 Seobeo_SSE_Server_Detach_Handle(&handle);
172 assert(!handle.is_sse);
173 assert(!Seobeo_SSE_Is_Open(p_managed));
174 Seobeo_SSE_Release(p_managed);
175 Seobeo_SSE_Server_Destroy();
176 close(sockets[0]);
177 close(sockets[1]);
178
179 Seobeo_Router_Init();
180 Seobeo_Router_Register_SSE("/events/:topic", route_handler);
181 Seobeo_Router_Register_SSE("/events/:topic/fixed", route_handler);
182 Dowa_Arena *p_arena = Dowa_Arena_Create(4096);
183 Seobeo_Request_Entry *p_request = NULL;
184 Seobeo_SSE_Handler handler = Seobeo_Router_Find_SSE_Handler(
185 "GET",
186 "/events/builds",
187 &p_request,
188 p_arena);
189 assert(handler == route_handler);
190 Seobeo_Request_Entry *p_topic = Dowa_HashMap_Get_Ptr(
191 p_request,
192 ":topic");
193 assert(p_topic && strcmp(p_topic->value, "builds") == 0);
194 assert(Seobeo_Router_Find_SSE_Handler(
195 "POST",
196 "/events/builds",
197 &p_request,
198 p_arena) == NULL);
199 Seobeo_Request_Entry *p_unmatched = NULL;
200 assert(Seobeo_Router_Find_SSE_Handler(
201 "GET",
202 "/events/builds/other",
203 &p_unmatched,
204 p_arena) == NULL);
205 assert(Dowa_HashMap_Get_Ptr(p_unmatched, ":topic") == NULL);
206 Dowa_Arena_Free(p_arena);
207 Seobeo_Router_Destroy();
208
209 printf("Seobeo SSE tests passed\n");
210 return 0;
211 }