docs: show ndjson post stream cancellation

Author: aleclarson

examples/ndjson-stream.ts

@@ -2,17 +2,34 @@ import type { HattipHandler } from '@hattip/core'
 import { createClient, createRouter } from 'rouzer'
 import * as http from 'rouzer/http'
 import * as ndjson from 'rouzer/ndjson'
+import { z } from 'zod'
 
 type Event = {
   id: number
   message: string
 }
 
+const EventFilter = z.object({
+  names: z.array(z.string()),
+  where: z.array(
+    z.object({
+      path: z.string(),
+      equals: z.string(),
+    })
+  ),
+})
+
 export const events = http.get('events', {
   response: ndjson.$type<Event>(),
 })
 
-export const routes = { events }
+// NDJSON responses work for POST routes with ordinary JSON body schemas too.
+export const stream = http.post('events/stream', {
+  body: EventFilter,
+  response: ndjson.$type<Event>(),
+})
+
+export const routes = { events, stream }
 
 /**
  * Tiny Hattip adapter used only to keep this example self-contained. Real apps
@@ -46,6 +63,17 @@ async function collect<T>(source: AsyncIterable<T>) {
   return values
 }
 
+async function readFirst<T>(source: AsyncIterable<T>) {
+  const iterator = source[Symbol.asyncIterator]()
+  try {
+    return (await iterator.next()).value
+  } finally {
+    // Closing the client iterator cancels the response body. For Rouzer NDJSON
+    // routes, that cancellation reaches the server source iterator's return().
+    await iterator.return?.()
+  }
+}
+
 export async function runNdjsonStreamExample() {
   const handler = createRouter({
     basePath: 'api/',
@@ -55,6 +83,14 @@ export async function runNdjsonStreamExample() {
       yield { id: 1, message: 'ready' }
       yield { id: 2, message: 'done' }
     },
+    async *stream({ body }) {
+      // The POST body was parsed and validated before the stream starts.
+      yield {
+        id: 1,
+        message: `${body.names[0]} for ${body.where[0]?.equals}`,
+      }
+      yield { id: 2, message: 'done' }
+    },
   })
 
   const client = createClient({
@@ -64,5 +100,16 @@ export async function runNdjsonStreamExample() {
     fetch: createLocalFetch(handler),
   })
 
-  return collect(await client.events())
+  const allEvents = await collect(await client.events())
+
+  // This call sends a JSON body, receives an AsyncIterable, and then stops after
+  // one event. Request signals can also be used to cancel long-lived streams.
+  const firstMatchingEvent = await readFirst(
+    await client.stream({
+      names: ['session.message'],
+      where: [{ path: 'id', equals: 'ses_123' }],
+    })
+  )
+
+  return { allEvents, firstMatchingEvent }
 }

test/examples.test.ts

@@ -33,8 +33,14 @@ test('typed error response example stays runnable', async () => {
 })
 
 test('NDJSON stream example stays runnable', async () => {
-  await expect(runNdjsonStreamExample()).resolves.toEqual([
-    { id: 1, message: 'ready' },
-    { id: 2, message: 'done' },
-  ])
+  await expect(runNdjsonStreamExample()).resolves.toEqual({
+    allEvents: [
+      { id: 1, message: 'ready' },
+      { id: 2, message: 'done' },
+    ],
+    firstMatchingEvent: {
+      id: 1,
+      message: 'session.message for ses_123',
+    },
+  })
 })