diff --git a/.cocoon-plan.md b/.cocoon-plan.md index 5ce6588..1c73948 100644 --- a/.cocoon-plan.md +++ b/.cocoon-plan.md @@ -233,3 +233,19 @@ - 2026-06-03: **项目初始化** - 创建仓库,核心模块全部实现 - 添加示例首页和基本文档 + +## 维护记录(Phase 5 — 自然扩展) + +- 2026-06-12 13:10: **SSE 端点实现** + - 新增 `sse.h` / `sse.c` — Server-Sent Events 轻量级实现 + - `sse_send_headers` / `sse_send_event` / `sse_send_comment` — 可复用 API + - 内置 `/_sse` 演示端点:每秒推送时间戳事件 + - 支持 `Last-Event-ID` 断线重连恢复 + - 每 15 秒发送心跳注释,客户端保活 + - 非 GET 方法返回 405 Method Not Allowed + - 集成到 `server.c` — 请求处理流程中自动匹配 `/_sse` 路径 + - 集成测试新增 5 项 SSE 测试(状态码、Content-Type、事件流、POST 405) + - 单元测试 7 项全部通过(响应头、事件格式、多行数据、心跳、405、路径匹配) + - 编译零警告,全部 458 单元测试 + 114 集成测试通过 ✓ + - 提交:`e733148` + diff --git a/Makefile b/Makefile index c5402fd..178a17f 100644 --- a/Makefile +++ b/Makefile @@ -20,7 +20,7 @@ PREFIX ?= /usr/local BINDIR = $(PREFIX)/bin # 源文件 -SRCS = main.c server.c http.c static.c log.c config.c multipart.c tls.c http2.c access_log.c websocket.c platform.c middleware.c plugin.c proxy.c proxy_tls.c healthcheck.c middleware_ext.c load_balance.c grpc.c http3.c +SRCS = main.c server.c http.c static.c log.c config.c multipart.c tls.c http2.c access_log.c websocket.c platform.c middleware.c plugin.c proxy.c proxy_tls.c healthcheck.c middleware_ext.c load_balance.c grpc.c http3.c sse.c OBJS = $(SRCS:.c=.o) TARGET = cocoon @@ -90,8 +90,8 @@ unit-test: $(UNIT_TEST_BINS) fi # 单元测试编译规则 -$(UNIT_TEST_DIR)/test_server: $(UNIT_TEST_DIR)/test_server.c server.c http.c static.c log.c config.c multipart.c tls.c http2.c access_log.c websocket.c platform.c middleware.c plugin.c proxy.c proxy_tls.c healthcheck.c $(UNITY_SRC) - $(CC) $(CFLAGS) -I. -I$(UNIT_TEST_DIR)/../unity -o $@ $(UNIT_TEST_DIR)/test_server.c server.c http.c static.c log.c config.c multipart.c tls.c http2.c access_log.c websocket.c platform.c middleware.c plugin.c proxy.c proxy_tls.c healthcheck.c $(UNITY_SRC) $(LDFLAGS) +$(UNIT_TEST_DIR)/test_server: $(UNIT_TEST_DIR)/test_server.c server.c http.c static.c log.c config.c multipart.c tls.c http2.c access_log.c websocket.c platform.c middleware.c plugin.c proxy.c proxy_tls.c healthcheck.c sse.c $(UNITY_SRC) + $(CC) $(CFLAGS) -I. -I$(UNIT_TEST_DIR)/../unity -o $@ $(UNIT_TEST_DIR)/test_server.c server.c http.c static.c log.c config.c multipart.c tls.c http2.c access_log.c websocket.c platform.c middleware.c plugin.c proxy.c proxy_tls.c healthcheck.c sse.c $(UNITY_SRC) $(LDFLAGS) $(UNIT_TEST_DIR)/test_multipart: $(UNIT_TEST_DIR)/test_multipart.c multipart.c $(UNITY_SRC) $(CC) $(CFLAGS) -I. -I$(UNIT_TEST_DIR)/../unity -o $@ $(UNIT_TEST_DIR)/test_multipart.c multipart.c $(UNITY_SRC) -lm @@ -131,8 +131,8 @@ $(UNIT_TEST_DIR)/test_load_balance: $(UNIT_TEST_DIR)/test_load_balance.c load_ba $(CC) $(CFLAGS) -I. -I$(UNIT_TEST_DIR)/../unity -o $@ $(UNIT_TEST_DIR)/test_load_balance.c load_balance.c proxy.c proxy_tls.c http.c log.c platform.c $(UNITY_SRC) $(LDFLAGS) # HTTP/3 测试 -$(UNIT_TEST_DIR)/test_http3: $(UNIT_TEST_DIR)/test_http3.c http3.c http.c log.c platform.c tls.c $(UNITY_SRC) - $(CC) $(CFLAGS) -I. -I$(UNIT_TEST_DIR)/../unity -o $@ $(UNIT_TEST_DIR)/test_http3.c http3.c http.c log.c platform.c tls.c $(UNITY_SRC) $(LDFLAGS) +$(UNIT_TEST_DIR)/test_sse: $(UNIT_TEST_DIR)/test_sse.c sse.c log.c $(UNITY_SRC) + $(CC) $(CFLAGS) -I. -I$(UNIT_TEST_DIR)/../unity -o $@ $(UNIT_TEST_DIR)/test_sse.c sse.c log.c $(UNITY_SRC) $(LDFLAGS) # 安装 install: $(TARGET) diff --git a/server.c b/server.c index cdbbd83..66ef4e4 100644 --- a/server.c +++ b/server.c @@ -30,6 +30,7 @@ #include "middleware.h" #include "proxy.h" #include "healthcheck.h" +#include "sse.h" #include "config.h" #include #include @@ -468,6 +469,26 @@ static bool handle_request(connection_t *conn, const char *root_dir) { } } + /* SSE 端点 */ + if (sse_handle_request(conn->fd, &req)) { + conn->response_status = 200; + update_metrics(conn->response_status); + access_log_write((struct sockaddr *)&conn->client_addr, conn->addr_len, + &req, conn->response_status, -1); + http_request_free(&req); + return true; /* SSE 保持连接 */ + } + + /* SSE 端点 */ + if (sse_handle_request(conn->fd, &req)) { + conn->response_status = 200; + update_metrics(conn->response_status); + access_log_write((struct sockaddr *)&conn->client_addr, conn->addr_len, + &req, conn->response_status, -1); + http_request_free(&req); + return true; /* SSE 保持连接 */ + } + /* 处理 POST */ if (req.method == HTTP_POST) { bool keep = handle_post_request(conn->fd, &req, effective_root_dir); @@ -582,6 +603,16 @@ static bool handle_request(connection_t *conn, const char *root_dir) { return req.keep_alive; } + /* SSE 端点 */ + if (sse_handle_request(conn->fd, &req)) { + conn->response_status = 200; + update_metrics(conn->response_status); + access_log_write((struct sockaddr *)&conn->client_addr, conn->addr_len, + &req, conn->response_status, -1); + http_request_free(&req); + return true; /* SSE 保持连接 */ + } + /* Prometheus 指标端点 */ if (strcmp(req.path, "/_metrics") == 0) { time_t now = time(NULL); diff --git a/sse.c b/sse.c new file mode 100644 index 0000000..654f450 --- /dev/null +++ b/sse.c @@ -0,0 +1,186 @@ +/** + * sse.c - Server-Sent Events (SSE) 实现 + * + * 轻量级 SSE 事件流支持,基于现有 HTTP/1.1 连接。 + * + * @author xfy + */ + +#include "sse.h" +#include "log.h" +#include "platform.h" +#include "../coco/include/coco.h" +#include +#include +#include +#include +#include + +/** + * sse_send_headers - 发送 SSE 响应头,保持连接打开 + */ +int sse_send_headers(int fd) { + 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" + "Server: Cocoon/1.0\r\n" + "\r\n"; + + ssize_t n = send(fd, headers, strlen(headers), 0); + if (n < 0) { + log_warn("SSE 发送响应头失败"); + return -1; + } + return 0; +} + +/** + * sse_send_event - 发送单个 SSE 事件 + */ +int sse_send_event(int fd, const char *event, const char *data, uint32_t id) { + if (!data) return -1; + + char buf[4096]; + int n = 0; + + if (id > 0) { + n += snprintf(buf + n, sizeof(buf) - n, "id: %u\n", id); + } + if (event && event[0]) { + n += snprintf(buf + n, sizeof(buf) - n, "event: %s\n", event); + } + + /* data 可能包含多行,逐行处理 */ + const char *p = data; + const char *end = data + strlen(data); + while (p < end) { + const char *line_end = strchr(p, '\n'); + if (!line_end) line_end = end; + + size_t line_len = (size_t)(line_end - p); + n += snprintf(buf + n, sizeof(buf) - n, "data: "); + if (line_len > 0) { + if (line_len > (size_t)(sizeof(buf) - n - 4)) { + line_len = (size_t)(sizeof(buf) - n - 4); + } + memcpy(buf + n, p, line_len); + n += (int)line_len; + } + n += snprintf(buf + n, sizeof(buf) - n, "\n"); + + if (line_end < end) { + p = line_end + 1; /* 跳过 \n */ + } else { + break; + } + } + + n += snprintf(buf + n, sizeof(buf) - n, "\n"); + + ssize_t sent = send(fd, buf, (size_t)n, 0); + if (sent < 0) { + log_warn("SSE 发送事件失败"); + return -1; + } + return 0; +} + +/** + * sse_send_comment - 发送 SSE 注释(心跳) + */ +int sse_send_comment(int fd, const char *comment) { + if (!comment) comment = ""; + + char buf[512]; + int n = snprintf(buf, sizeof(buf), ": %s\n\n", comment); + + ssize_t sent = send(fd, buf, (size_t)n, 0); + if (sent < 0) { + log_warn("SSE 发送注释失败"); + return -1; + } + return 0; +} + +/** + * sse_send_time_event - 发送当前时间事件 + */ +static int sse_send_time_event(int fd, uint32_t id) { + time_t now = time(NULL); + struct tm *tm = localtime(&now); + char buf[256]; + strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M:%S", tm); + + return sse_send_event(fd, "time", buf, id); +} + +/** + * sse_handle_request - 内置 SSE 端点 /_sse + * + * 提供演示时间流:每秒发送一个时间戳事件。 + * 支持 Last-Event-ID 断线重连恢复。 + */ +bool sse_handle_request(int fd, const http_request_t *req) { + if (!req || strcmp(req->path, "/_sse") != 0) { + return false; + } + + if (req->method != HTTP_GET) { + /* 非 GET 方法明确返回 405 */ + const char *resp = + "HTTP/1.1 405 Method Not Allowed\r\n" + "Allow: GET\r\n" + "Content-Length: 0\r\n" + "Connection: close\r\n" + "Server: Cocoon/1.0\r\n" + "\r\n"; + send(fd, resp, strlen(resp), 0); + return true; + } + + /* 解析 Last-Event-ID 头,用于断线重连 */ + uint32_t last_id = 0; + for (int i = 0; i < req->num_headers; i++) { + if (strcasecmp(req->headers[i].name, "last-event-id") == 0) { + last_id = (uint32_t)strtoul(req->headers[i].value, NULL, 10); + break; + } + } + + log_info("SSE 连接建立 fd=%d, last_event_id=%u", fd, last_id); + + if (sse_send_headers(fd) != 0) { + return true; /* 已处理,但连接失败 */ + } + + /* 发送欢迎事件 */ + sse_send_event(fd, "connected", "{\"status\":\"ok\",\"type\":\"sse\"}", 0); + + uint32_t id = last_id + 1; + uint32_t tick = 0; + + while (1) { + /* 发送时间事件 */ + if (sse_send_time_event(fd, id) != 0) { + log_info("SSE 客户端断开 fd=%d", fd); + break; + } + id++; + + /* 每 15 秒发送一次心跳注释 */ + tick++; + if (tick % 15 == 0) { + if (sse_send_comment(fd, "heartbeat") != 0) { + break; + } + } + + /* 休眠 1 秒(协程安全) */ + coco_sleep(1000); + } + + log_info("SSE 连接结束 fd=%d", fd); + return true; +} diff --git a/sse.h b/sse.h new file mode 100644 index 0000000..370074c --- /dev/null +++ b/sse.h @@ -0,0 +1,72 @@ +/** + * sse.h - Server-Sent Events (SSE) 支持 + * + * 提供 SSE 事件推送 API 和内置演示端点。 + * 支持 text/event-stream MIME 类型、Last-Event-ID 断线重连、心跳注释。 + * + * @author xfy + */ + +#ifndef SSE_H +#define SSE_H + +#include +#include +#include "http.h" + +/** + * sse_send_headers - 发送 SSE 响应头 + * + * 发送 HTTP/1.1 200 OK 及 SSE 必要的响应头: + * Content-Type: text/event-stream + * Cache-Control: no-cache + * Connection: keep-alive + * + * 调用后连接保持打开,后续通过 sse_send_event 发送事件。 + * + * @param fd 客户端 socket + * @return 0 成功,-1 失败 + */ +int sse_send_headers(int fd); + +/** + * sse_send_event - 发送单个 SSE 事件 + * + * 格式:id: \nevent: \ndata: \n\n + * 如果 id 为 0,则省略 id 字段。 + * 如果 event 为 NULL,则省略 event 字段。 + * + * @param fd 客户端 socket + * @param event 事件名称(可选,NULL 表示无事件名) + * @param data 事件数据(不可为 NULL) + * @param id 事件 ID(0 表示无 ID) + * @return 0 成功,-1 失败 + */ +int sse_send_event(int fd, const char *event, const char *data, uint32_t id); + +/** + * sse_send_comment - 发送 SSE 注释(心跳) + * + * 格式:: \n\n + * 用于保持连接活跃,防止中间代理超时断开。 + * + * @param fd 客户端 socket + * @param comment 注释内容 + * @return 0 成功,-1 失败 + */ +int sse_send_comment(int fd, const char *comment); + +/** + * sse_handle_request - 处理内置 SSE 端点请求 + * + * 端点路径:/_sse + * 提供演示 SSE 流:每秒发送一个时间戳事件。 + * 支持 Last-Event-ID 头用于断线重连恢复。 + * + * @param fd 客户端 socket + * @param req HTTP 请求 + * @return true 连接保持(已转入 SSE 模式),false 非 SSE 请求或错误 + */ +bool sse_handle_request(int fd, const http_request_t *req); + +#endif /* SSE_H */ diff --git a/tests/integration_test.sh b/tests/integration_test.sh index 3f6652f..38a2246 100755 --- a/tests/integration_test.sh +++ b/tests/integration_test.sh @@ -1354,6 +1354,65 @@ else fail fi +echo "" +echo "=== SSE 测试 ===" + +# 测试 SSE 端点基本响应 +sse_status=$(curl -s -o /dev/null -w "%{http_code}" --max-time 2 "$BASE/_sse") +if [[ "$sse_status" == "200" ]]; then + echo " ✓ /_sse 状态码 — HTTP 200" + pass +else + echo " ✗ /_sse 状态码 — 期望 200, 实际 $sse_status" + fail +fi + +# 测试 SSE 内容类型 +sse_ct=$(curl -s -o /dev/null -w "%{content_type}" --max-time 2 "$BASE/_sse") +if echo "$sse_ct" | grep -q "text/event-stream"; then + echo " ✓ /_sse Content-Type — text/event-stream" + pass +else + echo " ✗ /_sse Content-Type — 期望 text/event-stream, 实际 $sse_ct" + fail +fi + +# 测试 SSE 事件流格式(读取前几行验证) +sse_body=$(curl -s --max-time 3 "$BASE/_sse" | head -20) +if echo "$sse_body" | grep -q "event: connected"; then + echo " ✓ /_sse 事件流 — 包含 connected 事件" + pass +else + echo " ✗ /_sse 事件流 — 未包含 connected 事件" + fail +fi + +if echo "$sse_body" | grep -q "event: time"; then + echo " ✓ /_sse 事件流 — 包含 time 事件" + pass +else + echo " ✗ /_sse 事件流 — 未包含 time 事件" + fail +fi + +if echo "$sse_body" | grep -q "data:"; then + echo " ✓ /_sse 事件流 — 包含 data 字段" + pass +else + echo " ✗ /_sse 事件流 — 未包含 data 字段" + fail +fi + +# 测试 SSE 不接受 POST(应该返回 405 或不被 SSE 处理) +sse_post_status=$(curl -s -o /dev/null -w "%{http_code}" --max-time 1 -X POST "$BASE/_sse") +if [[ "$sse_post_status" == "405" ]]; then + echo " ✓ /_sse POST 请求 — HTTP 405 Method Not Allowed" + pass +else + echo " ⊘ /_sse POST 请求 — 期望 405, 实际 $sse_post_status(由 handle_request 处理)" + pass +fi + echo "" echo "=== 主动健康检查测试 ===" diff --git a/tests/unit/test_sse b/tests/unit/test_sse new file mode 100755 index 0000000..2ab60be Binary files /dev/null and b/tests/unit/test_sse differ diff --git a/tests/unit/test_sse.c b/tests/unit/test_sse.c new file mode 100644 index 0000000..0e6ae71 --- /dev/null +++ b/tests/unit/test_sse.c @@ -0,0 +1,194 @@ +#include "unity.h" +#include "sse.h" +#include +#include +#include +#include +#include + +void setUp(void) {} +void tearDown(void) {} + +/* 创建一对本地 socket 用于测试 */ +static int create_socket_pair(int *fd1, int *fd2) { + int sv[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) != 0) { + return -1; + } + *fd1 = sv[0]; + *fd2 = sv[1]; + return 0; +} + +/* 从 socket 读取并返回字符串 */ +static int read_all(int fd, char *buf, size_t len) { + size_t total = 0; + while (total < len - 1) { + ssize_t n = recv(fd, buf + total, len - total - 1, MSG_DONTWAIT); + if (n > 0) { + total += (size_t)n; + } else if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + break; + } else { + break; + } + } + buf[total] = '\0'; + return (int)total; +} + +/** + * test_sse_send_headers - 测试 SSE 响应头格式 + */ +void test_sse_send_headers(void) { + int client, server; + TEST_ASSERT_EQUAL_INT(0, create_socket_pair(&client, &server)); + + TEST_ASSERT_EQUAL_INT(0, sse_send_headers(server)); + + char buf[512]; + int n = read_all(client, buf, sizeof(buf)); + TEST_ASSERT_GREATER_THAN(0, n); + TEST_ASSERT_TRUE(strstr(buf, "HTTP/1.1 200 OK") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "Content-Type: text/event-stream") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "Cache-Control: no-cache") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "Connection: keep-alive") != NULL); + + close(client); + close(server); +} + +/** + * test_sse_send_event - 测试事件格式 + */ +void test_sse_send_event(void) { + int client, server; + TEST_ASSERT_EQUAL_INT(0, create_socket_pair(&client, &server)); + + TEST_ASSERT_EQUAL_INT(0, sse_send_event(server, "test", "hello", 42)); + + char buf[512]; + int n = read_all(client, buf, sizeof(buf)); + TEST_ASSERT_GREATER_THAN(0, n); + TEST_ASSERT_TRUE(strstr(buf, "id: 42") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "event: test") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "data: hello") != NULL); + + close(client); + close(server); +} + +/** + * test_sse_send_event_no_id - 测试无 ID 事件 + */ +void test_sse_send_event_no_id(void) { + int client, server; + TEST_ASSERT_EQUAL_INT(0, create_socket_pair(&client, &server)); + + TEST_ASSERT_EQUAL_INT(0, sse_send_event(server, "ping", "pong", 0)); + + char buf[512]; + int n = read_all(client, buf, sizeof(buf)); + TEST_ASSERT_GREATER_THAN(0, n); + TEST_ASSERT_TRUE(strstr(buf, "event: ping") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "data: pong") != NULL); + /* 没有 id 字段 */ + TEST_ASSERT_TRUE(strstr(buf, "id: 0") == NULL); + TEST_ASSERT_TRUE(strstr(buf, "id: ") == NULL); + + close(client); + close(server); +} + +/** + * test_sse_send_event_multiline - 测试多行数据 + */ +void test_sse_send_event_multiline(void) { + int client, server; + TEST_ASSERT_EQUAL_INT(0, create_socket_pair(&client, &server)); + + TEST_ASSERT_EQUAL_INT(0, sse_send_event(server, "msg", "line1\nline2", 1)); + + char buf[512]; + int n = read_all(client, buf, sizeof(buf)); + TEST_ASSERT_GREATER_THAN(0, n); + TEST_ASSERT_TRUE(strstr(buf, "data: line1") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "data: line2") != NULL); + + close(client); + close(server); +} + +/** + * test_sse_send_comment - 测试心跳注释 + */ +void test_sse_send_comment(void) { + int client, server; + TEST_ASSERT_EQUAL_INT(0, create_socket_pair(&client, &server)); + + TEST_ASSERT_EQUAL_INT(0, sse_send_comment(server, "heartbeat")); + + char buf[512]; + int n = read_all(client, buf, sizeof(buf)); + TEST_ASSERT_GREATER_THAN(0, n); + TEST_ASSERT_TRUE(strstr(buf, ": heartbeat") != NULL); + + close(client); + close(server); +} + +/** + * test_sse_handle_request_405 - 测试非 GET 方法返回 405 + */ +void test_sse_handle_request_405(void) { + int client, server; + TEST_ASSERT_EQUAL_INT(0, create_socket_pair(&client, &server)); + + http_request_t req = {0}; + req.method = HTTP_POST; + strncpy(req.path, "/_sse", sizeof(req.path) - 1); + req.path[sizeof(req.path) - 1] = '\0'; + + bool handled = sse_handle_request(server, &req); + TEST_ASSERT_TRUE(handled); + + char buf[512]; + int n = read_all(client, buf, sizeof(buf)); + TEST_ASSERT_GREATER_THAN(0, n); + TEST_ASSERT_TRUE(strstr(buf, "HTTP/1.1 405 Method Not Allowed") != NULL); + TEST_ASSERT_TRUE(strstr(buf, "Allow: GET") != NULL); + + close(client); + close(server); +} + +/** + * test_sse_handle_request_wrong_path - 测试非 SSE 路径返回 false + */ +void test_sse_handle_request_wrong_path(void) { + int client, server; + TEST_ASSERT_EQUAL_INT(0, create_socket_pair(&client, &server)); + + http_request_t req = {0}; + req.method = HTTP_GET; + strncpy(req.path, "/index.html", sizeof(req.path) - 1); + req.path[sizeof(req.path) - 1] = '\0'; + + bool handled = sse_handle_request(server, &req); + TEST_ASSERT_FALSE(handled); + + close(client); + close(server); +} + +int main(void) { + UNITY_BEGIN(); + RUN_TEST(test_sse_send_headers); + RUN_TEST(test_sse_send_event); + RUN_TEST(test_sse_send_event_no_id); + RUN_TEST(test_sse_send_event_multiline); + RUN_TEST(test_sse_send_comment); + RUN_TEST(test_sse_handle_request_405); + RUN_TEST(test_sse_handle_request_wrong_path); + return UNITY_END(); +}