Stream Newline Delimited JSON

Introduction

Orval generates code that properly types responses streamed from NDJSON. NDJSON is a technique to stream an array of JSON objects. This is mostly used when the data set is large.

How to use

Orval does not generate code for actually parsing the stream, but rather provides type safety. You can use the code in the example below to see how you can achieve reading streamed data. Proper type support is only supported when using the fetch client as either a standalone client, or as a httpClient.

Example

1 // orval.config.ts
2 import { defineConfig } from 'orval';
3 
4 export default defineConfig({
5   petstore: {
6     input: {
7       target: './stream.yaml',
8     },
9     output: {
10       client: 'fetch',
11       target: 'src/endpoints.ts',
12       schemas: 'src/model',
13     },
14   },
15 });

1 openapi: 3.1.0
2 info:
3   version: 1.0.0
4   title: Stream
5 paths:
6   /stream:
7     get:
8       operationId: stream
9       description: Stream results
10       responses:
11         '200':
12           description: The stream result.
13           content:
14             application/x-ndjson:
15               schema:
16                 $ref: '#/components/schemas/StreamEntry'
17 components:
18   schemas:
19     StreamEntry:
20       type: object
21       properties:
22         foo:
23           type: number
24         bar:
25           type: string

1 // Generated code
2 interface TypedResponse<T> extends Response {
3   json(): Promise<T>;
4 }
5 
6 /**
7  * Stream results
8  */
9 export type streamResponse200 = {
10   stream: TypedResponse<StreamEntry>;
11   status: 200;
12 };
13 export type streamResponseComposite = streamResponse200;
14 
15 export type streamResponse = streamResponseComposite & {
16   headers: Headers;
17 };
18 
19 export const getStreamUrl = () => {
20   return `/stream`;
21 };
22 
23 export const stream = async (
24   options?: RequestInit,
25 ): Promise<streamResponse> => {
26   const stream = await fetch(getStreamUrl(), {
27     ...options,
28     method: 'GET',
29     headers: { Accept: 'application/x-ndjson', ...options?.headers },
30   });
31 
32   return {
33     status: stream.status,
34     stream,
35     headers: stream.headers,
36   } as streamResponse;
37 };

1 // Calling code
2 export const readStream = <T extends object>(
3   response: Response & { json(): Promise<T> },
4   processLine: (value: T) => void | boolean,
5   onError?: (response?: Response) => any,
6 ): Promise<any> => {
7   if (!response.ok && onError) {
8     return onError(response);
9   }
10   if (!response.body) return Promise.resolve(() => {});
11 
12   const stream = response.body.getReader();
13   const matcher = /\r?\n/;
14   const decoder = new TextDecoder();
15   let buffer = '';
16 
17   const loop: () => Promise<undefined> = () =>
18     stream.read().then(({ done, value }) => {
19       if (done) {
20         if (buffer.length > 0) processLine(JSON.parse(buffer));
21       } else {
22         const chunk = decoder.decode(value, {
23           stream: true,
24         });
25         buffer += chunk;
26 
27         const parts = buffer.split(matcher);
28         buffer = parts.pop() ?? '';
29         const validParts = parts.filter((p) => p);
30         if (validParts.length !== 0) {
31           for (const i of validParts) {
32             const p = JSON.parse(i) as T;
33             processLine(p);
34           }
35           return loop();
36         }
37       }
38     });
39 
40   return loop();
41 };
42 
43 export const getResult = async () => {
44   const results: StreamEntry[] = [];
45 
46   const streamResponse = await stream();
47   if (streamResponse.status !== 200) return results;
48 
49   // The promise is resolved when the stream is complete.
50   await readStream(streamResponse.stream, (obj) => {
51     // obj is typed as StreamEntry
52     results.push(obj);
53   });
54   return results;
55 };