当前位置：首页 > news >正文

HTTP SSE 实现

news 2025/2/23 19:09:08

参考：
SSE协议
SSE技术详解：使用 HTTP 做服务端数据推送应用的技术

一句概扩

SSE可理解为：服务端和客户端建立连接之后双方均保持连接，但仅支持服务端向客户端推送数据。推送完毕之后关闭连接，无状态行。

下面是基于libhv实现的SSE

`SSE server`

int Handler::sse(const HttpContextPtr& ctx) {// SSEvent(message) every 1shv::setInterval(10000, [ctx](hv::TimerID timerID) {static int ncount = 0;if (ctx->writer->isConnected()) {char szTime[DATETIME_FMT_BUFLEN] = {0};datetime_t now = datetime_now();datetime_fmt(&now, szTime);ctx->writer->SSEvent(szTime);if (++ncount >= 10) {//hv::killTimer(timerID);ctx->writer->close();ncount = 0;}} else {hv::killTimer(timerID);}});return HTTP_STATUS_UNFINISHED;
}

`SSE client`

typedef std::function<void(const std::string& sid, const std::string& sevent, const std::string& sdata, const unsigned int retry_ms)> sse_msg_cb;
HV_INLINE int sse(http_method method, const char* url, const sse_msg_cb& msg_cb, const http_body& body = NoBody, const http_headers& headers = DefaultHeaders,const unsigned int timeout_s = -1) {hv::HttpClient cli;HttpRequest req;HttpResponse resp;req.url = url; //req.method = method;req.timeout = timeout_s; // 不超时if (&body != &NoBody) {req.body = body;}if (&headers != &DefaultHeaders) {req.headers = headers;}bool bstream = false;req.http_cb = [msg_cb, &bstream](HttpMessage* resp, http_parser_state state, const char* data, size_t size) {if (state == HP_HEADERS_COMPLETE) {if (resp->headers["Content-Type"] == "text/event-stream") {bstream = true;return 0;}}else if (state == HP_BODY) {/*binary body should check data*/// printf("%s", std::string(data, size).c_str());resp->body.append(data, size);if (!bstream) return 0;/*/n/n获取message*/size_t ifind = std::string::npos;while ((ifind = resp->body.find("\n\n")) != std::string::npos) {std::string msg = resp->body.substr(0, ifind + 2);resp->body.erase(0, ifind + 2);/*解析body,暂时不考虑多dataid:xxx\nevent:xxx\ndata:xxx\ndata:xxx\ndata:xxx\nretry:10000\n*/auto kvs = hv::splitKV(msg, '\n', ':');if (msg_cb && (kvs.count("id") || kvs.count("event") || kvs.count("data") || kvs.count("retry")))msg_cb(kvs.count("id") ? kvs["id"] : "", kvs.count("event") ? kvs["event"] : "", kvs.count("data") ? kvs["data"] : "",kvs.count("retry") ? atoi(kvs["retry"].c_str()) : 0);}}return 0;};return cli.send(&req, &resp);
}

测试Demo

 sse(HTTP_GET,"http://127.0.0.1:12900/sse", [](const std::string& sid, const std::string& sevent,const std::string& sdata, const unsigned int retry_ms) { printf("id:%s\r\nevent:%s\r\ndata:%s\r\nretry:%u\r\n\r\n",sid.c_str(),sevent.c_str(),sdata.c_str(),retry_ms);});