Fix: Skip non-message SSE events in processStream (#637)
Fixes #636 ## Problem The StreamableClientTransport tries to JSON-decode ALL SSE events, including ping events sent by servers for keep-alive purposes. This causes errors like 'invalid character 'p' looking for beginning of value' when encountering non-JSON event data. ## Solution Modified processStream to check evt.Name and skip events that are not 'message' events (or unnamed events, which default to 'message' per SSE spec). ## Changes - Added event name check in processStream before attempting JSON decode - Added test case for ping event filtering - Added test for scanning multiple events including ping events ## Testing - All existing tests pass - New tests verify ping events are properly filtered - Tested with DeepWiki MCP server which sends ping events
This commit is contained in:
parent
faddd764aa
commit
9ef2b2742e
2 changed files with 156 additions and 0 deletions
|
|
@ -1656,6 +1656,12 @@ func (c *streamableClientConn) processStream(requestSummary string, resp *http.R
|
||||||
lastEventID = evt.ID
|
lastEventID = evt.ID
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Skip non-message events (e.g., "ping" events used for keep-alive)
|
||||||
|
// According to SSE spec, events with no name default to "message"
|
||||||
|
if evt.Name != "" && evt.Name != "message" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
msg, err := jsonrpc.DecodeMessage(evt.Data)
|
msg, err := jsonrpc.DecodeMessage(evt.Data)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.fail(fmt.Errorf("%s: failed to decode event: %v", requestSummary, err))
|
c.fail(fmt.Errorf("%s: failed to decode event: %v", requestSummary, err))
|
||||||
|
|
|
||||||
|
|
@ -1746,3 +1746,153 @@ func mustNotPanic(t *testing.T, h http.Handler) http.Handler {
|
||||||
h.ServeHTTP(w, req)
|
h.ServeHTTP(w, req)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPingEventFiltering verifies that the streamable client correctly filters
|
||||||
|
// out SSE "ping" events, which are used for keep-alive but should not be
|
||||||
|
// treated as JSON-RPC messages.
|
||||||
|
//
|
||||||
|
// This test addresses issue #636: the client should skip non-"message" events
|
||||||
|
// according to the SSE specification. It tests the fix in processStream where
|
||||||
|
// events with evt.Name != "" && evt.Name != "message" are skipped.
|
||||||
|
func TestPingEventFiltering(t *testing.T) {
|
||||||
|
// This test verifies the low-level processStream filtering.
|
||||||
|
// We create a mock response with ping and message events.
|
||||||
|
|
||||||
|
sseData := `event: ping
|
||||||
|
data: ping
|
||||||
|
|
||||||
|
event: message
|
||||||
|
id: 1
|
||||||
|
data: {"jsonrpc":"2.0","id":1,"result":{}}
|
||||||
|
|
||||||
|
event: ping
|
||||||
|
data: keepalive
|
||||||
|
|
||||||
|
`
|
||||||
|
|
||||||
|
resp := &http.Response{
|
||||||
|
StatusCode: http.StatusOK,
|
||||||
|
Header: http.Header{"Content-Type": []string{"text/event-stream"}},
|
||||||
|
Body: io.NopCloser(bytes.NewReader([]byte(sseData))),
|
||||||
|
}
|
||||||
|
|
||||||
|
// Create a minimal streamableClientConn for testing
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
incoming := make(chan jsonrpc.Message, 10)
|
||||||
|
done := make(chan struct{})
|
||||||
|
|
||||||
|
conn := &streamableClientConn{
|
||||||
|
ctx: ctx,
|
||||||
|
done: done,
|
||||||
|
incoming: incoming,
|
||||||
|
}
|
||||||
|
|
||||||
|
// Create a test request
|
||||||
|
testReq := &jsonrpc.Request{
|
||||||
|
ID: jsonrpc2.Int64ID(1),
|
||||||
|
Method: "test",
|
||||||
|
}
|
||||||
|
|
||||||
|
// Process the stream
|
||||||
|
go conn.processStream("test", resp, testReq)
|
||||||
|
|
||||||
|
// Collect messages with timeout
|
||||||
|
var messages []jsonrpc.Message
|
||||||
|
timeout := time.After(1 * time.Second)
|
||||||
|
|
||||||
|
collectLoop:
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case msg := <-incoming:
|
||||||
|
messages = append(messages, msg)
|
||||||
|
// We expect only 1 message (the response), not the ping events
|
||||||
|
if len(messages) >= 1 {
|
||||||
|
break collectLoop
|
||||||
|
}
|
||||||
|
case <-timeout:
|
||||||
|
break collectLoop
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify we only received the actual message, not the ping events
|
||||||
|
if len(messages) != 1 {
|
||||||
|
t.Errorf("got %d messages, want 1 (ping events should be filtered)", len(messages))
|
||||||
|
for i, msg := range messages {
|
||||||
|
t.Logf("message %d: %T", i, msg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify the message is the response
|
||||||
|
if len(messages) > 0 {
|
||||||
|
resp, ok := messages[0].(*jsonrpc.Response)
|
||||||
|
if !ok {
|
||||||
|
t.Errorf("first message is %T, want *jsonrpc.Response", messages[0])
|
||||||
|
} else if resp.ID.Raw() != int64(1) {
|
||||||
|
t.Errorf("response ID is %v, want 1", resp.ID.Raw())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestScanEventsPingFiltering is a unit test for the low-level event scanning
|
||||||
|
// with ping events to verify scanEvents properly parses all event types.
|
||||||
|
func TestScanEventsPingFiltering(t *testing.T) {
|
||||||
|
// Create SSE stream with mixed events
|
||||||
|
sseData := `event: ping
|
||||||
|
data: ping
|
||||||
|
|
||||||
|
event: message
|
||||||
|
data: {"jsonrpc":"2.0","method":"test","params":{}}
|
||||||
|
|
||||||
|
event: ping
|
||||||
|
data: keepalive
|
||||||
|
|
||||||
|
event: message
|
||||||
|
data: {"jsonrpc":"2.0","method":"test2","params":{}}
|
||||||
|
|
||||||
|
`
|
||||||
|
|
||||||
|
reader := strings.NewReader(sseData)
|
||||||
|
var events []Event
|
||||||
|
|
||||||
|
// Scan all events
|
||||||
|
for evt, err := range scanEvents(reader) {
|
||||||
|
if err != nil {
|
||||||
|
if err != io.EOF {
|
||||||
|
t.Fatalf("scanEvents error: %v", err)
|
||||||
|
}
|
||||||
|
break
|
||||||
|
}
|
||||||
|
events = append(events, evt)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify we got all 4 events
|
||||||
|
if len(events) != 4 {
|
||||||
|
t.Fatalf("got %d events, want 4", len(events))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify event types
|
||||||
|
expectedNames := []string{"ping", "message", "ping", "message"}
|
||||||
|
for i, evt := range events {
|
||||||
|
if evt.Name != expectedNames[i] {
|
||||||
|
t.Errorf("event %d: got name %q, want %q", i, evt.Name, expectedNames[i])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify that we can decode the message events but would fail on ping events
|
||||||
|
for i, evt := range events {
|
||||||
|
if evt.Name == "message" {
|
||||||
|
_, err := jsonrpc.DecodeMessage(evt.Data)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("event %d: failed to decode message event: %v", i, err)
|
||||||
|
}
|
||||||
|
} else if evt.Name == "ping" {
|
||||||
|
// Ping events have non-JSON data and should fail decoding
|
||||||
|
_, err := jsonrpc.DecodeMessage(evt.Data)
|
||||||
|
if err == nil {
|
||||||
|
t.Errorf("event %d: ping event unexpectedly decoded as valid JSON-RPC", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue