feat: add optional JSON-RPC ping keepalive support - #99
Open
bytethm wants to merge 1 commit into
Open
Conversation
Add optional JSON-RPC ping keepalive mechanism for SSE and Streamable HTTP connections, complementing the existing SSE comment keepalive. Key changes: - Add ping configuration options (WithServerPingKeepAlive, WithPingInterval, WithPingTimeout) - Implement PingSession and PingAllSessions methods for SSE Server - Add automatic ping request handling in SSE Client - Support ping keepalive for Streamable HTTP GET SSE connections - Add comprehensive tests for backward compatibility and new features - Update documentation with configuration examples The new ping keepalive is disabled by default to maintain backward compatibility. Users can opt-in by using WithServerPingKeepAlive(true). Ping failures are logged but do not disconnect sessions, ensuring robust operation in unstable network conditions.
Reviewer's Guide在现有 SSE 注释式 keepalive 的基础上，新增可选的基于 JSON-RPC ping 的 keepalive 机制；通过顶层 Server API 将其配置传递到 SSE 服务器和可流式的 HTTP 处理器中；实现服务端的 ping 循环和基于会话的 ping 辅助方法；更新 SSE 客户端以响应 ping 请求；并为新行为及向后兼容性编写文档和测试。 基于 SSE 的 JSON-RPC ping keepalive 时序图sequenceDiagram
participant SSEServer
participant PingLoop
participant Client as sseClientTransport
SSEServer->>SSEServer: handleSSE
alt pingEnabled
SSEServer->>SSEServer: startPingLoop
activate PingLoop
loop every pingInterval
PingLoop->>SSEServer: PingAllSessions(ctx)
activate SSEServer
SSEServer->>SSEServer: Range over sessions
par for each sessionID
SSEServer->>SSEServer: PingSession(pingCtx, sessionID)
activate SSEServer
SSEServer->>Client: SendRequest(ctx, sessionID, JSONRPC ping)
activate Client
Client->>Client: handleIncomingRequest(data)
Client->>Client: handlePingRequest(request)
Client-->>SSEServer: JSONRPCResponse (empty result)
deactivate Client
SSEServer->>SSEServer: verify result not nil
deactivate SSEServer
and for each sessionID
end
SSEServer-->>PingLoop: all pings done or timeout
deactivate SSEServer
end
deactivate PingLoop
else ping disabled
SSEServer->>SSEServer: no ping loop started
end
SSEServer->>SSEServer: Close
SSEServer->>PingLoop: close stop channel
PingLoop-->>SSEServer: ping loop exits
流式 HTTP GET SSE 的 JSON-RPC ping keepalive 时序图sequenceDiagram
participant HTTPHandler as httpServerHandler
participant PingLoop
participant Client as sseClientTransport
HTTPHandler->>HTTPHandler: handleGet
alt pingEnabled
HTTPHandler->>HTTPHandler: pingStartMu.Lock
alt !pingStarted
HTTPHandler->>HTTPHandler: pingStarted = true
HTTPHandler->>HTTPHandler: pingStartMu.Unlock
HTTPHandler->>HTTPHandler: startPingLoop
activate PingLoop
else ping already started
HTTPHandler->>HTTPHandler: pingStartMu.Unlock
end
end
alt keepAliveEnabled
HTTPHandler->>HTTPHandler: handleGetSSECommentKeepAlive
loop every keepAliveInterval
HTTPHandler->>Client: send ": keepalive" SSE comment
end
end
loop every pingInterval
PingLoop->>HTTPHandler: PingAllSessions(ctx)
activate HTTPHandler
HTTPHandler->>HTTPHandler: collect getSSEConnections sessionIDs
par for each sessionID
HTTPHandler->>HTTPHandler: PingSession(pingCtx, sessionID)
activate HTTPHandler
HTTPHandler->>Client: SendRequest(ctx, sessionID, JSONRPC ping)
activate Client
Client->>Client: handleIncomingRequest(data)
Client->>Client: handlePingRequest(request)
Client-->>HTTPHandler: JSONRPCResponse (empty result)
deactivate Client
HTTPHandler->>HTTPHandler: verify result not nil
deactivate HTTPHandler
and for each sessionID
end
HTTPHandler-->>PingLoop: all pings done or timeout
deactivate HTTPHandler
end
HTTPHandler->>HTTPHandler: Close
HTTPHandler->>PingLoop: close stop channel
PingLoop-->>HTTPHandler: ping loop exits
服务器、SSE 服务器、HTTP 处理器和 SSE 客户端 ping keepalive 的更新类图classDiagram
class Server {
+serverConfig config
+NewServer(name string, version string, options ...ServerOption)
+initComponents()
}
class serverConfig {
+bool postSSEEnabled
+bool getSSEEnabled
+int notificationBufferSize
+bool keepAliveEnabled
+time_Duration keepAliveInterval
+bool pingEnabled
+time_Duration pingInterval
+time_Duration pingTimeout
}
class SSEServer {
+bool keepAlive
+time_Duration keepAliveInterval
+bool pingEnabled
+time_Duration pingInterval
+time_Duration pingTimeout
+bool pingStarted
+sync_Mutex pingStartMu
+chan_struct stop
+sync_Once stopOnce
+Logger logger
+atomic_Int64 requestID
+PingSession(ctx context_Context, sessionID string) error
+PingAllSessions(ctx context_Context)
+startPingLoop()
+Close() error
}
class httpServerHandler {
+requestHandler handler
+string serverPath
+responseManager responseManager
+bool keepAliveEnabled
+time_Duration keepAliveInterval
+bool pingEnabled
+time_Duration pingInterval
+time_Duration pingTimeout
+bool pingStarted
+sync_Mutex pingStartMu
+chan_struct stop
+sync_Once stopOnce
+handleGet(ctx context_Context, w http_ResponseWriter, r *http_Request)
+handleGetSSECommentKeepAlive(ctx context_Context, conn *getSSEConnection, sessionID string)
+PingSession(ctx context_Context, sessionID string) error
+PingAllSessions(ctx context_Context)
+startPingLoop()
+Close() error
}
class sseClientTransport {
+Logger logger
+handleIncomingRequest(data string)
+handleRootsListRequest(request *JSONRPCRequest)
+handlePingRequest(request *JSONRPCRequest)
+sendResponseMessage(response *JSONRPCResponse)
+sendErrorResponse(request *JSONRPCRequest, code int, message string)
}
class ServerOption {
}
class SSEOption {
}
%% Server options wiring keepalive and ping configuration
class WithServerKeepAlive {
+WithServerKeepAlive(enabled bool) ServerOption
}
class WithServerKeepAliveInterval {
+WithServerKeepAliveInterval(interval time_Duration) ServerOption
}
class WithServerPingKeepAlive {
+WithServerPingKeepAlive(enabled bool) ServerOption
}
class WithServerPingInterval {
+WithServerPingInterval(interval time_Duration) ServerOption
}
class WithServerPingTimeout {
+WithServerPingTimeout(timeout time_Duration) ServerOption
}
%% SSE server options
class WithKeepAliveInterval {
+WithKeepAliveInterval(interval time_Duration) SSEOption
}
class WithPingKeepAlive {
+WithPingKeepAlive(enabled bool) SSEOption
}
class WithPingInterval {
+WithPingInterval(interval time_Duration) SSEOption
}
class WithPingTimeout {
+WithPingTimeout(timeout time_Duration) SSEOption
}
%% HTTP transport internal options
class withKeepAliveConfig {
+withKeepAliveConfig(enabled bool, interval time_Duration) func_httpServerHandler
}
class withPingConfig {
+withPingConfig(enabled bool, interval time_Duration, timeout time_Duration) func_httpServerHandler
}
Server *-- serverConfig
Server --> SSEServer : uses
Server --> httpServerHandler : uses
ServerOption <|-- WithServerKeepAlive
ServerOption <|-- WithServerKeepAliveInterval
ServerOption <|-- WithServerPingKeepAlive
ServerOption <|-- WithServerPingInterval
ServerOption <|-- WithServerPingTimeout
SSEOption <|-- WithKeepAliveInterval
SSEOption <|-- WithPingKeepAlive
SSEOption <|-- WithPingInterval
SSEOption <|-- WithPingTimeout
httpServerHandler ..> withKeepAliveConfig : configured_by
httpServerHandler ..> withPingConfig : configured_by
SSEServer ..> sseClientTransport : sends_JSONRPC_ping
httpServerHandler ..> sseClientTransport : sends_JSONRPC_ping
sseClientTransport ..> SSEServer : responds_to_ping
sseClientTransport ..> httpServerHandler : responds_to_ping
文件级变更
Tips and commandsInteracting with Sourcery
Customizing Your Experience访问你的 dashboard 来：
Getting HelpOriginal review guide in EnglishReviewer's GuideAdds an optional JSON-RPC ping-based keepalive alongside existing SSE comment keepalive, wires its configuration through the top-level Server API into both the SSE server and streamable HTTP handler, implements server-side ping loops and per-session ping helpers, updates the SSE client to answer ping requests, and documents and tests the new behavior and backward compatibility. Sequence diagram for JSON-RPC ping keepalive over SSEsequenceDiagram
participant SSEServer
participant PingLoop
participant Client as sseClientTransport
SSEServer->>SSEServer: handleSSE
alt pingEnabled
SSEServer->>SSEServer: startPingLoop
activate PingLoop
loop every pingInterval
PingLoop->>SSEServer: PingAllSessions(ctx)
activate SSEServer
SSEServer->>SSEServer: Range over sessions
par for each sessionID
SSEServer->>SSEServer: PingSession(pingCtx, sessionID)
activate SSEServer
SSEServer->>Client: SendRequest(ctx, sessionID, JSONRPC ping)
activate Client
Client->>Client: handleIncomingRequest(data)
Client->>Client: handlePingRequest(request)
Client-->>SSEServer: JSONRPCResponse (empty result)
deactivate Client
SSEServer->>SSEServer: verify result not nil
deactivate SSEServer
and for each sessionID
end
SSEServer-->>PingLoop: all pings done or timeout
deactivate SSEServer
end
deactivate PingLoop
else ping disabled
SSEServer->>SSEServer: no ping loop started
end
SSEServer->>SSEServer: Close
SSEServer->>PingLoop: close stop channel
PingLoop-->>SSEServer: ping loop exits
Sequence diagram for JSON-RPC ping keepalive for Streamable HTTP GET SSEsequenceDiagram
participant HTTPHandler as httpServerHandler
participant PingLoop
participant Client as sseClientTransport
HTTPHandler->>HTTPHandler: handleGet
alt pingEnabled
HTTPHandler->>HTTPHandler: pingStartMu.Lock
alt !pingStarted
HTTPHandler->>HTTPHandler: pingStarted = true
HTTPHandler->>HTTPHandler: pingStartMu.Unlock
HTTPHandler->>HTTPHandler: startPingLoop
activate PingLoop
else ping already started
HTTPHandler->>HTTPHandler: pingStartMu.Unlock
end
end
alt keepAliveEnabled
HTTPHandler->>HTTPHandler: handleGetSSECommentKeepAlive
loop every keepAliveInterval
HTTPHandler->>Client: send ": keepalive" SSE comment
end
end
loop every pingInterval
PingLoop->>HTTPHandler: PingAllSessions(ctx)
activate HTTPHandler
HTTPHandler->>HTTPHandler: collect getSSEConnections sessionIDs
par for each sessionID
HTTPHandler->>HTTPHandler: PingSession(pingCtx, sessionID)
activate HTTPHandler
HTTPHandler->>Client: SendRequest(ctx, sessionID, JSONRPC ping)
activate Client
Client->>Client: handleIncomingRequest(data)
Client->>Client: handlePingRequest(request)
Client-->>HTTPHandler: JSONRPCResponse (empty result)
deactivate Client
HTTPHandler->>HTTPHandler: verify result not nil
deactivate HTTPHandler
and for each sessionID
end
HTTPHandler-->>PingLoop: all pings done or timeout
deactivate HTTPHandler
end
HTTPHandler->>HTTPHandler: Close
HTTPHandler->>PingLoop: close stop channel
PingLoop-->>HTTPHandler: ping loop exits
Updated class diagram for server, SSE server, HTTP handler, and SSE client ping keepaliveclassDiagram
class Server {
+serverConfig config
+NewServer(name string, version string, options ...ServerOption)
+initComponents()
}
class serverConfig {
+bool postSSEEnabled
+bool getSSEEnabled
+int notificationBufferSize
+bool keepAliveEnabled
+time_Duration keepAliveInterval
+bool pingEnabled
+time_Duration pingInterval
+time_Duration pingTimeout
}
class SSEServer {
+bool keepAlive
+time_Duration keepAliveInterval
+bool pingEnabled
+time_Duration pingInterval
+time_Duration pingTimeout
+bool pingStarted
+sync_Mutex pingStartMu
+chan_struct stop
+sync_Once stopOnce
+Logger logger
+atomic_Int64 requestID
+PingSession(ctx context_Context, sessionID string) error
+PingAllSessions(ctx context_Context)
+startPingLoop()
+Close() error
}
class httpServerHandler {
+requestHandler handler
+string serverPath
+responseManager responseManager
+bool keepAliveEnabled
+time_Duration keepAliveInterval
+bool pingEnabled
+time_Duration pingInterval
+time_Duration pingTimeout
+bool pingStarted
+sync_Mutex pingStartMu
+chan_struct stop
+sync_Once stopOnce
+handleGet(ctx context_Context, w http_ResponseWriter, r *http_Request)
+handleGetSSECommentKeepAlive(ctx context_Context, conn *getSSEConnection, sessionID string)
+PingSession(ctx context_Context, sessionID string) error
+PingAllSessions(ctx context_Context)
+startPingLoop()
+Close() error
}
class sseClientTransport {
+Logger logger
+handleIncomingRequest(data string)
+handleRootsListRequest(request *JSONRPCRequest)
+handlePingRequest(request *JSONRPCRequest)
+sendResponseMessage(response *JSONRPCResponse)
+sendErrorResponse(request *JSONRPCRequest, code int, message string)
}
class ServerOption {
}
class SSEOption {
}
%% Server options wiring keepalive and ping configuration
class WithServerKeepAlive {
+WithServerKeepAlive(enabled bool) ServerOption
}
class WithServerKeepAliveInterval {
+WithServerKeepAliveInterval(interval time_Duration) ServerOption
}
class WithServerPingKeepAlive {
+WithServerPingKeepAlive(enabled bool) ServerOption
}
class WithServerPingInterval {
+WithServerPingInterval(interval time_Duration) ServerOption
}
class WithServerPingTimeout {
+WithServerPingTimeout(timeout time_Duration) ServerOption
}
%% SSE server options
class WithKeepAliveInterval {
+WithKeepAliveInterval(interval time_Duration) SSEOption
}
class WithPingKeepAlive {
+WithPingKeepAlive(enabled bool) SSEOption
}
class WithPingInterval {
+WithPingInterval(interval time_Duration) SSEOption
}
class WithPingTimeout {
+WithPingTimeout(timeout time_Duration) SSEOption
}
%% HTTP transport internal options
class withKeepAliveConfig {
+withKeepAliveConfig(enabled bool, interval time_Duration) func_httpServerHandler
}
class withPingConfig {
+withPingConfig(enabled bool, interval time_Duration, timeout time_Duration) func_httpServerHandler
}
Server *-- serverConfig
Server --> SSEServer : uses
Server --> httpServerHandler : uses
ServerOption <|-- WithServerKeepAlive
ServerOption <|-- WithServerKeepAliveInterval
ServerOption <|-- WithServerPingKeepAlive
ServerOption <|-- WithServerPingInterval
ServerOption <|-- WithServerPingTimeout
SSEOption <|-- WithKeepAliveInterval
SSEOption <|-- WithPingKeepAlive
SSEOption <|-- WithPingInterval
SSEOption <|-- WithPingTimeout
httpServerHandler ..> withKeepAliveConfig : configured_by
httpServerHandler ..> withPingConfig : configured_by
SSEServer ..> sseClientTransport : sends_JSONRPC_ping
httpServerHandler ..> sseClientTransport : sends_JSONRPC_ping
sseClientTransport ..> SSEServer : responds_to_ping
sseClientTransport ..> httpServerHandler : responds_to_ping
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - 我发现了 1 个问题，并给出了一些整体性的反馈：
SSEServer和httpServerHandler中的 ping keepalive 逻辑几乎完全相同（字段、PingSession、PingAllSessions、startPingLoop、Close）；可以考虑抽取一个共享的 helper 或类型来减少重复代码，并更容易保持行为一致。- 在
handleGetSSECommentKeepAlive中，对 SSE 连接的写操作（fmt.Fprintf和Flush）忽略了错误；如果写操作重复失败（例如客户端已断开但尚未被检测到），你可能会希望记录日志和/或提前停止 keepalive 循环，以避免噪声日志或冗余工作。
给 AI 代理的提示
Please address the comments from this code review:
## Overall Comments
- The ping keepalive logic in `SSEServer` and `httpServerHandler` is nearly identical (fields, `PingSession`, `PingAllSessions`, `startPingLoop`, `Close`); consider extracting a shared helper or type to reduce duplication and keep the behavior in sync more easily.
- In `handleGetSSECommentKeepAlive`, writes to the SSE connection (`fmt.Fprintf` and `Flush`) ignore errors; if a write fails repeatedly (e.g., client gone but not yet detected), you may want to log and/or stop the keepalive loop early to avoid noisy logs or redundant work.
## Individual Comments
### Comment 1
<location> `streamable_server.go:698-707` </location>
<code_context>
+// handleGetSSECommentKeepAlive handles SSE comment keepalive for GET SSE connections.
+// It sends SSE comment lines at the configured interval to keep the connection alive.
+func (h *httpServerHandler) handleGetSSECommentKeepAlive(ctx context.Context, conn *getSSEConnection, sessionID string) {
+ if h.keepAliveInterval <= 0 {
+ return
+ }
+
+ ticker := time.NewTicker(h.keepAliveInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ticker.C:
+ conn.writeLock.Lock()
+ // Send SSE comment
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 可以考虑在 SSE keepalive 循环中处理写入/刷新错误，以避免静默忽略已断开的连接。
在 `handleGetSSECommentKeepAlive` 中，`Fprintf` 和 `Flush` 的调用没有检查返回的错误。如果客户端断开连接或 writer 失败，keepalive goroutine 会一直运行到上下文被取消，而不是尽快退出。请在写入/刷新失败时处理这些错误（并可选择记录日志），以便循环在写入/刷新失败时能提前终止。
```suggestion
case <-ticker.C:
conn.writeLock.Lock()
// Send SSE comment
if _, err := fmt.Fprintf(conn.writer, ": keepalive\n\n"); err != nil {
conn.writeLock.Unlock()
if h.logger != nil {
h.logger.Errorf("Failed to send keepalive comment to GET SSE session %s: %v", sessionID, err)
}
// Exit keepalive loop on write failure
return
}
conn.flusher.Flush()
conn.writeLock.Unlock()
if h.logger != nil {
h.logger.Debugf("Sent keepalive comment to GET SSE session: %s", sessionID)
}
```
</issue_to_address>帮我变得更有用！请对每条评论点 👍 或 👎，我会根据你的反馈来改进 Review 质量。
Original comment in English
Hey - I've found 1 issue, and left some high level feedback:
- The ping keepalive logic in
SSEServerandhttpServerHandleris nearly identical (fields,PingSession,PingAllSessions,startPingLoop,Close); consider extracting a shared helper or type to reduce duplication and keep the behavior in sync more easily. - In
handleGetSSECommentKeepAlive, writes to the SSE connection (fmt.FprintfandFlush) ignore errors; if a write fails repeatedly (e.g., client gone but not yet detected), you may want to log and/or stop the keepalive loop early to avoid noisy logs or redundant work.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The ping keepalive logic in `SSEServer` and `httpServerHandler` is nearly identical (fields, `PingSession`, `PingAllSessions`, `startPingLoop`, `Close`); consider extracting a shared helper or type to reduce duplication and keep the behavior in sync more easily.
- In `handleGetSSECommentKeepAlive`, writes to the SSE connection (`fmt.Fprintf` and `Flush`) ignore errors; if a write fails repeatedly (e.g., client gone but not yet detected), you may want to log and/or stop the keepalive loop early to avoid noisy logs or redundant work.
## Individual Comments
### Comment 1
<location> `streamable_server.go:698-707` </location>
<code_context>
+// handleGetSSECommentKeepAlive handles SSE comment keepalive for GET SSE connections.
+// It sends SSE comment lines at the configured interval to keep the connection alive.
+func (h *httpServerHandler) handleGetSSECommentKeepAlive(ctx context.Context, conn *getSSEConnection, sessionID string) {
+ if h.keepAliveInterval <= 0 {
+ return
+ }
+
+ ticker := time.NewTicker(h.keepAliveInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ticker.C:
+ conn.writeLock.Lock()
+ // Send SSE comment
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider handling write/flush errors in the SSE keepalive loop to avoid silently ignoring broken connections.
In `handleGetSSECommentKeepAlive`, the `Fprintf` and `Flush` calls are executed without checking their return errors. If the client disconnects or the writer fails, the keepalive goroutine will keep running until context cancellation instead of exiting promptly. Please handle these errors (and optionally log them) so the loop can terminate early on write/flush failure.
```suggestion
case <-ticker.C:
conn.writeLock.Lock()
// Send SSE comment
if _, err := fmt.Fprintf(conn.writer, ": keepalive\n\n"); err != nil {
conn.writeLock.Unlock()
if h.logger != nil {
h.logger.Errorf("Failed to send keepalive comment to GET SSE session %s: %v", sessionID, err)
}
// Exit keepalive loop on write failure
return
}
conn.flusher.Flush()
conn.writeLock.Unlock()
if h.logger != nil {
h.logger.Debugf("Sent keepalive comment to GET SSE session: %s", sessionID)
}
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+698
to
+707
| case <-ticker.C: | ||
| conn.writeLock.Lock() | ||
| // Send SSE comment | ||
| fmt.Fprintf(conn.writer, ": keepalive\n\n") | ||
| conn.flusher.Flush() | ||
| conn.writeLock.Unlock() | ||
|
|
||
| if h.logger != nil { | ||
| h.logger.Debugf("Sent keepalive comment to GET SSE session: %s", sessionID) | ||
| } |
There was a problem hiding this comment.
suggestion (bug_risk): 可以考虑在 SSE keepalive 循环中处理写入/刷新错误，以避免静默忽略已断开的连接。
在 handleGetSSECommentKeepAlive 中，Fprintf 和 Flush 的调用没有检查返回的错误。如果客户端断开连接或 writer 失败，keepalive goroutine 会一直运行到上下文被取消，而不是尽快退出。请在写入/刷新失败时处理这些错误（并可选择记录日志），以便循环在写入/刷新失败时能提前终止。
Suggested change
| case <-ticker.C: | |
| conn.writeLock.Lock() | |
| // Send SSE comment | |
| fmt.Fprintf(conn.writer, ": keepalive\n\n") | |
| conn.flusher.Flush() | |
| conn.writeLock.Unlock() | |
| if h.logger != nil { | |
| h.logger.Debugf("Sent keepalive comment to GET SSE session: %s", sessionID) | |
| } | |
| case <-ticker.C: | |
| conn.writeLock.Lock() | |
| // Send SSE comment | |
| if _, err := fmt.Fprintf(conn.writer, ": keepalive\n\n"); err != nil { | |
| conn.writeLock.Unlock() | |
| if h.logger != nil { | |
| h.logger.Errorf("Failed to send keepalive comment to GET SSE session %s: %v", sessionID, err) | |
| } | |
| // Exit keepalive loop on write failure | |
| return | |
| } | |
| conn.flusher.Flush() | |
| conn.writeLock.Unlock() | |
| if h.logger != nil { | |
| h.logger.Debugf("Sent keepalive comment to GET SSE session: %s", sessionID) | |
| } |
Original comment in English
suggestion (bug_risk): Consider handling write/flush errors in the SSE keepalive loop to avoid silently ignoring broken connections.
In handleGetSSECommentKeepAlive, the Fprintf and Flush calls are executed without checking their return errors. If the client disconnects or the writer fails, the keepalive goroutine will keep running until context cancellation instead of exiting promptly. Please handle these errors (and optionally log them) so the loop can terminate early on write/flush failure.
Suggested change
| case <-ticker.C: | |
| conn.writeLock.Lock() | |
| // Send SSE comment | |
| fmt.Fprintf(conn.writer, ": keepalive\n\n") | |
| conn.flusher.Flush() | |
| conn.writeLock.Unlock() | |
| if h.logger != nil { | |
| h.logger.Debugf("Sent keepalive comment to GET SSE session: %s", sessionID) | |
| } | |
| case <-ticker.C: | |
| conn.writeLock.Lock() | |
| // Send SSE comment | |
| if _, err := fmt.Fprintf(conn.writer, ": keepalive\n\n"); err != nil { | |
| conn.writeLock.Unlock() | |
| if h.logger != nil { | |
| h.logger.Errorf("Failed to send keepalive comment to GET SSE session %s: %v", sessionID, err) | |
| } | |
| // Exit keepalive loop on write failure | |
| return | |
| } | |
| conn.flusher.Flush() | |
| conn.writeLock.Unlock() | |
| if h.logger != nil { | |
| h.logger.Debugf("Sent keepalive comment to GET SSE session: %s", sessionID) | |
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Add optional JSON-RPC ping keepalive mechanism for SSE and Streamable HTTP connections, complementing the existing SSE comment keepalive.
Key changes:
The new ping keepalive is disabled by default to maintain backward compatibility. Users can opt-in by using WithServerPingKeepAlive(true).
Ping failures are logged but do not disconnect sessions, ensuring robust operation in unstable network conditions.
由 Sourcery 提供的摘要
在现有的 SSE 注释（comment）保活机制基础上，为核心 SSE 和可流式（Streamable）HTTP 传输新增可配置的基于 JSON-RPC ping 的保活机制，同时为保持向后兼容，默认仍关闭 ping。
新功能：
PingSession、PingAllSessions以及自动 ping 循环。增强：
文档：
测试：
Original summary in English
Summary by Sourcery
Add configurable JSON-RPC ping-based keepalive alongside existing SSE comment keepalive for both core SSE and Streamable HTTP transports, while keeping ping disabled by default for backward compatibility.
New Features:
Enhancements:
Documentation:
Tests: