From 64fadf3cf21b1fadb8c28d94213a0bcbe0027040 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Sun, 13 Apr 2025 08:26:49 +0100 Subject: [PATCH] resume streams on error - initial implementation --- src/client/streamableHttp.test.ts | 32 ++++-- src/client/streamableHttp.ts | 72 ++++++++---- src/server/streamableHttp.test.ts | 176 ++++++++++++++++++++++++++++- src/server/streamableHttp.ts | 180 ++++++++++++++++++++++++------ 4 files changed, 392 insertions(+), 68 deletions(-) diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts index 1572e912..8d9a6a9c 100644 --- a/src/client/streamableHttp.test.ts +++ b/src/client/streamableHttp.test.ts @@ -275,17 +275,21 @@ describe("StreamableHTTPClientTransport", () => { })).toBe(true); }); - it("should include last-event-id header when resuming a broken connection", async () => { - // First make a successful connection that provides an event ID + it("should handle reconnection with last-event-id for resumability", async () => { + // Set up a stream that will send an event with ID then error const encoder = new TextEncoder(); const stream = new ReadableStream({ start(controller) { const event = "id: event-123\nevent: message\ndata: {\"jsonrpc\": \"2.0\", \"method\": \"serverNotification\", \"params\": {}}\n\n"; controller.enqueue(encoder.encode(event)); - controller.close(); + // Simulate network error + setTimeout(() => { + controller.error(new Error("Network error")); + }, 10); } }); + // Mock the initial connection (global.fetch as jest.Mock).mockResolvedValueOnce({ ok: true, status: 200, @@ -293,23 +297,33 @@ describe("StreamableHTTPClientTransport", () => { body: stream }); + const errorSpy = jest.fn(); + transport.onerror = errorSpy; + await transport.start(); await transport["_startOrAuthStandaloneSSE"](); + + // Let the stream process and error await new Promise(resolve => setTimeout(resolve, 50)); - - // Now simulate attempting to reconnect + + // Verify error was caught + expect(errorSpy).toHaveBeenCalled(); + + // Mock the reconnection (the transport should try to reconnect after error) (global.fetch as jest.Mock).mockResolvedValueOnce({ ok: true, status: 200, headers: new Headers({ "content-type": "text/event-stream" }), body: null }); - - await transport["_startOrAuthStandaloneSSE"](); - - // Check that Last-Event-ID was included + + // Allow time for automatic reconnection + await new Promise(resolve => setTimeout(resolve, 1100)); // > 1 second delay + + // Verify that the client attempted to reconnect with the last-event-id const calls = (global.fetch as jest.Mock).mock.calls; const lastCall = calls[calls.length - 1]; + expect(lastCall[1].method).toBe("GET"); expect(lastCall[1].headers.get("last-event-id")).toBe("event-123"); }); diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index ea69ee77..9b3c7c20 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -49,7 +49,6 @@ export class StreamableHTTPClientTransport implements Transport { private _requestInit?: RequestInit; private _authProvider?: OAuthClientProvider; private _sessionId?: string; - private _lastEventId?: string; onclose?: () => void; onerror?: (error: Error) => void; @@ -102,16 +101,16 @@ export class StreamableHTTPClientTransport implements Transport { ); } - private async _startOrAuthStandaloneSSE(): Promise { + private async _startOrAuthStandaloneSSE(lastEventId?: string): Promise { try { // Try to open an initial SSE stream with GET to listen for server messages // This is optional according to the spec - server may not support it const headers = await this._commonHeaders(); headers.set("Accept", "text/event-stream"); - // Include Last-Event-ID header for resumable streams - if (this._lastEventId) { - headers.set("last-event-id", this._lastEventId); + // Include Last-Event-ID header for resumable streams if provided + if (lastEventId) { + headers.set("last-event-id", lastEventId); } const response = await fetch(this._url, { @@ -150,31 +149,61 @@ export class StreamableHTTPClientTransport implements Transport { return; } + let lastEventId: string | undefined; + const processStream = async () => { // Create a pipeline: binary stream -> text decoder -> SSE parser const eventStream = stream .pipeThrough(new TextDecoderStream()) .pipeThrough(new EventSourceParserStream()); - for await (const event of eventStream) { - // Update last event ID if provided - if (event.id) { - this._lastEventId = event.id; - } - // Handle message events (default event type is undefined per docs) - // or explicit 'message' event type - if (!event.event || event.event === "message") { - try { - const message = JSONRPCMessageSchema.parse(JSON.parse(event.data)); - this.onmessage?.(message); - } catch (error) { - this.onerror?.(error as Error); + try { + for await (const event of eventStream) { + // Update last event ID if provided + if (event.id) { + lastEventId = event.id; } + + // Handle message events (default event type is undefined per docs) + // or explicit 'message' event type + if (!event.event || event.event === "message") { + try { + const message = JSONRPCMessageSchema.parse(JSON.parse(event.data)); + this.onmessage?.(message); + } catch (error) { + this.onerror?.(error as Error); + } + } + } + } catch (error) { + // Handle stream errors - likely a network disconnect + this.onerror?.(new Error(`SSE stream disconnected: ${error instanceof Error ? error.message : String(error)}`)); + + // Attempt to reconnect if the stream disconnects unexpectedly + // Wait a short time before reconnecting to avoid rapid reconnection loops + if (this._abortController && !this._abortController.signal.aborted) { + setTimeout(() => { + // Use the last event ID to resume where we left off + this._startOrAuthStandaloneSSE(lastEventId).catch(reconnectError => { + this.onerror?.(new Error(`Failed to reconnect SSE stream: ${reconnectError instanceof Error ? reconnectError.message : String(reconnectError)}`)); + }); + }, 1000); // 1 second delay before reconnection attempt } } }; - processStream().catch(err => this.onerror?.(err)); + processStream().catch(err => { + this.onerror?.(err); + + // Try to reconnect on unexpected errors + if (this._abortController && !this._abortController.signal.aborted) { + setTimeout(() => { + this._startOrAuthStandaloneSSE(lastEventId).catch(reconnectError => { + this.onerror?.(new Error(`Failed to reconnect SSE stream: ${reconnectError instanceof Error ? reconnectError.message : String(reconnectError)}`)); + }); + }, 1000); + } + }); } async start() { @@ -252,7 +281,7 @@ export class StreamableHTTPClientTransport implements Transport { // if the accepted notification is initialized, we start the SSE stream // if it's supported by the server if (isJSONRPCNotification(message) && message.method === "notifications/initialized") { - // We don't need to handle 405 here anymore as it's handled in _startOrAuthStandaloneSSE + // Start without a lastEventId since this is a fresh connection this._startOrAuthStandaloneSSE().catch(err => this.onerror?.(err)); } return; @@ -268,6 +297,9 @@ export class StreamableHTTPClientTransport implements Transport { if (hasRequests) { if (contentType?.includes("text/event-stream")) { + // 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); } else if (contentType?.includes("application/json")) { // For non-streaming servers, we might get direct JSON responses diff --git a/src/server/streamableHttp.test.ts b/src/server/streamableHttp.test.ts index 949cafc7..a03a457d 100644 --- a/src/server/streamableHttp.test.ts +++ b/src/server/streamableHttp.test.ts @@ -1,7 +1,7 @@ import { createServer, type Server, IncomingMessage, ServerResponse } from "node:http"; import { AddressInfo } from "node:net"; import { randomUUID } from "node:crypto"; -import { StreamableHTTPServerTransport } from "./streamableHttp.js"; +import { EventStore, StreamableHTTPServerTransport } from "./streamableHttp.js"; import { McpServer } from "./mcp.js"; import { CallToolResult, JSONRPCMessage } from "../types.js"; import { z } from "zod"; @@ -13,6 +13,7 @@ interface TestServerConfig { sessionIdGenerator?: () => string | undefined; enableJsonResponse?: boolean; customRequestHandler?: (req: IncomingMessage, res: ServerResponse, parsedBody?: unknown) => Promise; + eventStore?: EventStore; } /** @@ -26,7 +27,7 @@ async function createTestServer(config: TestServerConfig = {}): Promise<{ }> { const mcpServer = new McpServer( { name: "test-server", version: "1.0.0" }, - { capabilities: {} } + { capabilities: { logging: {} } } ); mcpServer.tool( @@ -40,7 +41,8 @@ async function createTestServer(config: TestServerConfig = {}): Promise<{ const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: config.sessionIdGenerator ?? (() => randomUUID()), - enableJsonResponse: config.enableJsonResponse ?? false + enableJsonResponse: config.enableJsonResponse ?? false, + eventStore: config.eventStore }); await mcpServer.connect(transport); @@ -89,7 +91,10 @@ const TEST_MESSAGES = { params: { clientInfo: { name: "test-client", version: "1.0" }, protocolVersion: "2025-03-26", + capabilities: { + }, }, + id: "init-1", } as JSONRPCMessage, @@ -896,6 +901,171 @@ describe("StreamableHTTPServerTransport with pre-parsed body", () => { }); }); +// Test resumability support +describe("StreamableHTTPServerTransport with resumability", () => { + let server: Server; + let transport: StreamableHTTPServerTransport; + let baseUrl: URL; + let sessionId: string; + let mcpServer: McpServer; + const storedEvents: Map = new Map(); + + // Simple implementation of EventStore + const eventStore: EventStore = { + generateEventId(streamId: string): string { + return `${streamId}_${randomUUID()}`; + }, + getStreamIdFromEventId(eventId: string): string { + return eventId.split('_')[0]; // Extract stream ID from the event ID + }, + async storeEvent(streamId: string, message: JSONRPCMessage): Promise { + const eventId = this.generateEventId(streamId); + storedEvents.set(eventId, { eventId, message }); + return eventId; + }, + + async getEventsAfter(lastEventId: string): Promise> { + const streamId = lastEventId.split('_')[0]; // Extract stream ID from the event ID + const result: Array<{ eventId: string, message: JSONRPCMessage }> = []; + + // For test simplicity, just return all events with matching streamId that aren't the lastEventId + // This avoids issues with event ordering in tests + for (const [eventId, { message }] of storedEvents.entries()) { + if (eventId.startsWith(streamId) && eventId !== lastEventId) { + result.push({ eventId, message }); + } + } + + return result; + }, + }; + + beforeEach(async () => { + storedEvents.clear(); + const result = await createTestServer({ + sessionIdGenerator: () => randomUUID(), + eventStore + }); + + server = result.server; + transport = result.transport; + baseUrl = result.baseUrl; + mcpServer = result.mcpServer; + + // Verify resumability is enabled on the transport + expect((transport)['_eventStore']).toBeDefined(); + + // Initialize the server + const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize); + sessionId = initResponse.headers.get("mcp-session-id") as string; + expect(sessionId).toBeDefined(); + }); + + afterEach(async () => { + await stopTestServer({ server, transport }); + storedEvents.clear(); + }); + + it("should store and include event IDs in server SSE messages", async () => { + // Open a standalone SSE stream + const sseResponse = await fetch(baseUrl, { + method: "GET", + headers: { + Accept: "text/event-stream", + "mcp-session-id": sessionId, + }, + }); + + expect(sseResponse.status).toBe(200); + expect(sseResponse.headers.get("content-type")).toBe("text/event-stream"); + + // Send a notification that should be stored with an event ID + const notification: JSONRPCMessage = { + jsonrpc: "2.0", + method: "notifications/message", + params: { level: "info", data: "Test notification with event ID" }, + }; + + // Send the notification via transport + await transport.send(notification); + + // Read from the stream and verify we got the notification with an event ID + const reader = sseResponse.body?.getReader(); + const { value } = await reader!.read(); + const text = new TextDecoder().decode(value); + + // The response should contain an event ID + expect(text).toContain('id: '); + expect(text).toContain('"method":"notifications/message"'); + + // Extract the event ID + const idMatch = text.match(/id: ([^\n]+)/); + expect(idMatch).toBeTruthy(); + + // Verify the event was stored + const eventId = idMatch![1]; + expect(storedEvents.has(eventId)).toBe(true); + const storedEvent = storedEvents.get(eventId); + expect(eventId.startsWith('standalonesse')).toBe(true); + expect(storedEvent?.message).toMatchObject(notification); + }); + + + it("should store and replay MCP server tool notifications", async () => { + // Establish a standalone SSE stream + const sseResponse = await fetch(baseUrl, { + method: "GET", + headers: { + Accept: "text/event-stream", + "mcp-session-id": sessionId, + }, + }); + expect(sseResponse.status).toBe(200); // Send a server notification through the MCP server + 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); + + // Verify the notification was sent with an event ID + expect(text).toContain('id: '); + expect(text).toContain('First notification from MCP server'); + + // Extract the event ID + const idMatch = text.match(/id: ([^\n]+)/); + expect(idMatch).toBeTruthy(); + const firstEventId = idMatch![1]; + + // Send a second notification + await mcpServer.server.sendLoggingMessage({ level: "info", data: "Second notification from MCP server" }); + + // Close the first SSE stream to simulate a disconnect + await reader!.cancel(); + + // Reconnect with the Last-Event-ID to get missed messages + const reconnectResponse = await fetch(baseUrl, { + method: "GET", + headers: { + Accept: "text/event-stream", + "mcp-session-id": sessionId, + "last-event-id": firstEventId + }, + }); + + 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); + + // Verify we received the second notification that was sent after our stored eventId + expect(reconnectText).toContain('Second notification from MCP server'); + expect(reconnectText).toContain('id: '); + }); +}); + // Test stateless mode describe("StreamableHTTPServerTransport in stateless mode", () => { let server: Server; diff --git a/src/server/streamableHttp.ts b/src/server/streamableHttp.ts index 0eaaa673..0474fc21 100644 --- a/src/server/streamableHttp.ts +++ b/src/server/streamableHttp.ts @@ -1,11 +1,47 @@ import { IncomingMessage, ServerResponse } from "node:http"; import { Transport } from "../shared/transport.js"; -import { JSONRPCMessage, JSONRPCMessageSchema, RequestId } from "../types.js"; +import { isJSONRPCNotification, isJSONRPCRequest, isJSONRPCResponse, JSONRPCMessage, JSONRPCMessageSchema, RequestId } from "../types.js"; import getRawBody from "raw-body"; import contentType from "content-type"; +import { randomUUID } from "node:crypto"; const MAXIMUM_MESSAGE_SIZE = "4mb"; +/** + * Interface for resumability support via event storage + */ +export interface EventStore { + /** + * Generates a unique event ID for a given stream ID + * @param streamId The stream ID to include in the event ID + * @returns A unique event ID that includes the stream ID + */ + generateEventId(streamId: string): string; + + /** + * Stores an event for later retrieval + * @param streamId ID of the stream the event belongs to + * @param message The JSON-RPC message to store + * @returns The generated event ID for the stored event + */ + storeEvent(streamId: string, message: JSONRPCMessage): Promise; + + /** + * Retrieves events for a stream starting from a given event ID + * @param lastEventId The event ID to start from + * @returns Array of stored events with their event IDs + */ + getEventsAfter(lastEventId: string): Promise>; + + /** + * Extracts the stream ID from an event ID + * This is necessary for resumability to identify which stream the event belongs to + * @param eventId The event ID to extract stream ID from + * @returns The stream ID portion of the event ID + */ + getStreamIdFromEventId(eventId: string): string; +} + /** * Configuration options for StreamableHTTPServerTransport */ @@ -24,6 +60,12 @@ export interface StreamableHTTPServerTransportOptions { * Default is false (SSE streams are preferred). */ enableJsonResponse?: boolean; + + /** + * Event store for resumability support + * If provided, resumability will be enabled, allowing clients to reconnect and resume messages + */ + eventStore?: EventStore; } /** @@ -64,12 +106,13 @@ export class StreamableHTTPServerTransport implements Transport { // when sessionId is not set (undefined), it means the transport is in stateless mode private sessionIdGenerator: () => string | undefined; private _started: boolean = false; - private _responseMapping: Map = new Map(); + private _streamMapping: Map = new Map(); + private _requestToStreamMapping: Map = new Map(); private _requestResponseMap: Map = new Map(); private _initialized: boolean = false; private _enableJsonResponse: boolean = false; - private _standaloneSSE: ServerResponse | undefined; - + private _standaloneSseStreamId: string = 'standalonesse'; + private _eventStore?: EventStore; sessionId?: string | undefined; onclose?: () => void; @@ -79,6 +122,7 @@ export class StreamableHTTPServerTransport implements Transport { constructor(options: StreamableHTTPServerTransportOptions) { this.sessionIdGenerator = options.sessionIdGenerator; this._enableJsonResponse = options.enableJsonResponse ?? false; + this._eventStore = options.eventStore; } /** @@ -131,6 +175,14 @@ export class StreamableHTTPServerTransport implements Transport { if (!this.validateSession(req, res)) { return; } + // Handle resumability: check for Last-Event-ID header + if (this._eventStore) { + const lastEventId = req.headers['last-event-id'] as string | undefined; + if (lastEventId) { + await this.replayEvents(lastEventId, res); + return; + } + } // The server MUST either return Content-Type: text/event-stream in response to this HTTP GET, // or else return HTTP 405 Method Not Allowed @@ -144,12 +196,9 @@ export class StreamableHTTPServerTransport implements Transport { if (this.sessionId !== undefined) { headers["mcp-session-id"] = this.sessionId; } - // The server MAY include a Last-Event-ID header in the response to this HTTP GET. - // Resumability will be supported in the future // Check if there's already an active standalone SSE stream for this session - - if (this._standaloneSSE !== undefined) { + if (this._streamMapping.get(this._standaloneSseStreamId) !== undefined) { // Only one GET SSE stream is allowed per session res.writeHead(409).end(JSON.stringify({ jsonrpc: "2.0", @@ -161,19 +210,62 @@ export class StreamableHTTPServerTransport implements Transport { })); return; } - // We need to send headers immediately as message will arrive much later, + + // We need to send headers immediately as messages will arrive much later, // otherwise the client will just wait for the first message res.writeHead(200, headers).flushHeaders(); - // Assing the response to the standalone SSE stream - this._standaloneSSE = res; + // Assign the response to the standalone SSE stream + this._streamMapping.set(this._standaloneSseStreamId, res); // Set up close handler for client disconnects res.on("close", () => { - this._standaloneSSE = undefined; + this._streamMapping.delete(this._standaloneSseStreamId); }); } + /** + * Replays events that would have been sent after the specified event ID + * Only used when resumability is enabled + */ + private async replayEvents(lastEventId: string, res: ServerResponse): Promise { + if (!this._eventStore) { + return; + } + + try { + const events = await this._eventStore.getEventsAfter(lastEventId); + const streamId = this._eventStore.getStreamIdFromEventId(lastEventId); + + const oldResStream = this._streamMapping.get(streamId); + this._streamMapping.set(streamId, res); + if (oldResStream) { + // If we have an old response stream, close it + oldResStream.end(); + } + + for (const { eventId, message } of events) { + this.writeSSEEvent(res, message, eventId); + } + } catch (error) { + this.onerror?.(error as Error); + } + } + + /** + * Writes an event to the SSE stream with proper formatting + */ + private writeSSEEvent(res: ServerResponse, message: JSONRPCMessage, eventId?: string): boolean { + let eventData = `event: message\n`; + // Include event ID if provided - this is important for resumability + if (eventId) { + eventData += `id: ${eventId}\n`; + } + eventData += `data: ${JSON.stringify(message)}\n\n`; + + return res.write(eventData); + } + /** * Handles unsupported requests (PUT, PATCH, etc.) */ @@ -285,9 +377,9 @@ export class StreamableHTTPServerTransport implements Transport { // check if it contains requests - const hasRequests = messages.some(msg => 'method' in msg && 'id' in msg); + const hasRequests = messages.some(isJSONRPCRequest); const hasOnlyNotificationsOrResponses = messages.every(msg => - ('method' in msg && !('id' in msg)) || ('result' in msg || 'error' in msg)); + isJSONRPCNotification(msg) || isJSONRPCResponse(msg)); if (hasOnlyNotificationsOrResponses) { // if it only contains notifications or responses, return 202 @@ -300,6 +392,7 @@ export class StreamableHTTPServerTransport implements Transport { } else if (hasRequests) { // The default behavior is to use SSE streaming // but in some cases server will return JSON responses + const streamId = randomUUID(); if (!this._enableJsonResponse) { const headers: Record = { "Content-Type": "text/event-stream", @@ -318,19 +411,22 @@ export class StreamableHTTPServerTransport implements Transport { // We need to track by request ID to maintain the connection for (const message of messages) { if ('method' in message && 'id' in message) { - this._responseMapping.set(message.id, res); + this._streamMapping.set(streamId, res); + this._requestToStreamMapping.set(message.id, streamId); } } // Set up close handler for client disconnects res.on("close", () => { + // find a stream ID for this response // Remove all entries that reference this response - for (const [id, storedRes] of this._responseMapping.entries()) { - if (storedRes === res) { - this._responseMapping.delete(id); + for (const [id, stream] of this._requestToStreamMapping.entries()) { + if (streamId === stream) { + this._requestToStreamMapping.delete(id); this._requestResponseMap.delete(id); } } + this._streamMapping.delete(streamId); }); // handle each message @@ -431,16 +527,13 @@ export class StreamableHTTPServerTransport implements Transport { async close(): Promise { // Close all SSE connections - this._responseMapping.forEach((response) => { + this._streamMapping.forEach((response) => { response.end(); }); - this._responseMapping.clear(); + this._streamMapping.clear(); // Clear any pending responses this._requestResponseMap.clear(); - this._standaloneSSE?.end(); - this._standaloneSSE = undefined; - this.onclose?.(); } @@ -459,32 +552,47 @@ export class StreamableHTTPServerTransport implements Transport { if ('result' in message || 'error' in message) { throw new Error("Cannot send a response on a standalone SSE stream unless resuming a previous client request"); } - - if (this._standaloneSSE === undefined) { + const standaloneSse = this._streamMapping.get(this._standaloneSseStreamId) + if (standaloneSse === undefined) { // The spec says the server MAY send messages on the stream, so it's ok to discard if no stream return; } + // Generate and store event ID if event store is provided + let eventId: string | undefined; + if (this._eventStore) { + // Stores the event and gets the generated event ID + eventId = await this._eventStore.storeEvent(this._standaloneSseStreamId, message); + } + // Send the message to the standalone SSE stream - this._standaloneSSE.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`); + this.writeSSEEvent(standaloneSse, message, eventId); return; } // Get the response for this request - const response = this._responseMapping.get(requestId); - if (!response) { + const streamId = this._requestToStreamMapping.get(requestId); + const response = this._streamMapping.get(streamId!); + if (!streamId || !response) { throw new Error(`No connection established for request ID: ${String(requestId)}`); } if (!this._enableJsonResponse) { - response.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`); - } - if ('result' in message || 'error' in message) { - this._requestResponseMap.set(requestId, message); + // For SSE responses, generate event ID if event store is provided + let eventId: string | undefined; - // Get all request IDs that share the same request response object - const relatedIds = Array.from(this._responseMapping.entries()) - .filter(([_, res]) => res === response) + if (this._eventStore) { + eventId = await this._eventStore.storeEvent(streamId, message); + } + + // Write the event to the response stream + this.writeSSEEvent(response, message, eventId); + } + + if (isJSONRPCResponse(message)) { + this._requestResponseMap.set(requestId, message); + const relatedIds = Array.from(this._requestToStreamMapping.entries()) + .filter(([_, streamId]) => this._streamMapping.get(streamId) === response) .map(([id]) => id); // Check if we have responses for all requests using this connection @@ -516,7 +624,7 @@ export class StreamableHTTPServerTransport implements Transport { // Clean up for (const id of relatedIds) { this._requestResponseMap.delete(id); - this._responseMapping.delete(id); + this._requestToStreamMapping.delete(id); } } }