Skip to content

Commit e310fbb

Browse files
committed
added streaming client
1 parent e826f0d commit e310fbb

2 files changed

Lines changed: 228 additions & 0 deletions

File tree

‎prisma/schema.prisma‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -228,4 +228,10 @@ model PasswordResetToken {
228228
token String @unique
229229
expiresAt DateTime
230230
used Boolean @default(false)
231+
}
232+
233+
model DescriptionLookup {
234+
// Composite identity constructed as `${title} | ${type} | ${unitType}`
235+
id String @id
236+
description String
231237
}

‎src/hooks/useStreamClient.ts‎

Lines changed: 222 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,222 @@
1+
"use client";
2+
3+
import { useCallback, useMemo, useRef, useState } from "react";
4+
5+
interface StreamChunkMeta {
6+
chunkIndex: number;
7+
response: Response;
8+
}
9+
10+
type StreamChunkParser<TChunk> = (
11+
chunk: string,
12+
meta: StreamChunkMeta
13+
) => TChunk;
14+
15+
interface StreamResult<TChunk> {
16+
response: Response;
17+
chunks: TChunk[];
18+
}
19+
20+
interface StreamClientOptions<TChunk> extends Omit<RequestInit, "signal"> {
21+
encoding?: string;
22+
parseChunk?: StreamChunkParser<TChunk>;
23+
onChunk?: (chunk: TChunk, meta: StreamChunkMeta) => void;
24+
onDone?: (result: StreamResult<TChunk>) => void;
25+
onError?: (error: Error) => void;
26+
signal?: AbortSignal;
27+
}
28+
29+
interface UseStreamClientReturn<TChunk> {
30+
isStreaming: boolean;
31+
streamClient: {
32+
stream: (
33+
url: string,
34+
options?: StreamClientOptions<TChunk>
35+
) => Promise<StreamResult<TChunk>>;
36+
cancel: () => void;
37+
};
38+
}
39+
40+
const isAbortError = (error: unknown): boolean => {
41+
if (error instanceof DOMException) {
42+
return error.name === "AbortError";
43+
}
44+
45+
if (typeof error === "object" && error !== null) {
46+
const { name } = error as { name?: string };
47+
return name === "AbortError";
48+
}
49+
50+
return false;
51+
};
52+
53+
const defaultParseChunk = <TChunk,>(chunk: string): TChunk =>
54+
chunk as unknown as TChunk;
55+
56+
/**
57+
* Lightweight wrapper around the Fetch streaming APIs that yields decoded string chunks
58+
* and lets consumers provide a typed parser so they can work with structured data.
59+
*
60+
* Example:
61+
* const { isStreaming, streamClient } = useStreamClient<MyChunk>();
62+
* await streamClient.stream("/api/stream", {
63+
* parseChunk: (chunk) => JSON.parse(chunk) as MyChunk,
64+
* onChunk: (data) => setItems((items) => [...items, data]),
65+
* });
66+
*/
67+
export function useStreamClient<TChunk = string>(): UseStreamClientReturn<TChunk> {
68+
const [isStreaming, setIsStreaming] = useState(false);
69+
const activeControllerRef = useRef<AbortController | null>(null);
70+
71+
const cancel = useCallback(() => {
72+
if (activeControllerRef.current) {
73+
activeControllerRef.current.abort();
74+
activeControllerRef.current = null;
75+
setIsStreaming(false);
76+
}
77+
}, []);
78+
79+
const stream = useCallback(
80+
async (
81+
url: string,
82+
options: StreamClientOptions<TChunk> = {}
83+
): Promise<StreamResult<TChunk>> => {
84+
const {
85+
encoding = "utf-8",
86+
parseChunk = defaultParseChunk<TChunk>,
87+
onChunk,
88+
onDone,
89+
onError,
90+
signal: externalSignal,
91+
...requestInit
92+
} = options;
93+
94+
cancel();
95+
96+
const controller = new AbortController();
97+
const { signal } = controller;
98+
activeControllerRef.current = controller;
99+
100+
let detachExternalAbort: (() => void) | undefined;
101+
102+
if (externalSignal) {
103+
if (externalSignal.aborted) {
104+
controller.abort(externalSignal.reason);
105+
} else {
106+
const handleAbort = () => controller.abort(externalSignal.reason);
107+
externalSignal.addEventListener("abort", handleAbort);
108+
detachExternalAbort = () => {
109+
externalSignal.removeEventListener("abort", handleAbort);
110+
};
111+
}
112+
}
113+
114+
setIsStreaming(true);
115+
116+
const textDecoder = new TextDecoder(encoding);
117+
const chunks: TChunk[] = [];
118+
119+
try {
120+
const response = await fetch(url, {
121+
...requestInit,
122+
signal,
123+
});
124+
125+
if (!response.ok) {
126+
let message = `${response.status} ${response.statusText}`;
127+
128+
try {
129+
const errorText = await response.text();
130+
message = errorText || message;
131+
} catch {
132+
// Swallow text parsing errors, fall back to status text.
133+
}
134+
135+
throw new Error(message);
136+
}
137+
138+
if (!response.body) {
139+
throw new Error("Response does not contain a readable stream");
140+
}
141+
142+
const reader = response.body.getReader();
143+
let chunkIndex = 0;
144+
145+
const emitChunk = (chunkText: string) => {
146+
if (!chunkText) {
147+
return;
148+
}
149+
150+
const meta: StreamChunkMeta = {
151+
chunkIndex,
152+
response,
153+
};
154+
155+
const parsedChunk = parseChunk(chunkText, meta);
156+
chunks.push(parsedChunk);
157+
onChunk?.(parsedChunk, meta);
158+
chunkIndex += 1;
159+
};
160+
161+
while (true) {
162+
const { done, value } = await reader.read();
163+
164+
if (done) {
165+
break;
166+
}
167+
168+
if (value) {
169+
const chunkText = textDecoder.decode(value, { stream: true });
170+
emitChunk(chunkText);
171+
}
172+
}
173+
174+
const trailingChunk = textDecoder.decode();
175+
emitChunk(trailingChunk);
176+
177+
const result: StreamResult<TChunk> = {
178+
response,
179+
chunks,
180+
};
181+
182+
onDone?.(result);
183+
184+
return result;
185+
} catch (error) {
186+
if (!isAbortError(error)) {
187+
onError?.(error as Error);
188+
}
189+
190+
throw error;
191+
} finally {
192+
detachExternalAbort?.();
193+
194+
if (activeControllerRef.current === controller) {
195+
activeControllerRef.current = null;
196+
setIsStreaming(false);
197+
}
198+
}
199+
},
200+
[cancel]
201+
);
202+
203+
const streamClient = useMemo(
204+
() => ({
205+
stream,
206+
cancel,
207+
}),
208+
[stream, cancel]
209+
);
210+
211+
return {
212+
isStreaming,
213+
streamClient,
214+
};
215+
}
216+
217+
export type {
218+
StreamChunkMeta,
219+
StreamChunkParser,
220+
StreamClientOptions,
221+
StreamResult,
222+
};

0 commit comments

Comments
 (0)