From c3f08567a2d2078f1f7b890612ef5244e92f8a48 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Apr 2025 14:48:47 +0100 Subject: [PATCH] share implementation of in memory store between examples and integration tests --- src/examples/server/simpleStreamableHttp.ts | 80 +------------------ .../sseAndStreamableHttpCompatibleServer.ts | 79 +----------------- src/examples/shared/inMemoryEventStore.ts | 77 ++++++++++++++++++ .../taskResumability.test.ts | 63 +-------------- 4 files changed, 86 insertions(+), 213 deletions(-) create mode 100644 src/examples/shared/inMemoryEventStore.ts diff --git a/src/examples/server/simpleStreamableHttp.ts b/src/examples/server/simpleStreamableHttp.ts index 0ae0f910..3e7e7c52 100644 --- a/src/examples/server/simpleStreamableHttp.ts +++ b/src/examples/server/simpleStreamableHttp.ts @@ -1,84 +1,10 @@ import express, { Request, Response } from 'express'; import { randomUUID } from 'node:crypto'; import { McpServer } from '../../server/mcp.js'; -import { EventStore, StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; +import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; import { z } from 'zod'; -import { CallToolResult, GetPromptResult, isInitializeRequest, JSONRPCMessage, ReadResourceResult } from '../../types.js'; - -// Create a simple in-memory EventStore for resumability -class InMemoryEventStore implements EventStore { - private events: Map = new Map(); - - /** - * Generates a unique event ID for a given stream ID - */ - private generateEventId(streamId: string): string { - return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`; - } - - private getStreamIdFromEventId(eventId: string): string { - const parts = eventId.split('_'); - return parts.length > 0 ? parts[0] : ''; - } - - /** - * Stores an event with a generated event ID - * Implements EventStore.storeEvent - */ - async storeEvent(streamId: string, message: JSONRPCMessage): Promise { - const eventId = this.generateEventId(streamId); - console.log(`Storing event ${eventId} for stream ${streamId}`); - this.events.set(eventId, { streamId, message }); - return eventId; - } - - /** - * Replays events that occurred after a specific event ID - * Implements EventStore.replayEventsAfter - */ - async replayEventsAfter(lastEventId: string, - { send }: { send: (eventId: string, message: JSONRPCMessage) => Promise } - ): Promise { - if (!lastEventId || !this.events.has(lastEventId)) { - console.log(`No events found for lastEventId: ${lastEventId}`); - return ''; - } - - // Extract the stream ID from the event ID - const streamId = this.getStreamIdFromEventId(lastEventId); - if (!streamId) { - console.log(`Could not extract streamId from lastEventId: ${lastEventId}`); - return ''; - } - - let foundLastEvent = false; - let eventCount = 0; - - // Sort events by eventId for chronological ordering - const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0])); - - for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) { - // Only include events from the same stream - if (eventStreamId !== streamId) { - continue; - } - - // Start sending events after we find the lastEventId - if (eventId === lastEventId) { - foundLastEvent = true; - continue; - } - - if (foundLastEvent) { - await send(eventId, message); - eventCount++; - } - } - - console.log(`Replayed ${eventCount} events after ${lastEventId} for stream ${streamId}`); - return streamId; - } -} +import { CallToolResult, GetPromptResult, isInitializeRequest, ReadResourceResult } from '../../types.js'; +import { InMemoryEventStore } from '../shared/inMemoryEventStore.js'; // Create an MCP server with implementation details const server = new McpServer({ diff --git a/src/examples/server/sseAndStreamableHttpCompatibleServer.ts b/src/examples/server/sseAndStreamableHttpCompatibleServer.ts index 5a79a4e5..378c0e1f 100644 --- a/src/examples/server/sseAndStreamableHttpCompatibleServer.ts +++ b/src/examples/server/sseAndStreamableHttpCompatibleServer.ts @@ -1,10 +1,11 @@ import express, { Request, Response } from 'express'; import { randomUUID } from "node:crypto"; import { McpServer } from '../../server/mcp.js'; -import { EventStore, StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; +import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; import { SSEServerTransport } from '../../server/sse.js'; import { z } from 'zod'; -import { CallToolResult, isInitializeRequest, JSONRPCMessage } from '../../types.js'; +import { CallToolResult, isInitializeRequest } from '../../types.js'; +import { InMemoryEventStore } from '../shared/inMemoryEventStore.js'; /** * This example server demonstrates backwards compatibility with both: @@ -17,81 +18,7 @@ import { CallToolResult, isInitializeRequest, JSONRPCMessage } from '../../types * - /request: The deprecated POST endpoint for older clients (POST to send messages) */ -// Simple in-memory event store for resumability -class InMemoryEventStore implements EventStore { - private events: Map = new Map(); - /** - * Generates a unique event ID for a given stream ID - */ - private generateEventId(streamId: string): string { - return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`; - } - - private getStreamIdFromEventId(eventId: string): string { - const parts = eventId.split('_'); - return parts.length > 0 ? parts[0] : ''; - } - - /** - * Stores an event with a generated event ID - * Implements EventStore.storeEvent - */ - async storeEvent(streamId: string, message: JSONRPCMessage): Promise { - const eventId = this.generateEventId(streamId); - console.log(`Storing event ${eventId} for stream ${streamId}`); - this.events.set(eventId, { streamId, message }); - return eventId; - } - - /** - * Replays events that occurred after a specific event ID - * Implements EventStore.replayEventsAfter - */ - async replayEventsAfter(lastEventId: string, - { send }: { send: (eventId: string, message: JSONRPCMessage) => Promise } - ): Promise { - if (!lastEventId || !this.events.has(lastEventId)) { - console.log(`No events found for lastEventId: ${lastEventId}`); - return ''; - } - - const streamId = this.getStreamIdFromEventId(lastEventId); - if (!streamId) { - console.log(`Could not extract streamId from lastEventId: ${lastEventId}`); - return ''; - } - - let foundLastEvent = false; - let eventCount = 0; - - // Sort events by eventId for chronological ordering - const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0])); - - for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) { - // Only include events from the same stream - if (eventStreamId !== streamId) { - continue; - } - - // Start sending events after we find the lastEventId - if (eventId === lastEventId) { - foundLastEvent = true; - continue; - } - - if (foundLastEvent) { - await send(eventId, message); - eventCount++; - } - } - - console.log(`Replayed ${eventCount} events after ${lastEventId} for stream ${streamId}`); - return streamId; - } -} - -// Create a shared MCP server instance const server = new McpServer({ name: 'backwards-compatible-server', version: '1.0.0', diff --git a/src/examples/shared/inMemoryEventStore.ts b/src/examples/shared/inMemoryEventStore.ts new file mode 100644 index 00000000..fbdebe12 --- /dev/null +++ b/src/examples/shared/inMemoryEventStore.ts @@ -0,0 +1,77 @@ +import { JSONRPCMessage } from '../../types.js'; +import { EventStore } from '../../server/streamableHttp.js'; + +/** + * Simple in-memory implementation of the EventStore interface for resumability + * This is primarily intended for examples and testing, not for production use + * where a persistent storage solution would be more appropriate. + */ +export class InMemoryEventStore implements EventStore { + private events: Map = new Map(); + + /** + * Generates a unique event ID for a given stream ID + */ + private generateEventId(streamId: string): string { + return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`; + } + + /** + * Extracts the stream ID from an event ID + */ + private getStreamIdFromEventId(eventId: string): string { + const parts = eventId.split('_'); + return parts.length > 0 ? parts[0] : ''; + } + + /** + * Stores an event with a generated event ID + * Implements EventStore.storeEvent + */ + async storeEvent(streamId: string, message: JSONRPCMessage): Promise { + const eventId = this.generateEventId(streamId); + this.events.set(eventId, { streamId, message }); + return eventId; + } + + /** + * Replays events that occurred after a specific event ID + * Implements EventStore.replayEventsAfter + */ + async replayEventsAfter(lastEventId: string, + { send }: { send: (eventId: string, message: JSONRPCMessage) => Promise } + ): Promise { + if (!lastEventId || !this.events.has(lastEventId)) { + return ''; + } + + // Extract the stream ID from the event ID + const streamId = this.getStreamIdFromEventId(lastEventId); + if (!streamId) { + return ''; + } + + let foundLastEvent = false; + + // Sort events by eventId for chronological ordering + const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0])); + + for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) { + // Only include events from the same stream + if (eventStreamId !== streamId) { + continue; + } + + // Start sending events after we find the lastEventId + if (eventId === lastEventId) { + foundLastEvent = true; + continue; + } + + if (foundLastEvent) { + await send(eventId, message); + } + } + return streamId; + } +} \ No newline at end of file diff --git a/src/integration-tests/taskResumability.test.ts b/src/integration-tests/taskResumability.test.ts index 3bd23498..efd2611f 100644 --- a/src/integration-tests/taskResumability.test.ts +++ b/src/integration-tests/taskResumability.test.ts @@ -4,68 +4,11 @@ import { randomUUID } from 'node:crypto'; import { Client } from '../client/index.js'; import { StreamableHTTPClientTransport } from '../client/streamableHttp.js'; import { McpServer } from '../server/mcp.js'; -import { EventStore, StreamableHTTPServerTransport } from '../server/streamableHttp.js'; -import { CallToolResultSchema, JSONRPCMessage, LoggingMessageNotificationSchema } from '../types.js'; +import { StreamableHTTPServerTransport } from '../server/streamableHttp.js'; +import { CallToolResultSchema, LoggingMessageNotificationSchema } from '../types.js'; import { z } from 'zod'; +import { InMemoryEventStore } from '../examples/shared/inMemoryEventStore.js'; -/** - * Simple in-memory event store implementation for resumability - */ -class InMemoryEventStore implements EventStore { - private events: Map = new Map(); - - private generateEventId(streamId: string): string { - return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`; - } - - private getStreamIdFromEventId(eventId: string): string { - const parts = eventId.split('_'); - return parts.length > 0 ? parts[0] : ''; - } - - async storeEvent(streamId: string, message: JSONRPCMessage): Promise { - const eventId = this.generateEventId(streamId); - this.events.set(eventId, { streamId, message }); - return eventId; - } - - async replayEventsAfter(lastEventId: string, - { send }: { send: (eventId: string, message: JSONRPCMessage) => Promise } - ): Promise { - if (!lastEventId || !this.events.has(lastEventId)) { - return ''; - } - - // Extract the stream ID from the event ID - const streamId = this.getStreamIdFromEventId(lastEventId); - if (!streamId) { - return ''; - } - let foundLastEvent = false; - - // Sort events by eventId for chronological ordering - const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0])); - - for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) { - // Only include events from the same stream - if (eventStreamId !== streamId) { - continue; - } - - // Start collecting events after we find the lastEventId - if (eventId === lastEventId) { - foundLastEvent = true; - continue; - } - - if (foundLastEvent) { - await send(eventId, message); - } - } - - return streamId; - } -} describe('Transport resumability', () => {