129 lines
3.4 KiB
TypeScript
129 lines
3.4 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { IncomingMessage, ServerResponse } from "node:http";
|
|
import { validateMessage } from "../shared/message.js";
|
|
import { Transport } from "../shared/transport.js";
|
|
import { JSONRPCMessage } from "../types/index.js";
|
|
import getRawBody from "raw-body";
|
|
import contentType from "content-type";
|
|
|
|
const MAXIMUM_MESSAGE_SIZE = "4mb";
|
|
|
|
/**
|
|
* Server transport for SSE: this will send messages over an SSE connection and receive messages from HTTP POST requests.
|
|
*
|
|
* This transport is only available in Node.js environments.
|
|
*/
|
|
export class SSEServerTransport implements Transport {
|
|
private _sseResponse?: ServerResponse;
|
|
private _sessionId: string;
|
|
|
|
onclose?: () => void;
|
|
onerror?: (error: Error) => void;
|
|
onmessage?: (message: JSONRPCMessage) => void;
|
|
|
|
/**
|
|
* Creates a new SSE server transport, which will direct the client to POST messages to the relative or absolute URL identified by `_endpoint`.
|
|
*/
|
|
constructor(private _endpoint: string) {
|
|
this._sessionId = randomUUID();
|
|
}
|
|
|
|
/**
|
|
* Handles the initial SSE connection request.
|
|
*
|
|
* This should be called when a GET request is made to establish the SSE stream.
|
|
*/
|
|
async connectSSE(req: IncomingMessage, res: ServerResponse): Promise<void> {
|
|
if (this._sseResponse) {
|
|
throw new Error("Already connected!");
|
|
}
|
|
|
|
res.writeHead(200, {
|
|
"Content-Type": "text/event-stream",
|
|
"Cache-Control": "no-cache",
|
|
Connection: "keep-alive",
|
|
});
|
|
|
|
// Send the endpoint event
|
|
res.write(
|
|
`event: endpoint\ndata: ${encodeURI(this._endpoint)}?sessionId=${this._sessionId}\n\n`,
|
|
);
|
|
|
|
this._sseResponse = res;
|
|
res.on("close", () => {
|
|
this._sseResponse = undefined;
|
|
this.onclose?.();
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Handles incoming POST messages.
|
|
*
|
|
* This should be called when a POST request is made to send a message to the server.
|
|
*/
|
|
async handlePostMessage(
|
|
req: IncomingMessage,
|
|
res: ServerResponse,
|
|
): Promise<void> {
|
|
if (!this._sseResponse) {
|
|
const message = "SSE connection not established";
|
|
res.writeHead(500).end(message);
|
|
throw new Error(message);
|
|
}
|
|
|
|
let body: string;
|
|
try {
|
|
const ct = contentType.parse(req.headers["content-type"] ?? "");
|
|
if (ct.type !== "application/json") {
|
|
throw new Error(`Unsupported content-type: ${ct}`);
|
|
}
|
|
|
|
body = await getRawBody(req, {
|
|
limit: MAXIMUM_MESSAGE_SIZE,
|
|
encoding: ct.parameters.charset ?? "utf-8",
|
|
});
|
|
} catch (error) {
|
|
res.writeHead(400).end(String(error));
|
|
this.onerror?.(error as Error);
|
|
return;
|
|
}
|
|
|
|
let message: JSONRPCMessage;
|
|
try {
|
|
message = JSON.parse(body);
|
|
validateMessage(message);
|
|
} catch (error) {
|
|
this.onerror?.(error as Error);
|
|
res.writeHead(400).end(`Invalid message: ${body}`);
|
|
return;
|
|
}
|
|
|
|
this.onmessage?.(message);
|
|
res.writeHead(202).end("Accepted");
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
this._sseResponse?.end();
|
|
this._sseResponse = undefined;
|
|
this.onclose?.();
|
|
}
|
|
|
|
async send(message: JSONRPCMessage): Promise<void> {
|
|
if (!this._sseResponse) {
|
|
throw new Error("Not connected");
|
|
}
|
|
|
|
this._sseResponse.write(
|
|
`event: message\ndata: ${JSON.stringify(message)}\n\n`,
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Returns the session ID for this transport.
|
|
*
|
|
* This can be used to route incoming POST requests.
|
|
*/
|
|
get sessionId(): string {
|
|
return this._sessionId;
|
|
}
|
|
}
|