Skip to content

[v1.x] test(server): read SSE streams until expected events arrive instead of assuming one fetch chunk #3174

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
MuhammadFaizan-prog wants to merge 1 commit into
base: v1.x
Choose a base branch
from
+26 −12
Draft

[v1.x] test(server): read SSE streams until expected events arrive instead of assuming one fetch chunk #3174

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 26 additions & 12 deletions test/server/streamableHttp.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,27 @@ async function readSSEEvent(response: Response): Promise<string> {
return new TextDecoder().decode(value);
}

// Reads until the predicate holds, the stream ends or the timeout fires; SSE events can span chunks (#2661).
async function readSSEUntil(response: Response, predicate: (text: string) => boolean, timeoutMs = 2000): Promise<string> {
const reader = response.body!.getReader();
const decoder = new TextDecoder();
let text = '';
const timeout = setTimeout(() => void reader.cancel(), timeoutMs);
try {
while (!predicate(text)) {
const { value, done } = await reader.read();
if (done) {
break;
}
text += decoder.decode(value, { stream: true });
}
} finally {
clearTimeout(timeout);
reader.releaseLock();
}
return text;
}

/**
* Helper to send JSON-RPC request
*/
Expand Down Expand Up @@ -657,11 +678,8 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
expect(response.status).toBe(200);
expect(response.headers.get('content-type')).toBe('text/event-stream');

const reader = response.body?.getReader();

// The responses may come in any order or together in one chunk
const { value } = await reader!.read();
const text = new TextDecoder().decode(value);
// The responses may come in any order and in separate chunks
const text = await readSSEUntil(response, t => t.includes('"id":"req-1"') && t.includes('"id":"req-2"'));

// Check that both responses were sent on the same stream
expect(text).toContain('"id":"req-1"');
Expand Down Expand Up @@ -1413,9 +1431,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'First notification from MCP server' });

// Read the notification from the SSE stream
const reader = sseResponse.body?.getReader();
const { value } = await reader!.read();
const text = new TextDecoder().decode(value);
const text = await readSSEUntil(sseResponse, t => t.includes('First notification from MCP server'));

// Verify the notification was sent with an event ID
expect(text).toContain('id: ');
Expand All @@ -1430,7 +1446,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'Second notification from MCP server' });

// Close the first SSE stream to simulate a disconnect
await reader!.cancel();
await sseResponse.body!.cancel();

// Reconnect with the Last-Event-ID to get missed messages
const reconnectResponse = await fetch(baseUrl, {
Expand All @@ -1446,9 +1462,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
expect(reconnectResponse.status).toBe(200);

// Read the replayed notification
const reconnectReader = reconnectResponse.body?.getReader();
const reconnectData = await reconnectReader!.read();
const reconnectText = new TextDecoder().decode(reconnectData.value);
const reconnectText = await readSSEUntil(reconnectResponse, t => t.includes('Second notification from MCP server'));

// Verify we received the second notification that was sent after our stored eventId
expect(reconnectText).toContain('Second notification from MCP server');
Expand Down
Loading