1// @effect-diagnostics globalJsonInEffect:off
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4
5const SSE_EVENT_SEPARATOR = /(?:\r\n|\r(?!\n)|\n){2}/u;
6
7async function* sseDataLines(response: Response): AsyncGenerator<string> {
8 if (response.body === null) return;
9 const decoder = new TextDecoder();
10 let buffered = "";
11 for await (const chunk of response.body) {
12 buffered += decoder.decode(chunk as Uint8Array, { stream: true });
13 let separator = SSE_EVENT_SEPARATOR.exec(buffered);
14 while (separator !== null) {
15 const rawEvent = buffered.slice(0, separator.index);
16 buffered = buffered.slice(separator.index + separator[0].length);
17 const data = rawEvent
18 .split(/\r\n|\r|\n/u)
19 .filter((line) => line.startsWith("data:"))
20 .map((line) => line.slice("data:".length).trimStart())
21 .join("\n");
22 if (data.length > 0) yield data;
23 separator = SSE_EVENT_SEPARATOR.exec(buffered);
24 }
25 }
26}
27
28export const discardResponseBody = (response: Response): Effect.Effect<void> =>
29 Effect.promise(() => response.body?.cancel().catch(() => undefined) ?? Promise.resolve());
30
31export function responsePayloads<E>(
32 response: Response,
33 onError: (cause: unknown) => E,
34): Stream.Stream<unknown, E> {
35 if (response.status === 202 || response.status === 204) {
36 return Stream.unwrap(discardResponseBody(response).pipe(Effect.as(Stream.empty)));
37 }
38 const mediaType = (response.headers.get("content-type") ?? "").split(";", 1)[0]!.trim();
39 if (mediaType.toLowerCase() === "text/event-stream") {
40 return Stream.fromAsyncIterable(sseDataLines(response), onError).pipe(
41 Stream.mapEffect((data) => Effect.try({ try: () => JSON.parse(data), catch: onError })),
42 );
43 }
44 return Stream.unwrap(
45 Effect.tryPromise({ try: () => response.text(), catch: onError }).pipe(
46 Effect.flatMap((text) =>
47 text.trim().length === 0
48 ? Effect.succeed(Stream.empty)
49 : Effect.try({ try: () => JSON.parse(text) as unknown, catch: onError }).pipe(
50 Effect.map((payload) => Stream.fromIterable([payload])),
51 ),
52 ),
53 ),
54 );
55}