Error: {error}
;
+ if (!time) return Connecting...
;
+
+ return Server time: {new Date(time).toISOString()}
;
+}
+```
+
+## Connection Lifecycle
+
+Manage the connection lifecycle with provided helpers:
+
+### `isClosed()`
+
+Check if the client has disconnected:
+
+```ts
+const endpoint = eventsFactory.buildVoid({
+ input: z.object({}),
+ handler: async ({ ctx: { emit, isClosed } }) => {
+ let counter = 0;
+
+ while (!isClosed()) {
+ emit("count", counter++);
+ await setTimeout(1000);
+ }
+
+ // Cleanup after client disconnects
+ console.log("Stream closed, sent", counter, "events");
+ },
+});
+```
+
+### `signal` (AbortSignal)
+
+Use the abort signal for cleanup and cancellation:
+
+```ts
+const endpoint = eventsFactory.buildVoid({
+ input: z.object({ topic: z.string() }),
+ handler: async ({ input, ctx: { emit, signal }, logger }) => {
+ const subscription = messageBus.subscribe(input.topic);
+
+ // Listen for disconnect
+ signal.addEventListener("abort", () => {
+ logger.info("Client disconnected, cleaning up");
+ subscription.unsubscribe();
+ });
+
+ // Forward messages to stream
+ for await (const message of subscription) {
+ if (signal.aborted) break;
+ emit("message", message);
+ }
+ },
+});
+```
+
+## With Middleware
+
+Add authentication and other middlewares to event streams:
+
+```ts
+import { z } from "zod";
+import { EventStreamFactory } from "express-zod-api";
+import { authMiddleware } from "./middlewares";
+
+const authenticatedEventsFactory = new EventStreamFactory({
+ notification: z.object({
+ id: z.string(),
+ type: z.enum(["info", "warning", "error"]),
+ message: z.string(),
+ timestamp: z.number(),
+ }),
+})
+ .addMiddleware(authMiddleware);
+
+const notificationsEndpoint = authenticatedEventsFactory.buildVoid({
+ input: z.object({}),
+ handler: async ({ ctx: { user, emit, isClosed, signal } }) => {
+ // user is available from authMiddleware
+ const stream = await subscribeToUserNotifications(user.id);
+
+ signal.addEventListener("abort", () => {
+ stream.unsubscribe();
+ });
+
+ for await (const notification of stream) {
+ if (signal.aborted) break;
+ emit("notification", notification);
+ }
+ },
+});
+```
+
+## Real-World Examples
+
+### Live Metrics Dashboard
+
+```ts
+import { EventStreamFactory } from "express-zod-api";
+import { z } from "zod";
+import { setTimeout } from "node:timers/promises";
+
+const metricsFactory = new EventStreamFactory({
+ metrics: z.object({
+ cpu: z.number().min(0).max(100),
+ memory: z.number().min(0).max(100),
+ requests: z.number().int().nonnegative(),
+ errors: z.number().int().nonnegative(),
+ timestamp: z.number(),
+ }),
+});
+
+const metricsEndpoint = metricsFactory.buildVoid({
+ input: z.object({
+ interval: z.number().int().min(1000).default(5000),
+ }),
+ handler: async ({ input, ctx: { emit, isClosed } }) => {
+ while (!isClosed()) {
+ const metrics = await collectSystemMetrics();
+
+ emit("metrics", {
+ ...metrics,
+ timestamp: Date.now(),
+ });
+
+ await setTimeout(input.interval);
+ }
+ },
+});
+```
+
+### Progress Tracking
+
+```ts
+const progressFactory = new EventStreamFactory({
+ progress: z.object({
+ taskId: z.string(),
+ percentage: z.number().min(0).max(100),
+ status: z.enum(["pending", "processing", "completed", "failed"]),
+ message: z.string(),
+ }),
+});
+
+const taskProgressEndpoint = progressFactory.buildVoid({
+ input: z.object({
+ taskId: z.string().uuid(),
+ }),
+ handler: async ({ input, ctx: { emit, isClosed, signal } }) => {
+ const task = await Task.findById(input.taskId);
+
+ task.on("progress", (progress) => {
+ if (!isClosed()) {
+ emit("progress", {
+ taskId: input.taskId,
+ percentage: progress.percentage,
+ status: progress.status,
+ message: progress.message,
+ });
+ }
+ });
+
+ // Wait for completion or disconnect
+ await Promise.race([
+ task.waitForCompletion(),
+ new Promise((resolve) => {
+ signal.addEventListener("abort", resolve);
+ }),
+ ]);
+
+ task.removeAllListeners();
+ },
+});
+```
+
+### Log Streaming
+
+```ts
+import { tail } from "tail";
+
+const logsFactory = new EventStreamFactory({
+ log: z.object({
+ level: z.enum(["info", "warn", "error", "debug"]),
+ message: z.string(),
+ timestamp: z.string(),
+ source: z.string(),
+ }),
+});
+
+const logStreamEndpoint = logsFactory.buildVoid({
+ input: z.object({
+ service: z.string(),
+ level: z.enum(["info", "warn", "error", "debug"]).optional(),
+ }),
+ handler: async ({ input, ctx: { emit, signal }, logger }) => {
+ const logFile = `/var/log/${input.service}.log`;
+ const tailer = new tail(logFile);
+
+ tailer.on("line", (line) => {
+ const log = parseLogLine(line);
+ if (!input.level || log.level === input.level) {
+ emit("log", log);
+ }
+ });
+
+ signal.addEventListener("abort", () => {
+ logger.info("Stopping log stream");
+ tailer.unwatch();
+ });
+ },
+});
+```
+
+## Error Handling
+
+Handle errors in event streams:
+
+```ts
+const endpoint = eventsFactory.buildVoid({
+ input: z.object({ trigger: z.string().optional() }),
+ handler: async ({ input, ctx: { emit, isClosed } }) => {
+ // Trigger error for testing
+ if (input.trigger === "failure") {
+ throw new Error("Intentional failure");
+ }
+
+ try {
+ while (!isClosed()) {
+ const data = await fetchData();
+ emit("data", data);
+ await setTimeout(1000);
+ }
+ } catch (error) {
+ // Errors terminate the stream and send error response
+ throw createHttpError(500, "Data fetch failed");
+ }
+ },
+});
+```
+
+## Routing
+
+Add SSE endpoints to your routing:
+
+```ts
+import { Routing } from "express-zod-api";
+
+const routing: Routing = {
+ v1: {
+ events: {
+ time: subscriptionEndpoint,
+ metrics: metricsEndpoint,
+ notifications: notificationsEndpoint,
+ },
+ },
+};
+```
+
+## Generated Client Support
+
+The `Integration` generator creates a `Subscription` class for type-safe SSE consumption:
+
+```ts
+import { Integration } from "express-zod-api";
+
+const client = new Integration({
+ routing,
+ config,
+ variant: "client",
+});
+
+const code = await client.printFormatted();
+```
+
+Generated client usage:
+
+```ts
+import { Subscription } from "./generated-client";
+
+const subscription = new Subscription("get /v1/events/time", {
+ interval: 1000,
+});
+
+subscription.on("time", (timestamp) => {
+ console.log("Time:", new Date(timestamp));
+});
+
+// Access the underlying EventSource
+subscription.source.addEventListener("error", (err) => {
+ console.error("Connection error", err);
+});
+
+// Close the connection
+subscription.source.close();
+```
+
+## Best Practices
+
+
+
+
+## The Request Lifecycle
+
+When a request hits your API, it goes through several stages:
+
+
+
+
+## Key Features
+
+Express Zod API integrates Express.js, Zod validation, and TypeScript to provide a complete framework for building robust APIs.
+
++ Express Zod API +
++ A TypeScript framework to help you get an API server up and running with I/O schema validation and custom middlewares in minutes. +
+ +