Merge pull request #54 from MatthewWid/batching

Session event batching

Author: MatthewWid

CHANGELOG.md

@@ -7,8 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+## 0.10.0 - 2023-09-28
+
 ### Added
 
+* Added the [`Session#batch`](./docs/api.md#sessionbatch-batcher-eventbuffer--buffer-eventbuffer--void--promisevoid--promisevoid) method that can be used to batch multiple events into a single transmission over the wire.
 * Added the [`EventBuffer`](./docs/api.md#eventbuffer) class that can be used to write raw spec-compliant SSE fields into a text buffer that can be sent directly over the wire.
 
 ### Deprecated

docs/api.md

@@ -95,6 +95,17 @@ This uses the [`push`](#session%23push%3A-(event%3A-string%2C-data%3A-any)-%3D>-
 |-|-|-|-|
 |`eventName`|`string`|`"iteration"`|Event name to use when dispatching a data event from the yielded value to the client.|
 
+#### `Session#batch`: `(batcher: EventBuffer | ((buffer: EventBuffer) => void | Promise<void>)) => Promise<void>`
+
+Batch and send multiple events at once.
+
+If given an [`EventBuffer`](#eventbuffer) instance, its contents will be sent to the client.
+
+If given a callback, it will be passed an instance of [`EventBuffer`](#eventbuffer) which uses the same serializer and sanitizer as the session.  
+Once its execution completes - or once it resolves if it returns a promise - the contents of the passed [`EventBuffer`](#eventbuffer) will be sent to the client.
+
+Returns a promise that resolves once all data from the event buffer has been successfully sent to the client.
+
 #### `Session#event`: `(type: string) => this`
 
 **⚠ DEPRECATED:** This method is deprecated. [See here](https://github.com/MatthewWid/better-sse/issues/52). 
@@ -228,8 +239,6 @@ Takes the [same arguments as the Channel class constructor](#new-channel()).
 
 An `EventBuffer` allows you to write [raw spec-compliant SSE fields](https://html.spec.whatwg.org/multipage/server-sent-events.html#processField) into a text buffer that can be sent directly over the wire.
 
-This is made available for users with more advanced use-cases who need to create an event text stream from scratch themselves. Most users will not need to access this directly and can use the simplified helper methods provided by the [`Session`](#session) class instead.
-
 #### `new EventBuffer([options = {}])`
 
 `options` is an object with the following properties:

package.json

@@ -1,7 +1,7 @@
 {
 	"name": "better-sse",
 	"description": "Dead simple, dependency-less, spec-compliant server-side events implementation for Node, written in TypeScript.",
-	"version": "0.9.0",
+	"version": "0.10.0",
 	"main": "./build/index.js",
 	"types": "./build/index.d.ts",
 	"license": "MIT",

src/EventBuffer.ts

@@ -22,8 +22,6 @@ interface EventBufferOptions {
 
 /**
  * An `EventBuffer` allows you to write raw spec-compliant SSE fields into a text buffer that can be sent directly over the wire.
- *
- * This is made available for users with more advanced use-cases who need to create an event text stream from scratch themselves. Most users will not need to access this directly and can use the simplified helper methods provided by the `Session` class instead.
  */
 class EventBuffer {
 	private buffer = "";
@@ -121,7 +119,7 @@ class EventBuffer {
 	};
 
 	/**
-	 * Create, write and dispatch an event with the given data to the client all at once.
+	 * Create, write and dispatch an event with the given data all at once.
 	 *
 	 * This is equivalent to calling the methods `event`, `id`, `data` and `dispatch` in that order.
 	 *
@@ -151,7 +149,7 @@ class EventBuffer {
 	 * If no event name is given in the `options` object, the event name is set to `"stream"`.
 	 *
 	 * @param stream - Readable stream to consume data from.
-	 * @param options - Options to alter how the stream is flushed to the client.
+	 * @param options - Event name to use for each event created.
 	 *
 	 * @returns A promise that resolves or rejects based on the success of the stream write finishing.
 	 */

src/Session.test.ts

@@ -12,6 +12,7 @@ import {
 	getBuffer,
 } from "./lib/testUtils";
 import {Session} from "./Session";
+import {EventBuffer} from "./EventBuffer";
 
 let server: http.Server;
 let url: string;
@@ -521,6 +522,75 @@ describe("push", () => {
 		}));
 });
 
+describe("batching", () => {
+	const data = "test-data";
+
+	it("given a synchronous callback, creates a new event buffer and writes its contents to the response after execution has finished", () =>
+		new Promise<void>((done) => {
+			server.on("request", async (req, res) => {
+				const session = new Session(req, res);
+
+				await waitForConnect(session);
+
+				const write = vi.spyOn(res, "write");
+
+				await session.batch((buffer) => {
+					buffer.push(data);
+				});
+
+				expect(write.mock.calls[0][0]).toContain(data);
+
+				done();
+			});
+
+			eventsource = new EventSource(url);
+		}));
+
+	it("given an asynchronous callback, creates a new event buffer and writes its contents to the response after the returned promise has resolved", () =>
+		new Promise<void>((done) => {
+			server.on("request", async (req, res) => {
+				const session = new Session(req, res);
+
+				await waitForConnect(session);
+
+				const write = vi.spyOn(res, "write");
+
+				await session.batch(async (buffer) => {
+					buffer.push(data);
+				});
+
+				expect(write.mock.calls[0][0]).toContain(data);
+
+				done();
+			});
+
+			eventsource = new EventSource(url);
+		}));
+
+	it("given an event buffer, writes its contents to the response", () =>
+		new Promise<void>((done) => {
+			server.on("request", async (req, res) => {
+				const session = new Session(req, res);
+
+				await waitForConnect(session);
+
+				const write = vi.spyOn(res, "write");
+
+				const buffer = new EventBuffer();
+
+				buffer.push(data);
+
+				await session.batch(buffer);
+
+				expect(write.mock.calls[0][0]).toContain(data);
+
+				done();
+			});
+
+			eventsource = new EventSource(url);
+		}));
+});
+
 describe("polyfill support", () => {
 	const lastEventId = "123456";
 

src/Session.ts

@@ -9,6 +9,8 @@ import {TypedEmitter, EventMap} from "./lib/TypedEmitter";
 import {generateId} from "./lib/generateId";
 import {createPushFromStream} from "./lib/createPushFromStream";
 import {createPushFromIterable} from "./lib/createPushFromIterable";
+import {serialize, SerializerFunction} from "./lib/serialize";
+import {sanitize, SanitizerFunction} from "./lib/sanitize";
 
 interface SessionOptions
 	extends Pick<EventBufferOptions, "serializer" | "sanitizer"> {
@@ -135,6 +137,8 @@ class Session<
 		write: (chunk: string) => void;
 	};
 
+	private serialize: SerializerFunction;
+	private sanitize: SanitizerFunction;
 	private trustClientEventId: boolean;
 	private initialRetry: number | null;
 	private keepAliveInterval: number | null;
@@ -153,10 +157,13 @@ class Session<
 
 		this.res = res;
 
-		this.buffer = new EventBuffer({
-			serializer: options.serializer,
-			sanitizer: options.sanitizer,
-		});
+		const serializer = options.serializer ?? serialize;
+		const sanitizer = options.sanitizer ?? sanitize;
+
+		this.serialize = serializer;
+		this.sanitize = sanitizer;
+
+		this.buffer = new EventBuffer({serializer, sanitizer});
 
 		this.trustClientEventId = options.trustClientEventId ?? true;
 
@@ -377,6 +384,37 @@ class Session<
 	 * @returns A promise that resolves once all data has been successfully yielded from the iterable.
 	 */
 	iterate = createPushFromIterable(this.push);
+
+	/**
+	 * Batch and send multiple events at once.
+	 *
+	 * If given an `EventBuffer` instance, its contents will be sent to the client.
+	 *
+	 * If given a callback, it will be passed an instance of `EventBuffer` which uses the same serializer and sanitizer as the session.
+	 * Once its execution completes - or once it resolves if it returns a promise - the contents of the passed `EventBuffer` will be sent to the client.
+	 *
+	 * @param batcher - Event buffer to get contents from, or callback that takes an event buffer to write to.
+	 *
+	 * @returns A promise that resolves once all data from the event buffer has been successfully sent to the client.
+	 *
+	 * @see EventBuffer
+	 */
+	batch = async (
+		batcher: EventBuffer | ((buffer: EventBuffer) => void | Promise<void>)
+	) => {
+		if (batcher instanceof EventBuffer) {
+			this.res.write(batcher.read());
+		} else {
+			const buffer = new EventBuffer({
+				serializer: this.serialize,
+				sanitizer: this.sanitize,
+			});
+
+			await batcher(buffer);
+
+			this.res.write(buffer.read());
+		}
+	};
 }
 
 export type {SessionOptions, SessionEvents, DefaultSessionState};