Skip to content

Commit 6f49c9c

Browse files
Keniis0712Keniis0712
committed
feat(http): 通用二进制流读取 anet_async_http_stream_raw
SDK 文件下载需要'按字节块交付'的流式读取 (非按行/非 SSE)。与 SSE 同 gen_emit 引擎, 但交付单元是原始字节块 —— 不切行/不认前缀/不解析。库先攒 并跳过 HTTP 响应头 (到 \r\n\r\n), 之后 body 字节逐块 emit, 回调拿到的是 纯 body 字节流。 明确划分 (与用户对齐): asyncweb 只做'读字节流'这半 (纯传输); 私有帧解析 (小端 uint32 块号/长度、0xffffffff 哨兵)、Range 多线程、blake3、fs 落盘 全归 SDK —— asyncweb 不碰 <stdio.h>/文件系统, 保持纯网络/异步核心定位。 故取消原先宽泛的'二进制下载/WS 上传' Phase C, 收窄为一个通用 raw 流原语。 新增 tests/test_stream_raw: 裸 server 回带 header 的 256 字节二进制 body, 客户端 raw 消费验证跳过 header + 逐块收全 + 顺序。openssl-asan/ubsan 14测、 wolfssl 14测全绿 (test_sse 不受影响)。
1 parent 02b5f01 commit 6f49c9c

5 files changed

Lines changed: 369 additions & 0 deletions

File tree

‎CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -188,6 +188,7 @@ if(ASYNCWEB_STANDALONE)
188188
tests/test_server_concurrent.c
189189
tests/test_client_robustness.c
190190
tests/test_sse.c
191+
tests/test_stream_raw.c
191192
)
192193
target_link_libraries(test_asyncweb PRIVATE asyncweb)
193194
# (低层 socket/stream API 现在都在公开 <asyncweb/...> 里, 经 asyncweb PUBLIC
@@ -210,6 +211,7 @@ if(ASYNCWEB_STANDALONE)
210211
add_test_with_dir(test_server_concurrent test_asyncweb server_concurrent)
211212
add_test_with_dir(test_client_robustness test_asyncweb client_robustness)
212213
add_test_with_dir(test_sse test_asyncweb sse)
214+
add_test_with_dir(test_stream_raw test_asyncweb stream_raw)
213215
endif()
214216

215217
# ============================================================

‎include/asyncweb/http.h‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,24 @@ typedef void (*anet_sse_cb_t)(const anet_sse_event_t *ev, void *userdata);
6767
task_t* anet_async_http_stream(anet_async_http_request_t *req,
6868
anet_sse_cb_t on_event, void *userdata);
6969

70+
/* ---- 通用二进制流读取 ----
71+
* 与 SSE 同引擎, 但交付单元是原始字节块 —— 不切行、不认前缀、不解析。库先解析
72+
* 并跳过 HTTP 响应头 (状态行 + headers 到 \r\n\r\n), 之后的 body 字节逐块 emit,
73+
* 所以回调拿到的是纯 body 字节流 (适合自定义二进制帧协议 / 文件下载)。
74+
* 私有帧解析 / 落盘 / 哈希 / Range 由调用方 (如 SDK) 处理 —— asyncweb 不碰 fs。 */
75+
76+
// 一块原始 body 字节 (仅回调内有效, 需留存请自行拷贝)
77+
typedef struct {
78+
const char *data;
79+
size_t len;
80+
} anet_http_chunk_t;
81+
82+
typedef void (*anet_http_chunk_cb_t)(const anet_http_chunk_t *chunk, void *userdata);
83+
84+
// 发起一个 raw 流式请求。参数同上; body 字节 (跳过 HTTP header 后) 经 on_chunk 交付。
85+
task_t* anet_async_http_stream_raw(anet_async_http_request_t *req,
86+
anet_http_chunk_cb_t on_chunk, void *userdata);
87+
7088
// 简化的异步POST请求
7189
task_t* anet_async_http_post(const char *url, const char *content_type, const char *body, anet_async_http_response_t *response);
7290

‎src/http.c‎

Lines changed: 187 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -917,6 +917,193 @@ task_t* anet_async_http_stream(anet_async_http_request_t *req,
917917
return t;
918918
}
919919

920+
/* ============================================================
921+
* 通用二进制流读取 (raw stream) 实现
922+
* ============================================================ */
923+
924+
typedef struct {
925+
const char *method;
926+
const char *host;
927+
uint16_t port;
928+
int use_tls;
929+
const char *path;
930+
const char **headers;
931+
const char *body;
932+
anet_http_chunk_cb_t on_chunk;
933+
void *userdata;
934+
935+
anet_palsock_t sock;
936+
anet_socket_t *async_sock;
937+
async_ssl_t *async_ssl;
938+
anet_stream_t *stream;
939+
char *request;
940+
941+
// header 跳过状态: 累积响应头直到 \r\n\r\n, 之后 body 字节直接透传。
942+
char *hdrbuf;
943+
size_t hdrbuf_len;
944+
size_t hdrbuf_cap;
945+
int header_done;
946+
} raw_stream_internal_t;
947+
948+
static void raw_emit_bridge(void *item, void *userdata) {
949+
raw_stream_internal_t *st = (raw_stream_internal_t*)userdata;
950+
if (st->on_chunk) {
951+
st->on_chunk((const anet_http_chunk_t*)item, st->userdata);
952+
}
953+
}
954+
955+
// 往 header 缓冲追加一段字节, 保持 NUL 结尾 (供 strstr 找 \r\n\r\n)。失败返回 -1。
956+
static int raw_hdr_push(raw_stream_internal_t *st, const char *p, size_t n) {
957+
if (st->hdrbuf_len + n + 1 > st->hdrbuf_cap) {
958+
size_t ncap = st->hdrbuf_cap ? st->hdrbuf_cap * 2 : 1024;
959+
while (ncap < st->hdrbuf_len + n + 1) ncap *= 2;
960+
char *nb = realloc(st->hdrbuf, ncap);
961+
if (!nb) return -1;
962+
st->hdrbuf = nb;
963+
st->hdrbuf_cap = ncap;
964+
}
965+
memcpy(st->hdrbuf + st->hdrbuf_len, p, n);
966+
st->hdrbuf_len += n;
967+
st->hdrbuf[st->hdrbuf_len] = '\0';
968+
return 0;
969+
}
970+
971+
task_t* task_arg(anet_async_http_stream_raw_) {
972+
gen_dec_vars(
973+
raw_stream_internal_t *st;
974+
future_t *fut;
975+
future_t *resolve_fut;
976+
task_t *task;
977+
struct sockaddr_storage addr;
978+
int addr_len;
979+
char buffer[HTTP_BUFFER_SIZE];
980+
int bytes_read;
981+
anet_http_chunk_t chunk; /* 跨 gen_emit 存活 */
982+
);
983+
gen_begin(ctx);
984+
985+
{
986+
/* 顶层驱动: arg 为 NULL, 从 userdata 取参数 (同 SSE, 见 gen_emit 约束)。 */
987+
gen_var(st) = (raw_stream_internal_t*)gen_userdata();
988+
gen_var(st)->sock = -1;
989+
}
990+
991+
gen_var(resolve_fut) = anet_palsock_resolve_async(gen_var(st)->host, AF_UNSPEC);
992+
if (!gen_var(resolve_fut)) { gen_return(anet_res_status(ANET_ERR)); }
993+
gen_yield(gen_var(resolve_fut));
994+
{
995+
anet_resolve_result_t *r = (anet_resolve_result_t*)future_result(gen_var(resolve_fut));
996+
if (!r || r->ok != 0) { free(r); gen_return(anet_res_status(ANET_ERR)); }
997+
gen_var(addr) = r->addr;
998+
gen_var(addr_len) = r->addr_len;
999+
free(r);
1000+
}
1001+
if (gen_var(addr).ss_family == AF_INET) {
1002+
((struct sockaddr_in*)&gen_var(addr))->sin_port = htons(gen_var(st)->port);
1003+
} else if (gen_var(addr).ss_family == AF_INET6) {
1004+
((struct sockaddr_in6*)&gen_var(addr))->sin6_port = htons(gen_var(st)->port);
1005+
}
1006+
1007+
gen_var(st)->sock = anet_palsock_create(gen_var(addr).ss_family, SOCK_STREAM, 0, 1);
1008+
if (!anet_palsock_is_valid(gen_var(st)->sock)) { gen_return(anet_res_status(ANET_ERR)); }
1009+
gen_var(st)->async_sock = anet_socket_create(gen_var(st)->sock);
1010+
if (!gen_var(st)->async_sock) { gen_return(anet_res_status(ANET_ERR)); }
1011+
gen_var(fut) = anet_socket_connect(gen_var(st)->async_sock, (struct sockaddr*)&gen_var(addr), gen_var(addr_len));
1012+
gen_yield(gen_var(fut));
1013+
if (future_is_rejected(gen_var(fut))) { gen_return(anet_res_status(ANET_ERR)); }
1014+
if (gen_var(st)->use_tls) {
1015+
gen_var(st)->async_ssl = async_ssl_create(ASYNC_SSL_CLIENT, gen_var(st)->host);
1016+
if (!gen_var(st)->async_ssl) { gen_return(anet_res_status(ANET_ERR)); }
1017+
async_ssl_attach_socket(gen_var(st)->async_ssl, gen_var(st)->async_sock);
1018+
gen_var(task) = async_ssl_handshake(gen_var(st)->async_ssl);
1019+
gen_yield_from_task(gen_var(task));
1020+
if (anet_code_of(future_result(gen_var(task)->future)) != 0) { gen_return(anet_res_status(ANET_ERR)); }
1021+
gen_var(st)->stream = anet_stream_from_ssl(gen_var(st)->async_ssl);
1022+
} else {
1023+
gen_var(st)->stream = anet_stream_from_socket(gen_var(st)->async_sock);
1024+
}
1025+
if (!gen_var(st)->stream) { gen_return(anet_res_status(ANET_ERR)); }
1026+
1027+
gen_var(st)->request = create_request_string(
1028+
gen_var(st)->method, gen_var(st)->host, gen_var(st)->port,
1029+
gen_var(st)->path, gen_var(st)->headers, gen_var(st)->body);
1030+
if (!gen_var(st)->request) { gen_return(anet_res_status(ANET_ERR)); }
1031+
gen_var(task) = anet_stream_write_all(gen_var(st)->stream, gen_var(st)->request, strlen(gen_var(st)->request));
1032+
gen_yield_from_task(gen_var(task));
1033+
if (anet_code_of(future_result(gen_var(task)->future)) != 0) { gen_return(anet_res_status(ANET_ERR)); }
1034+
free(gen_var(st)->request);
1035+
gen_var(st)->request = NULL;
1036+
1037+
// 读取: 先攒 header 到 \r\n\r\n, 之后 body 字节整块 emit。
1038+
while (1) {
1039+
gen_var(task) = anet_stream_read(gen_var(st)->stream, sizeof(gen_var(buffer)), gen_var(buffer));
1040+
gen_yield_from_task(gen_var(task));
1041+
gen_var(bytes_read) = (int)anet_code_of(future_result(gen_var(task)->future));
1042+
if (gen_var(bytes_read) <= 0) {
1043+
gen_return(anet_res_status(ANET_OK)); // 流结束
1044+
}
1045+
1046+
if (gen_var(st)->header_done) {
1047+
gen_var(chunk).data = gen_var(buffer);
1048+
gen_var(chunk).len = (size_t)gen_var(bytes_read);
1049+
gen_emit(&gen_var(chunk));
1050+
} else {
1051+
if (raw_hdr_push(gen_var(st), gen_var(buffer), (size_t)gen_var(bytes_read)) != 0) {
1052+
gen_return(anet_res_status(ANET_ERR));
1053+
}
1054+
char *he = strstr(gen_var(st)->hdrbuf, "\r\n\r\n");
1055+
if (he) {
1056+
gen_var(st)->header_done = 1;
1057+
size_t body_off = (size_t)((he + 4) - gen_var(st)->hdrbuf);
1058+
if (gen_var(st)->hdrbuf_len > body_off) {
1059+
gen_var(chunk).data = gen_var(st)->hdrbuf + body_off;
1060+
gen_var(chunk).len = gen_var(st)->hdrbuf_len - body_off;
1061+
gen_emit(&gen_var(chunk));
1062+
}
1063+
}
1064+
}
1065+
}
1066+
1067+
gen_cleanup();
1068+
if (gen_var(st)) {
1069+
free(gen_var(st)->request);
1070+
free(gen_var(st)->hdrbuf);
1071+
free(gen_var(st)->stream);
1072+
if (gen_var(st)->async_ssl) {
1073+
async_ssl_destroy(gen_var(st)->async_ssl);
1074+
} else if (gen_var(st)->async_sock) {
1075+
anet_socket_close(gen_var(st)->async_sock);
1076+
free(gen_var(st)->async_sock);
1077+
} else if (anet_palsock_is_valid(gen_var(st)->sock)) {
1078+
anet_palsock_close(gen_var(st)->sock);
1079+
}
1080+
free(gen_var(st));
1081+
gen_var(st) = NULL;
1082+
}
1083+
gen_end(NULL);
1084+
}
1085+
1086+
task_t* anet_async_http_stream_raw(anet_async_http_request_t *req,
1087+
anet_http_chunk_cb_t on_chunk, void *userdata) {
1088+
if (!req) return NULL;
1089+
raw_stream_internal_t *st = calloc(1, sizeof(*st));
1090+
if (!st) return NULL;
1091+
st->method = req->method;
1092+
st->host = req->host;
1093+
st->port = req->port;
1094+
st->use_tls = req->use_tls;
1095+
st->path = req->path;
1096+
st->headers = req->headers;
1097+
st->body = req->body;
1098+
st->on_chunk = on_chunk;
1099+
st->userdata = userdata;
1100+
1101+
task_t *t = anet_async_http_stream_raw_(st);
1102+
if (!t) { free(st); return NULL; }
1103+
task_set_on_emit(t, raw_emit_bridge, st);
1104+
return t;
1105+
}
1106+
9201107
// 简化的异步GET请求
9211108
task_t* task_arg(anet_async_http_get_) {
9221109
gen_dec_vars(

‎tests/main.c‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ int test_server_adversarial();
1717
int test_server_concurrent();
1818
int test_client_robustness();
1919
int test_sse();
20+
int test_stream_raw();
2021

2122
int main(int argc, char** argv) {
2223
setvbuf(stdout, NULL, _IONBF, 0);
@@ -70,6 +71,9 @@ int main(int argc, char** argv) {
7071
if (strcmp(argv[1], "sse") == 0) {
7172
return test_sse();
7273
}
74+
if (strcmp(argv[1], "stream_raw") == 0) {
75+
return test_stream_raw();
76+
}
7377

7478
printf("Unknown test: %s\n", argv[1]);
7579
printf("Available tests: sync_http, async_http, sync_ws, async_ws, server, server6, http_server, https_server, keepalive, ws_server, wss_server\n");

0 commit comments

Comments
 (0)