processStream reads from a single response body, sending events to the incoming channel. It returns the ID of the last processed event and a flag indicating if the connection was closed by the client. If resp is nil, it returns "", false.
(ctx context.Context, requestSummary string, resp *http.Response, forCall *jsonrpc.Request)
| 2075 | // indicating if the connection was closed by the client. If resp is nil, it |
| 2076 | // returns "", false. |
| 2077 | func (c *streamableClientConn) processStream(ctx context.Context, requestSummary string, resp *http.Response, forCall *jsonrpc.Request) (lastEventID string, reconnectDelay time.Duration, clientClosed bool) { |
| 2078 | defer func() { |
| 2079 | // Drain any remaining unprocessed body. This allows the connection to be re-used after closing. |
| 2080 | io.Copy(io.Discard, resp.Body) |
| 2081 | resp.Body.Close() |
| 2082 | }() |
| 2083 | for evt, err := range scanEvents(resp.Body) { |
| 2084 | if err != nil { |
| 2085 | if ctx.Err() != nil { |
| 2086 | return "", 0, true // don't reconnect: client cancelled |
| 2087 | } |
| 2088 | |
| 2089 | // Malformed events are hard errors that indicate corrupted data or protocol |
| 2090 | // violations. These should fail the connection permanently. |
| 2091 | if errors.Is(err, errMalformedEvent) { |
| 2092 | c.fail(fmt.Errorf("%s: %v", requestSummary, err)) |
| 2093 | return "", 0, true |
| 2094 | } |
| 2095 | |
| 2096 | break |
| 2097 | } |
| 2098 | |
| 2099 | if evt.ID != "" { |
| 2100 | lastEventID = evt.ID |
| 2101 | } |
| 2102 | |
| 2103 | if evt.Retry != "" { |
| 2104 | if n, err := strconv.ParseInt(evt.Retry, 10, 64); err == nil { |
| 2105 | reconnectDelay = time.Duration(n) * time.Millisecond |
| 2106 | } |
| 2107 | } |
| 2108 | |
| 2109 | // According to SSE specification |
| 2110 | // (https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation) |
| 2111 | // events with an empty data buffer are allowed. |
| 2112 | // In MCP these can be priming events (SEP-1699) that carry only a Last-Event-ID for stream resumption. |
| 2113 | if len(evt.Data) == 0 { |
| 2114 | continue |
| 2115 | } |
| 2116 | |
| 2117 | // According to SSE spec, events with no name default to "message" |
| 2118 | if evt.Name != "" && evt.Name != "message" { |
| 2119 | continue |
| 2120 | } |
| 2121 | |
| 2122 | msg, err := jsonrpc.DecodeMessage(evt.Data) |
| 2123 | if err != nil { |
| 2124 | c.fail(fmt.Errorf("%s: failed to decode event: %v", requestSummary, err)) |
| 2125 | return "", 0, true |
| 2126 | } |
| 2127 | |
| 2128 | select { |
| 2129 | case c.incoming <- msg: |
| 2130 | // Check if this is the response to our call, which terminates the request. |
| 2131 | // (it could also be a server->client request or notification). |
| 2132 | if jsonResp, ok := msg.(*jsonrpc.Response); ok && forCall != nil { |
| 2133 | // TODO: we should never get a response when forReq is nil (the standalone SSE request). |
| 2134 | // We should detect this case. |