SSE

CHANGELOG.md

@@ -2,6 +2,40 @@
 
 ## Version 21
 
+### v21.5.0
+
+- Feat: Introducing [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events):
+  - Basic implementation of the event streams feature is now available using `EventStreamFactory` class;
+  - The new factory is similar to `EndpointsFactory` including the middlewares support;
+  - Client application can subscribe to the event stream using `EventSource` class instance;
+  - `Documentation` and `Integration` do not have yet a special depiction of such endpoints;
+  - This feature is a lightweight alternative to [Zod Sockets](https://github.com/RobinTail/zod-sockets).
+
+```ts
+import { z } from "zod";
+import { EventStreamFactory } from "express-zod-api";
+import { setTimeout } from "node:timers/promises";
+
+const subscriptionEndpoint = EventStreamFactory({
+  events: { time: z.number().int().positive() },
+}).buildVoid({
+  input: z.object({}), // optional input schema
+  handler: async ({ options: { emit, isClosed } }) => {
+    while (!isClosed()) {
+      emit("time", Date.now());
+      await setTimeout(1000);
+    }
+  },
+});
+```
+
+```js
+const source = new EventSource("https://example.com/api/v1/time");
+source.addEventListener("time", (event) => {
+  const data = JSON.parse(event.data); // number
+});
+```
+
 ### v21.4.0
 
 - Return type of public methods `getTags()` and `getScopes()` of `Endpoint` corrected to `ReadyonlyArray<string>`;

README.md

@@ -1195,13 +1195,38 @@ createConfig({
 
 ## Subscriptions
 
-If you want the user of a client application to be able to subscribe to subsequent updates initiated by the server, the
-capabilities of this framework and the HTTP protocol itself would not be enough in this case. I have developed an
-additional websocket operating framework, [Zod Sockets](https://github.com/RobinTail/zod-sockets), which has similar
-principles and capabilities. Check out an example of the synergy between two frameworks on handling subscription events
-in order to emit (broadcast) the `time` event every second with a current time in its payload:
+If you want the user of a client application to be able to subscribe to subsequent updates initiated by the server,
+consider [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events) (SSE) feature.
+Client application can subscribe to the event stream using `EventSource` class instance. The following example
+demonstrates the implementation emitting the `time` event each second.
 
-https://github.com/RobinTail/zod-sockets#subscriptions
+```typescript
+import { z } from "zod";
+import { EventStreamFactory } from "express-zod-api";
+import { setTimeout } from "node:timers/promises";
+
+const subscriptionEndpoint = EventStreamFactory({
+  events: { time: z.number().int().positive() },
+}).buildVoid({
+  input: z.object({}), // optional input schema
+  handler: async ({ options: { emit, isClosed } }) => {
+    while (!isClosed()) {
+      emit("time", Date.now());
+      await setTimeout(1000);
+    }
+  },
+});
+```
+
+```js
+const source = new EventSource("https://example.com/api/v1/time");
+source.addEventListener("time", (event) => {
+  const data = JSON.parse(event.data); // number
+});
+```
+
+If you need more capabilities, such as bidirectional event sending, I have developed an additional websocket operating
+framework, [Zod Sockets](https://github.com/RobinTail/zod-sockets), which has similar principles and capabilities.
 
 # Integration and Documentation
 

example/config.ts

@@ -22,6 +22,7 @@ export const config = createConfig({
   tags: {
     users: "Everything about the users",
     files: "Everything about the files processing",
+    subscriptions: "Everything about the subscriptions",
   },
 });
 

example/endpoints/time-subscription.ts

@@ -0,0 +1,23 @@
+import { z } from "zod";
+import { setTimeout } from "node:timers/promises";
+import { eventsFactory } from "../factories";
+
+export const subscriptionEndpoint = eventsFactory.buildVoid({
+  tag: "subscriptions",
+  input: z.object({
+    trigger: z.string().optional(),
+  }),
+  handler: async ({
+    input: { trigger },
+    options: { emit, isClosed },
+    logger,
+  }) => {
+    if (trigger === "failure") throw new Error("Intentional failure");
+    while (!isClosed()) {
+      logger.debug("emitting");
+      emit("time", Date.now());
+      await setTimeout(1000);
+    }
+    logger.debug("closed");
+  },
+});

example/example.client.ts

@@ -177,6 +177,23 @@ type PostV1AvatarRawResponse =
   | PostV1AvatarRawPositiveResponse
   | PostV1AvatarRawNegativeResponse;
 
+type GetV1EventsTimeInput = {
+  trigger?: string | undefined;
+};
+
+type GetV1EventsTimePositiveResponse = {
+  data: number;
+  event: "time";
+  id?: string | undefined;
+  retry?: number | undefined;
+};
+
+type GetV1EventsTimeNegativeResponse = string;
+
+type GetV1EventsTimeResponse =
+  | GetV1EventsTimePositiveResponse
+  | GetV1EventsTimeNegativeResponse;
+
 export type Path =
   | "/v1/user/retrieve"
   | "/v1/user/:id/remove"
@@ -186,7 +203,8 @@ export type Path =
   | "/v1/avatar/send"
   | "/v1/avatar/stream"
   | "/v1/avatar/upload"
-  | "/v1/avatar/raw";
+  | "/v1/avatar/raw"
+  | "/v1/events/time";
 
 export type Method = "get" | "post" | "put" | "delete" | "patch";
 
@@ -200,6 +218,7 @@ export interface Input {
   "get /v1/avatar/stream": GetV1AvatarStreamInput;
   "post /v1/avatar/upload": PostV1AvatarUploadInput;
   "post /v1/avatar/raw": PostV1AvatarRawInput;
+  "get /v1/events/time": GetV1EventsTimeInput;
 }
 
 export interface PositiveResponse {
@@ -212,6 +231,7 @@ export interface PositiveResponse {
   "get /v1/avatar/stream": GetV1AvatarStreamPositiveResponse;
   "post /v1/avatar/upload": PostV1AvatarUploadPositiveResponse;
   "post /v1/avatar/raw": PostV1AvatarRawPositiveResponse;
+  "get /v1/events/time": GetV1EventsTimePositiveResponse;
 }
 
 export interface NegativeResponse {
@@ -224,6 +244,7 @@ export interface NegativeResponse {
   "get /v1/avatar/stream": GetV1AvatarStreamNegativeResponse;
   "post /v1/avatar/upload": PostV1AvatarUploadNegativeResponse;
   "post /v1/avatar/raw": PostV1AvatarRawNegativeResponse;
+  "get /v1/events/time": GetV1EventsTimeNegativeResponse;
 }
 
 export interface Response {
@@ -236,6 +257,7 @@ export interface Response {
   "get /v1/avatar/stream": GetV1AvatarStreamResponse;
   "post /v1/avatar/upload": PostV1AvatarUploadResponse;
   "post /v1/avatar/raw": PostV1AvatarRawResponse;
+  "get /v1/events/time": GetV1EventsTimeResponse;
 }
 
 export type MethodPath = keyof Input;
@@ -259,6 +281,7 @@ export const endpointTags = {
   "get /v1/avatar/stream": ["users", "files"],
   "post /v1/avatar/upload": ["files"],
   "post /v1/avatar/raw": ["files"],
+  "get /v1/events/time": ["subscriptions"],
 };
 
 export type Implementation = (

example/example.documentation.yaml

@@ -566,6 +566,50 @@ paths:
                     status: error
                     error:
                       message: Sample error message
+  /v1/events/time:
+    get:
+      operationId: GetV1EventsTime
+      tags:
+        - subscriptions
+      parameters:
+        - name: trigger
+          in: query
+          required: false
+          description: GET /v1/events/time Parameter
+          schema:
+            type: string
+      responses:
+        "200":
+          description: GET /v1/events/time Positive response
+          content:
+            text/event-stream:
+              schema:
+                type: object
+                properties:
+                  data:
+                    type: integer
+                    format: int64
+                    exclusiveMinimum: 0
+                    maximum: 9007199254740991
+                  event:
+                    type: string
+                    const: time
+                  id:
+                    type: string
+                  retry:
+                    type: integer
+                    format: int64
+                    exclusiveMinimum: 0
+                    maximum: 9007199254740991
+                required:
+                  - data
+                  - event
+        "400":
+          description: GET /v1/events/time Negative response
+          content:
+            text/plain:
+              schema:
+                type: string
 components:
   schemas:
     Schema1:
@@ -603,5 +647,7 @@ tags:
     description: Everything about the users
   - name: files
     description: Everything about the files processing
+  - name: subscriptions
+    description: Everything about the subscriptions
 servers:
   - url: https://example.com

example/factories.ts

@@ -5,6 +5,7 @@ import {
   defaultResultHandler,
   ez,
   ensureHttpError,
+  EventStreamFactory,
 } from "../src";
 import { config } from "./config";
 import { authMiddleware } from "./middlewares";
@@ -110,3 +111,8 @@ export const noContentFactory = new EndpointsFactory({
     },
   }),
 });
+
+export const eventsFactory = new EventStreamFactory({
+  config,
+  events: { time: z.number().int().positive() },
+});

example/routing.ts

@@ -3,6 +3,7 @@ import { rawAcceptingEndpoint } from "./endpoints/accept-raw";
 import { createUserEndpoint } from "./endpoints/create-user";
 import { deleteUserEndpoint } from "./endpoints/delete-user";
 import { listUsersEndpoint } from "./endpoints/list-users";
+import { subscriptionEndpoint } from "./endpoints/time-subscription";
 import { uploadAvatarEndpoint } from "./endpoints/upload-avatar";
 import { retrieveUserEndpoint } from "./endpoints/retrieve-user";
 import { sendAvatarEndpoint } from "./endpoints/send-avatar";
@@ -35,6 +36,9 @@ export const routing: Routing = {
       // raw body acceptance example
       raw: rawAcceptingEndpoint,
     },
+    events: {
+      time: subscriptionEndpoint,
+    },
   },
   // path /public serves static files from /example/assets
   public: new ServeStatic(join("example", "assets"), {

src/content-type.ts

@@ -2,6 +2,7 @@ export const contentTypes = {
   json: "application/json",
   upload: "multipart/form-data",
   raw: "application/octet-stream",
+  sse: "text/event-stream",
 };
 
 export type ContentType = keyof typeof contentTypes;

src/index.ts

@@ -28,6 +28,7 @@ export {
 } from "./errors";
 export { testEndpoint, testMiddleware } from "./testing";
 export { Integration } from "./integration";
+export { EventStreamFactory } from "./sse";
 
 export { ez } from "./proprietary-schemas";
 

src/server.ts

@@ -82,6 +82,7 @@ export const createServer = async (config: ServerConfig, routing: Routing) => {
   }
 
   const parsers: Parsers = {
+    sse: [],
     json: [config.jsonParser || express.json()],
     raw: [config.rawParser || express.raw(), moveRaw],
     upload: config.upload