Skip to content

Commit 38663d5

Browse files
qimuyaqimuya
authored
add server http_sse_service
1 parent f3d9961 commit 38663d5

3 files changed

Lines changed: 312 additions & 0 deletions

File tree

‎trpc/server/http_sse/BUILD‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,3 +32,28 @@ cc_library(
3232
"//trpc/util/log:logging",
3333
],
3434
)
35+
36+
37+
# trpc/server/http_sse/BUILD
38+
licenses(["notice"])
39+
40+
package(default_visibility = ["//visibility:public"])
41+
42+
cc_library(
43+
name = "http_sse_service",
44+
srcs = ["http_sse_service.cc"],
45+
hdrs = ["http_sse_service.h"],
46+
deps = [
47+
"//trpc/codec/http:http_protocol",
48+
"//trpc/codec/http_sse:http_sse_proto_checker",
49+
"//trpc/util:deferred",
50+
"//trpc/util/log:logging",
51+
"//trpc/util:time",
52+
"//trpc/util/http:request",
53+
"//trpc/util/http:response",
54+
"//trpc/util/http:routes",
55+
"//trpc/server:service",
56+
"//trpc/util/http:http_handler_groups",
57+
],
58+
)
59+

‎trpc/server/http_sse/http_sse_service.cc‎

Lines changed: 233 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,233 @@
1+
// trpc/server/http_sse/http_sse_server.cc
2+
//
3+
// SSE-specialized HTTP service based on HttpService::HandleTransportMessage.
4+
// Minimal SSE-specific modifications: detect SSE requests and enable SSE headers + streaming.
5+
// Other logic (filters, timeout, dispatch, error handling) copied from HttpService.
6+
//
7+
8+
#include "trpc/server/http_sse/http_sse_service.h"
9+
#include "trpc/codec/http/http_protocol.h"
10+
#include "trpc/codec/http_sse/http_sse_proto_checker.h" // IsValidSseRequest
11+
#include "trpc/util/deferred.h"
12+
#include "trpc/util/log/logging.h"
13+
#include "trpc/util/time.h"
14+
15+
namespace trpc {
16+
17+
void HttpSseService::HandleTransportMessage(STransportReqMsg* recv, STransportRspMsg** send) noexcept {
18+
ServerContextPtr& context = recv->context;
19+
20+
// request / response protocol objects (same as HttpService)
21+
http::RequestPtr& req =
22+
static_cast<HttpRequestProtocol*>(context->GetRequestMsg().get())->request; // NOLINT: safe unchecked cast
23+
http::Response& rsp =
24+
static_cast<HttpResponseProtocol*>(context->GetResponseMsg().get())->response; // NOLINT: safe unchecked cast
25+
rsp.SetHeaderOnly(req->GetMethodType() == http::HEAD);
26+
27+
// attach request/response pointers to context for filters/handlers
28+
context->SetRequestData(req.get());
29+
context->SetResponseData(&rsp);
30+
31+
// set encode-type and function name, and compute real timeout
32+
context->SetReqEncodeType(MimeToEncodeType(req->GetHeader(http::kHeaderContentType)));
33+
context->SetFuncName(req->GetRouteUrl());
34+
context->SetRealTimeout();
35+
36+
// configure request stream default deadline (for blocking reads)
37+
auto& req_stream = req->GetStream();
38+
req_stream.SetDefaultDeadline(trpc::ReadSteadyClock() + std::chrono::milliseconds(context->GetTimeout()));
39+
40+
// route lookup
41+
std::string uri_path = trpc::http::NormalizeUrl(req->GetRouteUrlView());
42+
http::HandlerBase* handler = routes_.GetHandler(uri_path, req);
43+
44+
// If the handler is a stream handler, check SSE specifics and handle accordingly
45+
if (handler && handler->IsStream()) {
46+
// detect SSE request: Accept contains "text/event-stream" and method is GET
47+
bool is_sse = IsValidSseRequest(req.get());
48+
49+
if (is_sse) {
50+
TRPC_LOG_INFO("HttpSseService: detected SSE request for path=" << uri_path);
51+
52+
// Enable streaming on response
53+
rsp.EnableStream(context.get());
54+
55+
// Pre-set SSE headers on the response so handler sees them by default
56+
// These are the canonical SSE headers. Handler may override if needed.
57+
rsp.SetMimeType("text/event-stream");
58+
rsp.SetHeader("Cache-Control", "no-cache");
59+
rsp.SetHeader("Connection", "keep-alive");
60+
// allow CORS by default (matches codec helper earlier)
61+
//rsp.SetHeader("Access-Control-Allow-Origin", "*");
62+
//rsp.SetHeader("Access-Control-Allow-Headers", "Cache-Control");
63+
// use chunked transfer encoding for streaming responses
64+
rsp.SetHeader(http::kHeaderTransferEncoding, http::kTransferEncodingChunked);
65+
66+
// Call handler (streaming mode). Handler should write events via rsp.GetStream() or use your SseStreamWriter.
67+
Handle(uri_path, handler, context, req, rsp, send);
68+
69+
// After handler returns, close stream and check possible stream-reset
70+
try {
71+
rsp.GetStream().Close();
72+
} catch (...) {
73+
}
74+
75+
if (context->GetStatus().StreamRst()) {
76+
TRPC_FMT_TRACE("{} {}, error: {}", req->GetMethod(), req->GetRouteUrlView(), context->GetStatus().ToString());
77+
context->CloseConnection();
78+
}
79+
80+
// Release request stream
81+
req_stream.Close();
82+
return;
83+
}
84+
// else: it's a stream handler but not SSE -> fall through to normal stream handling below
85+
}
86+
87+
// Non-stream or non-SSE stream: follow original HttpService behavior
88+
if (!handler || !handler->IsStream()) {
89+
Status status = req_stream.AppendToRequest(req->GetMaxBodySize());
90+
if (status.OK()) {
91+
Handle(uri_path, handler, context, req, rsp, send);
92+
} else {
93+
HandleError(context, req, rsp, status);
94+
}
95+
} else { // stream handler but not SSE
96+
rsp.EnableStream(context.get());
97+
Handle(uri_path, handler, context, req, rsp, send);
98+
rsp.GetStream().Close();
99+
if (context->GetStatus().StreamRst()) {
100+
TRPC_FMT_TRACE("{} {}, error: {}", req->GetMethod(), req->GetRouteUrlView(), context->GetStatus().ToString());
101+
context->CloseConnection();
102+
}
103+
}
104+
105+
// Release request stream
106+
req_stream.Close();
107+
}
108+
109+
Status HttpSseService::Dispatch(const std::string& path, http::HandlerBase* handler, ServerContextPtr& context,
110+
http::RequestPtr& req, http::Response& rsp) {
111+
return routes_.Handle(path, handler, context, req, rsp);
112+
}
113+
114+
void HttpSseService::Handle(const std::string& path, http::HandlerBase* handler, ServerContextPtr& context,
115+
http::RequestPtr& req, http::Response& rsp, STransportRspMsg** send) {
116+
Deferred _([&context]() {
117+
context->SetRequestData(nullptr);
118+
context->SetResponseData(nullptr);
119+
});
120+
121+
// For some monitor plugins.
122+
auto msg_filter_status = GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_POST_RECV_MSG, context);
123+
// For some tracing or log replay plugins.
124+
auto rpc_filter_status = GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_PRE_RPC_INVOKE, context);
125+
126+
// Rejected by filters.
127+
if (TRPC_UNLIKELY(msg_filter_status == FilterStatus::REJECT || rpc_filter_status == FilterStatus::REJECT)) {
128+
// For some tracing or log replay plugins.
129+
GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_POST_RPC_INVOKE, context);
130+
// For some monitor plugins.
131+
GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_PRE_SEND_MSG, context);
132+
133+
auto& status = context->GetStatus();
134+
status.SetFrameworkRetCode(GetDefaultServerRetCode(codec::ServerRetCode::INVOKE_UNKNOW_ERROR));
135+
status.SetErrorMessage("filter reject");
136+
http::Response reject_rsp;
137+
*send = trpc::object_pool::New<STransportRspMsg>();
138+
(*send)->context = context;
139+
reject_rsp.GenerateExceptionReply(http::ResponseStatus::kForbidden, req->GetVersion(), "request reject");
140+
std::move(reject_rsp).SerializeToString((*send)->buffer);
141+
return;
142+
}
143+
144+
// Timeout.
145+
CheckTimeout(context);
146+
if (!context->GetStatus().OK()) {
147+
// For some tracing or log replay plugins.
148+
GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_POST_RPC_INVOKE, context);
149+
// For some monitor plugins.
150+
GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_PRE_SEND_MSG, context);
151+
152+
TRPC_LOG_ERROR("CheckTimeout failed, ip: " << context->GetIp()
153+
<< ", service queue_timeout: " << GetServiceAdapterOption().queue_timeout
154+
<< ", client timeout:" << context->GetTimeout());
155+
156+
auto& status = context->GetStatus();
157+
status.SetFrameworkRetCode(GetDefaultServerRetCode(codec::ServerRetCode::TIMEOUT_ERROR));
158+
status.SetErrorMessage("client timeout");
159+
http::Response timeout_rsp;
160+
static const std::string request_timeout_ex = http::JsonException(http::RequestTimeout()).ToJson();
161+
timeout_rsp.GenerateExceptionReply(http::ResponseStatus::kGatewayTimeout, req->GetVersion(), request_timeout_ex);
162+
NoncontiguousBuffer buffer;
163+
std::move(timeout_rsp).SerializeToString(buffer);
164+
context->SendResponse(std::move(buffer));
165+
context->CloseConnection();
166+
return;
167+
}
168+
169+
// Runs user handler.
170+
auto status = Dispatch(path, handler, context, req, rsp);
171+
if (context->IsResponse()) {
172+
context->SetStatus(std::move(status));
173+
// For some tracing or log replay plugins.
174+
GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_POST_RPC_INVOKE, context);
175+
// For some monitor plugins.
176+
GetFilterController().RunMessageServerFilters(FilterPoint::SERVER_PRE_SEND_MSG, context);
177+
178+
*send = trpc::object_pool::New<STransportRspMsg>();
179+
(*send)->context = context;
180+
181+
if (context->CheckHandleTimeout()) {
182+
http::Response timeout_rsp;
183+
static const std::string request_handle_timeout_ex =
184+
http::JsonException(http::RequestTimeout("Request Handle Timeout")).ToJson();
185+
timeout_rsp.GenerateExceptionReply(http::ResponseStatus::kGatewayTimeout, req->GetVersion(),
186+
request_handle_timeout_ex);
187+
std::move(timeout_rsp).SerializeToString((*send)->buffer);
188+
} else {
189+
std::move(rsp).SerializeToString((*send)->buffer);
190+
}
191+
}
192+
}
193+
194+
void HttpSseService::HandleError(ServerContextPtr& context, http::RequestPtr& req, http::Response& rsp,
195+
const Status& status) {
196+
req->GetStream().Close(stream::HttpReadStream::ReadState::kErrorBit);
197+
if (status.GetFrameworkRetCode() == stream::kStreamStatusServerReadTimeout.GetFrameworkRetCode()) {
198+
rsp.GenerateExceptionReply(http::ResponseStatus::kRequestTimeout, req->GetVersion(), "request timeout");
199+
} else if (status.GetFrameworkRetCode() == stream::kStreamStatusServerMessageExceedLimit.GetFrameworkRetCode()) {
200+
rsp.GenerateExceptionReply(http::ResponseStatus::kRequestEntityTooLarge, req->GetVersion(),
201+
"request entity too large");
202+
} else { // unexpected error
203+
rsp.GenerateExceptionReply(http::ResponseStatus::kInternalServerError, req->GetVersion(), "unknown error");
204+
}
205+
TRPC_LOG_DEBUG("HTTP read error, ip: " << context->GetIp() << ", status: " << status.ToString());
206+
NoncontiguousBuffer buffer;
207+
std::move(rsp).SerializeToString(buffer);
208+
context->SendResponse(std::move(buffer));
209+
context->SetRequestData(nullptr);
210+
context->SetResponseData(nullptr);
211+
context->CloseConnection();
212+
}
213+
214+
void HttpSseService::CheckTimeout(const ServerContextPtr& context) {
215+
uint64_t now_ms = static_cast<int64_t>(trpc::time::GetMilliSeconds());
216+
uint64_t timeout = std::min(GetServiceAdapterOption().queue_timeout, context->GetTimeout());
217+
218+
if (context->GetRecvTimestamp() + timeout <= now_ms) {
219+
bool use_queue_timeout = GetServiceAdapterOption().queue_timeout < context->GetTimeout();
220+
if (!use_queue_timeout && context->IsUseFullLinkTimeout()) {
221+
context->GetStatus().SetFrameworkRetCode(
222+
context->GetServerCodec()->GetProtocolRetCode(trpc::codec::ServerRetCode::FULL_LINK_TIMEOUT_ERROR));
223+
context->GetStatus().SetErrorMessage("request full-link timeout.");
224+
} else {
225+
context->GetStatus().SetFrameworkRetCode(
226+
context->GetServerCodec()->GetProtocolRetCode(trpc::codec::ServerRetCode::TIMEOUT_ERROR));
227+
context->GetStatus().SetErrorMessage("request timeout.");
228+
}
229+
}
230+
}
231+
232+
} // namespace trpc
233+

‎trpc/server/http_sse/http_sse_service.h‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
// trpc/server/http_sse/http_sse_service.h
2+
//
3+
// SSE-specialized HTTP service header.
4+
5+
#pragma once
6+
7+
#include <functional>
8+
#include <string>
9+
10+
#include "trpc/server/server_context.h"
11+
#include "trpc/server/service.h"
12+
#include "trpc/util/http/http_handler_groups.h"
13+
#include "trpc/util/http/request.h"
14+
#include "trpc/util/http/response.h"
15+
#include "trpc/util/http/routes.h"
16+
17+
namespace trpc {
18+
19+
/// @brief SSE-specialized HTTP service. Minimal modifications compared to HttpService:
20+
/// detects SSE requests and enables streaming + SSE headers for handlers that are stream-capable.
21+
class HttpSseService : public Service {
22+
public:
23+
/// Process transport message (override Service API).
24+
void HandleTransportMessage(STransportReqMsg* recv, STransportRspMsg** send) noexcept override;
25+
26+
/// Set routes using a routes setter function (same interface as HttpService).
27+
void SetRoutes(const std::function<void(http::Routes& r)>& func) { func(routes_); }
28+
29+
/// Set routes using handler groups helper.
30+
void SetRoutes(const std::function<void(http::HttpHandlerGroups r)>& func) { func(http::HttpHandlerGroups(routes_)); }
31+
32+
/// Gets routes object (for tests / introspection).
33+
http::HttpRoutes& GetRoutes() { return routes_; }
34+
35+
protected:
36+
/// Dispatch request to handler (wrapper around routes_.Handle).
37+
Status Dispatch(const std::string& path, http::HandlerBase* handler, ServerContextPtr& context, http::RequestPtr& req,
38+
http::Response& rsp);
39+
40+
private:
41+
/// Internal handler calling logic copied/adapted from HttpService.
42+
void Handle(const std::string& path, http::HandlerBase* handler, ServerContextPtr& context, http::RequestPtr& req,
43+
http::Response& rsp, STransportRspMsg** send);
44+
45+
static void HandleError(ServerContextPtr& context, http::RequestPtr& req, http::Response& rsp, const Status& status);
46+
47+
void CheckTimeout(const ServerContextPtr& context);
48+
49+
protected:
50+
http::Routes routes_;
51+
};
52+
53+
} // namespace trpc
54+

0 commit comments

Comments
 (0)