Repository: unjs/h3
Pull request #586 (2023-12-01 07:11:04)

feat: add utilities for server sent events
<!---
☝️ PR title should follow conventional commits (https://conventionalcommits.org) 
-->

### 🔗 Linked issue

https://github.com/unjs/h3/issues/477

<!-- Please ensure there is an open issue and mention its number as #123 -->

### ❓ Type of change

<!-- What types of changes does your code introduce? Put an `x` in all the boxes that apply. -->

- [x] 📖 Documentation (updates to the documentation, readme, or JSdoc annotations)
- [ ] 🐞 Bug fix (a non-breaking change that fixes an issue)
- [ ] 👌 Enhancement (improving an existing functionality like performance)
- [x] ✨ New feature (a non-breaking change that adds functionality)
- [ ] 🧹 Chore (updates to the build process or auxiliary tools and libraries)
- [ ] ⚠️ Breaking change (fix or feature that would cause existing functionality to change)

### 📚 Description

<!-- Describe your changes in detail -->

This PR adds utilities for server sent events. This enables developers to easily create endpoints that return an event stream as outlined [here](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format)

The new utilities can be used like so:

```ts
eventHandler((event) => {
  const eventStream = createEventStream(event);
  
  // start sending the stream to the client
  eventStream.start();

  // send a message every second
  const interval = setInterval(() => {
    await eventStream.push({data: 'hello world'});
  }, 1000)
  
  // cleanup when a client disconnects
  eventStream.on('disconnect', () => {
    clearInterval(interval);
  })
})
```

__Other Changes__
Unit tests have been added to tests these new features
The README.md has been updated to include the `createEventStream` utility in its utility list.

<!-- Why is this change required? What problem does it solve? -->
<!-- If it resolves an open issue, please link to the issue here. For example "Resolves #1337" -->

### 📝 Checklist

<!-- Put an `x` in all the boxes that apply. -->
<!-- If your change requires a documentation PR, please link it appropriately -->
<!-- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [x] I have linked an issue or discussion.
- [x] I have updated the documentation accordingly.


joshmossas: One thing that I'm kind of iffy on is whether it's better to explicitly start sending the stream to the client with `eventStream.start()` or whether we should just automatically start sending the stream to the client when `createEventStream()` is called.

Right now I think being more explicit is better, but I'm open to differing opinions on this. 
arily: just my personal thoughts. 
I think start() only makes sense if 
- have `pause()` / `stop()`
-  `flush()` if you will buffer any unsent events

and these control flows should inside Apps not transports.

how about provide a utility controlling state?
```js
const bSSE = buffered(createEventStream(event))

setInterval(() => bSSE.push({}), 100)

let started = true
setInterval(() => {
  (started = !started) ? (bSSE.start(), bSSE.flush()) : bSSE.stop()
}, 1000)

// transmit every other second
```
joshmossas: @arily thanks for the input. It's made me reevaluate how this API should be structured.

Right now `start` isn't a control flow thing. It's being used to send the stream to the client. I only put it in there because I think people might not want the stream to be immediately sent as the response. Technically, you can still access the stream with `eventStream.stream` so people could call `sendStream(event, eventStream.stream)`  instead of `start` and essentially get the same result.

However this all has made me think that naming the method `start` might be wrong. Maybe something like `eventStream.init()` would be better since it doesn't have any overlap with control flow concepts such as `pause` and `stop`?
 
Although this has made me realize that it may be even better to make a separate utility for returning the `eventStream` to the client. For example something like `sendEventStream(event, eventStream)`. This would also be in-keeping with some of the other H3 utilities. (ex: `sendStream`, `sendWebResponse`, `sendNoContent`, `sendError`, and `send`) so it has that added benefit.

We could also add logic allowing devs to simply return an `eventStream` inside a handler and H3 would just take care of it like it does with streams and web responses. 

So basically the adjusted API would look like this:

```ts
// using the send utilty
eventHandler((event) => {
  const eventStream = createEventStream(event);
  setInterval(() => {
    eventStream.push("hello world");
  })
  sendEventStream(event, eventStream);
}

// using a simple return
eventHandler((event) => {
  const eventStream = createEventStream(event)

  setInterval(() => {
    eventStream.push("hello world")
  }, 1000)

  return eventStream;
})
```
joshmossas: I do like the idea of `pause`, `resume`, and `flush` methods though. It could be worth implementing them.
arily: Ah sorry my wording might not be accurate, by "control flow" I mean control the start/ stop of the flow of transmitting event.

joshmossas: Yes I think I was following you. It was my own usage of `start()` as a method name that was confusing. 

We need to return the eventStream to the client (which is what `start()` did before). Then we also need to be able to `stop()` and `start()` writing data to the stream itself, which I believe is what you were referring to. 

By making a separate `sendEventStream()` utility I think that makes it clearer as to what is happening.

joshmossas: I went ahead and implemented the new API. It looks like this:

```ts
eventHandler((event) => {
  const eventStream = createEventStream(event)
  setInterval(async () => {
     await eventStream.push("Hello world")
  }, 1000)
  return eventStream
})
```
I think it feels more clear this way.

I also implemented the `pause()`, `resume()`, and `flush()`. The semantics of the exact naming can probably be changed but here's a video of the functionality in action:

[Screencast from 12-30-2023 12:42:06 AM.webm](https://github.com/unjs/h3/assets/31548851/bcc87159-b40c-44eb-8e77-f2c1d6047361)
arily: Very nice! Just another thought, I think for GC to properly recycle any streams we are done for, or to terminate the connection before client does, we may need a `close()` method too, just like other streams. 
https://developer.mozilla.org/en-US/docs/Web/API/WritableStream/close
https://developer.mozilla.org/en-US/docs/Web/API/WritableStreamDefaultWriter/close
joshmossas: Yeah so I've already implemented a `end()` method which terminates the connection before the client does. I'm unsure of whether I need to manually close the stream in addition to this or not. I'll investigate further to confirm when I get the chance.
joshmossas: I've published the changes I've made here as a standalone npm package in case anyone wants to make use of the changes now
https://github.com/joshmossas/h3-sse

I'll be maintaining this repo until native SSE support comes to H3. (Either by this PR getting merged or some other change)
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) Report
Attention: Patch coverage is `61.79104%` with `128 lines` in your changes are missing coverage. Please review.
> Project coverage is 76.96%. Comparing base [(`a58d7c9`)](https://app.codecov.io/gh/unjs/h3/commit/a58d7c977b30ef1782bb390a38fed3b28f8afca9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) to head [(`db16b1f`)](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs).
> Report is 18 commits behind head on main.

| [Files](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) | Patch % | Lines |
|---|---|---|
| [src/utils/sse.ts](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs#diff-c3JjL3V0aWxzL3NzZS50cw==) | 71.84% | [78 Missing :warning: ](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) |
| [examples/server-sent-events.ts](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs#diff-ZXhhbXBsZXMvc2VydmVyLXNlbnQtZXZlbnRzLnRz) | 0.00% | [49 Missing and 1 partial :warning: ](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #586      +/-   ##
==========================================
- Coverage   77.83%   76.96%   -0.88%     
==========================================
  Files          47       49       +2     
  Lines        4286     4862     +576     
  Branches      611      637      +26     
==========================================
+ Hits         3336     3742     +406     
- Misses        933     1102     +169     
- Partials       17       18       +1     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/unjs/h3/pull/586?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs).

pi0: Hi dear @joshmossas thanks so much for putting all efforts on this!

I quickly checked implementation it is flawless and nice! I have made few refactors out of my instincts in order to move this forward as fast as possible.

- Utils are refactored to `sse/` with just one `createEventStream` util exposed
- Usage is `return eventStream.send()` in the end. I know it is less intuitive. This allows to move faster without adding direct handlers to the core matcher for now. Eventually we might do this.
- I have enabled `autoclose` by default (haven't found a reason why not to) and it can be opted-out with `{ autoClose: false }` option.
- Close hook is via `onClosed` and registers both handles. I found it easier to recommend people use one trustable hook for cleanup.
- Docs are autogenerated from jsdocs into advanced section, later I might add it to better place below WebSocket section.
- Util marked as `@experimental` mainly because platform support is not tested but I'm positive it is a really good implementation.

Thinking to merge and try on nightly but please consider my changes _blind changes_ and feel free to suggest any amends or make subsequent PRs.



Patch:
diff --git a/docs/2.utils/98.advanced.md b/docs/2.utils/98.advanced.md
index cbd5414..6d482f7 100644
--- a/docs/2.utils/98.advanced.md
+++ b/docs/2.utils/98.advanced.md
@@ -142,3 +142,37 @@ Check if the incoming request is a CORS request.
 Check if the incoming request is a CORS preflight request.
 
 <!-- /automd -->
+
+## Server Sent Events (SSE)
+
+<!-- automd:jsdocs src="../../src/utils/sse/index.ts" -->
+
+### `createEventStream(event, opts?)`
+
+Initialize an EventStream instance for creating [server sent events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events)
+
+**Example:**
+
+```ts
+import { createEventStream, sendEventStream } from "h3";
+
+eventHandler((event) => {
+  const eventStream = createEventStream(event);
+
+  // Send a message every second
+  const interval = setInterval(async () => {
+    await eventStream.push("Hello world");
+  }, 1000);
+
+  // cleanup the interval and close the stream when the connection is terminated
+  eventStream.onClosed(async () => {
+    console.log("closing SSE...");
+    clearInterval(interval);
+    await eventStream.close();
+  });
+
+  return eventStream.send();
+});
+```
+
+<!-- /automd -->
diff --git a/examples/server-sent-events.ts b/examples/server-sent-events.ts
new file mode 100644
index 0000000..5902f4c
--- /dev/null
+++ b/examples/server-sent-events.ts
@@ -0,0 +1,26 @@
+import { createApp, createRouter, eventHandler, createEventStream } from "h3";
+
+export const app = createApp();
+
+const router = createRouter();
+app.use(router);
+
+router.get(
+  "/",
+  eventHandler((event) => {
+    const eventStream = createEventStream(event);
+
+    // Send a message every second
+    const interval = setInterval(async () => {
+      await eventStream.push("Hello world");
+    }, 1000);
+
+    // cleanup the interval and close the stream when the connection is terminated
+    eventStream.onClosed(async () => {
+      clearInterval(interval);
+      await eventStream.close();
+    });
+
+    return eventStream.send();
+  }),
+);
diff --git a/src/utils/index.ts b/src/utils/index.ts
index f24486b..c23210e 100644
--- a/src/utils/index.ts
+++ b/src/utils/index.ts
@@ -10,4 +10,5 @@ export * from "./request";
 export * from "./response";
 export * from "./sanitize";
 export * from "./session";
+export * from "./sse";
 export * from "./static";
diff --git a/src/utils/sse/event-stream.ts b/src/utils/sse/event-stream.ts
new file mode 100644
index 0000000..262a508
--- /dev/null
+++ b/src/utils/sse/event-stream.ts
@@ -0,0 +1,172 @@
+import type { H3Event } from "../../event";
+import { sendStream, setResponseStatus } from "../response";
+import {
+  formatEventStreamMessage,
+  formatEventStreamMessages,
+  setEventStreamHeaders,
+} from "./utils";
+import { EventStreamMessage, EventStreamOptions } from "./types";
+
+/**
+ * A helper class for [server sent events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format)
+ */
+export class EventStream {
+  private readonly _h3Event: H3Event;
+  private readonly _transformStream = new TransformStream();
+  private readonly _writer: WritableStreamDefaultWriter;
+  private readonly _encoder: TextEncoder = new TextEncoder();
+
+  private _writerIsClosed = false;
+  private _paused = false;
+  private _unsentData: undefined | string;
+  private _disposed = false;
+  private _handled = false;
+
+  constructor(event: H3Event, opts: EventStreamOptions = {}) {
+    this._h3Event = event;
+    this._writer = this._transformStream.writable.getWriter();
+    this._writer.closed.then(() => {
+      this._writerIsClosed = true;
+    });
+    if (opts.autoclose !== false) {
+      this._h3Event.node.req.on("close", () => this.close());
+    }
+  }
+
+  /**
+   * Publish new event(s) for the client
+   */
+  async push(message: string): Promise<void>;
+  async push(message: string[]): Promise<void>;
+  async push(message: EventStreamMessage): Promise<void>;
+  async push(message: EventStreamMessage[]): Promise<void>;
+  async push(
+    message: EventStreamMessage | EventStreamMessage[] | string | string[],
+  ) {
+    if (typeof message === "string") {
+      await this._sendEvent({ data: message });
+      return;
+    }
+    if (Array.isArray(message)) {
+      if (message.length === 0) {
+        return;
+      }
+      if (typeof message[0] === "string") {
+        const msgs: EventStreamMessage[] = [];
+        for (const item of message as string[]) {
+          msgs.push({ data: item });
+        }
+        await this._sendEvents(msgs);
+        return;
+      }
+      await this._sendEvents(message as EventStreamMessage[]);
+      return;
+    }
+    await this._sendEvent(message);
+  }
+
+  private async _sendEvent(message: EventStreamMessage) {
+    if (this._writerIsClosed) {
+      return;
+    }
+    if (this._paused && !this._unsentData) {
+      this._unsentData = formatEventStreamMessage(message);
+      return;
+    }
+    if (this._paused) {
+      this._unsentData += formatEventStreamMessage(message);
+      return;
+    }
+    await this._writer
+      .write(this._encoder.encode(formatEventStreamMessage(message)))
+      .catch();
+  }
+
+  private async _sendEvents(messages: EventStreamMessage[]) {
+    if (this._writerIsClosed) {
+      return;
+    }
+    const payload = formatEventStreamMessages(messages);
+    if (this._paused && !this._unsentData) {
+      this._unsentData = payload;
+      return;
+    }
+    if (this._paused) {
+      this._unsentData += payload;
+      return;
+    }
+
+    await this._writer.write(this._encoder.encode(payload)).catch();
+  }
+
+  pause() {
+    this._paused = true;
+  }
+
+  get isPaused() {
+    return this._paused;
+  }
+
+  async resume() {
+    this._paused = false;
+    await this.flush();
+  }
+
+  async flush() {
+    if (this._writerIsClosed) {
+      return;
+    }
+    if (this._unsentData?.length) {
+      await this._writer.write(this._encoder.encode(this._unsentData));
+      this._unsentData = undefined;
+    }
+  }
+
+  /**
+   * Close the stream and the connection if the stream is being sent to the client
+   */
+  async close() {
+    if (this._disposed) {
+      return;
+    }
+    if (!this._writerIsClosed) {
+      try {
+        await this._writer.close();
+      } catch {}
+    }
+    // check if the stream has been given to the client before closing the connection
+    if (
+      this._h3Event._handled &&
+      this._handled &&
+      !this._h3Event.node.res.closed
+    ) {
+      this._h3Event.node.res.end();
+    }
+    this._disposed = true;
+  }
+
+  /**
+   * Triggers callback when the writable stream is closed.
+   * It is also triggered after calling the `close()` method.
+   * It also triggers when the request connection has been closed by either the client or the server.
+   */
+  onClosed(cb: () => any) {
+    this._writer.closed.then(cb);
+    this._h3Event.node?.req.on("close", cb);
+  }
+
+  async send() {
+    setEventStreamHeaders(this._h3Event);
+    setResponseStatus(this._h3Event, 200);
+    this._h3Event._handled = true;
+    this._handled = true;
+    await sendStream(this._h3Event, this._transformStream.readable);
+  }
+}
+
+export function isEventStream(input: unknown): input is EventStream {
+  if (typeof input !== "object" || input === null) {
+    return false;
+  }
+  return input instanceof EventStream;
+}
diff --git a/src/utils/sse/index.ts b/src/utils/sse/index.ts
new file mode 100644
index 0000000..9e418d1
--- /dev/null
+++ b/src/utils/sse/index.ts
@@ -0,0 +1,38 @@
+import type { H3Event } from "../../event";
+import { EventStream } from "./event-stream";
+import { EventStreamOptions } from "./types";
+/**
+ * Initialize an EventStream instance for creating [server sent events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events)
+ *
+ * @experimental This function is experimental and might be unstable in some environments.
+ *
+ * @example
+ *
+ * ```ts
+ * import { createEventStream, sendEventStream } from "h3";
+ *
+ * eventHandler((event) => {
+ *   const eventStream = createEventStream(event);
+ *
+ *   // Send a message every second
+ *   const interval = setInterval(async () => {
+ *     await eventStream.push("Hello world");
+ *   }, 1000);
+ *
+ *   // cleanup the interval and close the stream when the connection is terminated
+ *   eventStream.onClosed(async () => {
+ *     console.log("closing SSE...");
+ *     clearInterval(interval);
+ *     await eventStream.close();
+ *   });
+ *
+ *   return eventStream.send();
+ * });
+ * ```
+ */
+export function createEventStream(
+  event: H3Event,
+  opts?: EventStreamOptions,
+): EventStream {
+  return new EventStream(event, opts);
+}
diff --git a/src/utils/sse/types.ts b/src/utils/sse/types.ts
new file mode 100644
index 0000000..6a3f1f1
--- /dev/null
+++ b/src/utils/sse/types.ts
@@ -0,0 +1,18 @@
+export interface EventStreamOptions {
+  /**
+   * Automatically close the writable stream when the request is closed
+   *
+   * Default is `true`
+   */
+  autoclose?: boolean;
+}
+
+/**
+ * See https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#fields
+ */
+export interface EventStreamMessage {
+  id?: string;
+  event?: string;
+  retry?: number;
+  data: string;
+}
diff --git a/src/utils/sse/utils.ts b/src/utils/sse/utils.ts
new file mode 100644
index 0000000..2a838eb
--- /dev/null
+++ b/src/utils/sse/utils.ts
@@ -0,0 +1,54 @@
+import { HTTPHeaderName } from "../../types";
+import { H3Event } from "../../event";
+import { getHeader } from "../request";
+import { setResponseHeaders } from "../response";
+import { EventStreamMessage } from "./types";
+
+export function formatEventStreamMessage(message: EventStreamMessage): string {
+  let result = "";
+  if (message.id) {
+    result += `id: ${message.id}\n`;
+  }
+  if (message.event) {
+    result += `event: ${message.event}\n`;
+  }
+  if (typeof message.retry === "number" && Number.isInteger(message.retry)) {
+    result += `retry: ${message.retry}\n`;
+  }
+  result += `data: ${message.data}\n\n`;
+  return result;
+}
+
+export function formatEventStreamMessages(
+  messages: EventStreamMessage[],
+): string {
+  let result = "";
+  for (const msg of messages) {
+    result += formatEventStreamMessage(msg);
+  }
+  return result;
+}
+
+export function setEventStreamHeaders(event: H3Event) {
+  const headers: Partial<
+    Record<HTTPHeaderName, string | number | readonly string[]>
+  > = {
+    "Content-Type": "text/event-stream",
+    "Cache-Control":
+      "private, no-cache, no-store, no-transform, must-revalidate, max-age=0",
+    "X-Accel-Buffering": "no", // prevent nginx from buffering the response
+  };
+
+  if (!isHttp2Request(event)) {
+    headers.Connection = "keep-alive";
+  }
+
+  setResponseHeaders(event, headers);
+}
+
+export function isHttp2Request(event: H3Event) {
+  return (
+    getHeader(event, ":path") !== undefined &&
+    getHeader(event, ":method") !== undefined
+  );
+}


Test patch:
diff --git a/test/sse.test.ts b/test/sse.test.ts
new file mode 100644
index 0000000..e18db2e
--- /dev/null
+++ b/test/sse.test.ts
@@ -0,0 +1,125 @@
+import supertest, { SuperTest, Test } from "supertest";
+import { describe, it, beforeEach, expect } from "vitest";
+import {
+  App,
+  createApp,
+  createEventStream,
+  eventHandler,
+  getQuery,
+  toNodeListener,
+} from "../src";
+import {
+  formatEventStreamMessage,
+  formatEventStreamMessages,
+} from "../src/utils/sse/utils";
+
+describe("Server Sent Events (SSE)", () => {
+  let app: App;
+  let request: SuperTest<Test>;
+  beforeEach(() => {
+    app = createApp({ debug: true });
+    app.use(
+      "/sse",
+      eventHandler((event) => {
+        const includeMeta = getQuery(event).includeMeta !== undefined;
+        const eventStream = createEventStream(event);
+        const interval = setInterval(() => {
+          if (includeMeta) {
+            eventStream.push({
+              id: "1",
+              event: "custom-event",
+              data: "hello world",
+            });
+            return;
+          }
+          eventStream.push("hello world");
+        });
+        eventStream.onClosed(async () => {
+          await eventStream.close();
+          clearInterval(interval);
+        });
+        return eventStream.send();
+      }),
+    );
+    request = supertest(toNodeListener(app)) as any;
+  });
+  it("streams events", async () => {
+    let messageCount = 0;
+    request
+      .get("/sse")
+      .expect(200)
+      .expect("Content-Type", "text/event-stream")
+      .buffer()
+      .parse((res, callback) => {
+        res.on("data", (chunk: Buffer) => {
+          messageCount++;
+          const message = chunk.toString();
+          expect(message).toEqual("data: hello world\n\n");
+        });
+        res.on("end", () => {
+          callback(null, "");
+        });
+      })
+      .then()
+      .catch();
+    await new Promise((resolve) => {
+      setTimeout(() => {
+        resolve(true);
+      }, 100);
+    });
+    expect(messageCount > 10).toBe(true);
+  });
+  it("streams events with metadata", async () => {
+    let messageCount = 0;
+    request
+      .get("/sse?includeMeta=true")
+      .expect(200)
+      .expect("Content-Type", "text/event-stream")
+      .buffer()
+      .parse((res, callback) => {
+        res.on("data", (chunk: Buffer) => {
+          messageCount++;
+          const message = chunk.toString();
+          expect(message).toEqual(
+            `id: 1\nevent: custom-event\ndata: hello world\n\n`,
+          );
+        });
+        res.on("end", () => {
+          callback(null, "");
+        });
+      })
+      .then()
+      .catch();
+    await new Promise((resolve) => {
+      setTimeout(() => {
+        resolve(true);
+      }, 100);
+    });
+    expect(messageCount > 10).toBe(true);
+  });
+});
+
+it("properly formats sse messages", () => {
+  const result = formatEventStreamMessage({ data: "hello world" });
+  expect(result).toEqual(`data: hello world\n\n`);
+  const result2 = formatEventStreamMessage({
+    id: "1",
+    event: "custom-event",
+    retry: 10,
+    data: "hello world",
+  });
+  expect(result2).toEqual(
+    `id: 1\nevent: custom-event\nretry: 10\ndata: hello world\n\n`,
+  );
+});
+
+it("properly formats multiple sse messages", () => {
+  const result = formatEventStreamMessages([
+    {
+      data: "hello world",
+    },
+
+    { id: "1", data: "hello world 2" },
+  ]);
+  expect(result).toEqual(`data: hello world\n\nid: 1\ndata: hello world 2\n\n`);
+});