fix(731): StreamableHTTPClientTransport Fails to Reconnect on Non-Resumable Streams (#732)
This commit is contained in:
parent
af61a086ae
commit
16ea27783c
1 changed files with 127 additions and 16 deletions
|
|
@ -231,7 +231,7 @@ const response = await (this._fetch ?? fetch)(this._url, {
|
|||
);
|
||||
}
|
||||
|
||||
this._handleSseStream(response.body, options);
|
||||
this._handleSseStream(response.body, options, true);
|
||||
} catch (error) {
|
||||
this.onerror?.(error as Error);
|
||||
throw error;
|
||||
|
|
@ -300,7 +300,11 @@ const response = await (this._fetch ?? fetch)(this._url, {
|
|||
}, delay);
|
||||
}
|
||||
|
||||
private _handleSseStream(stream: ReadableStream<Uint8Array> | null, options: StartSSEOptions): void {
|
||||
private _handleSseStream(
|
||||
stream: ReadableStream<Uint8Array> | null,
|
||||
options: StartSSEOptions,
|
||||
isReconnectable: boolean,
|
||||
): void {
|
||||
if (!stream) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -347,20 +351,22 @@ const response = await (this._fetch ?? fetch)(this._url, {
|
|||
this.onerror?.(new Error(`SSE stream disconnected: ${error}`));
|
||||
|
||||
// Attempt to reconnect if the stream disconnects unexpectedly and we aren't closing
|
||||
if (this._abortController && !this._abortController.signal.aborted) {
|
||||
if (
|
||||
isReconnectable &&
|
||||
this._abortController &&
|
||||
!this._abortController.signal.aborted
|
||||
) {
|
||||
// Use the exponential backoff reconnection strategy
|
||||
if (lastEventId !== undefined) {
|
||||
try {
|
||||
this._scheduleReconnection({
|
||||
resumptionToken: lastEventId,
|
||||
onresumptiontoken,
|
||||
replayMessageId
|
||||
}, 0);
|
||||
}
|
||||
catch (error) {
|
||||
this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`));
|
||||
try {
|
||||
this._scheduleReconnection({
|
||||
resumptionToken: lastEventId,
|
||||
onresumptiontoken,
|
||||
replayMessageId
|
||||
}, 0);
|
||||
}
|
||||
catch (error) {
|
||||
this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`));
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -473,7 +479,7 @@ const response = await (this._fetch ?? fetch)(this._url, init);
|
|||
// Handle SSE stream responses for requests
|
||||
// We use the same handler as standalone streams, which now supports
|
||||
// reconnection with the last event ID
|
||||
this._handleSseStream(response.body, { onresumptiontoken });
|
||||
this._handleSseStream(response.body, { onresumptiontoken }, false);
|
||||
} else if (contentType?.includes("application/json")) {
|
||||
// For non-streaming servers, we might get direct JSON responses
|
||||
const data = await response.json();
|
||||
|
|
|
|||
Loading…
Reference in a new issue