Function processStream

webapp/_webapp/src/query/utils.ts:30–64

(
  stream: ReadableStream<Uint8Array>,
  schema: DescMessage,
  onMessage: (chunk: T) => void,
)

}

export const processStream = async <T>(
  stream: ReadableStream<Uint8Array>,
  schema: DescMessage,
  onMessage: (chunk: T) => void,
) => {
  const { slowStreamingMode } = useDevtoolStore.getState();
  const reader = stream.getReader();
  const decoder = new TextDecoder("utf-8");
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });

    let boundary;
    while ((boundary = buffer.indexOf("\n")) !== -1) {
      const message = buffer.slice(0, boundary);
      buffer = buffer.slice(boundary + 1);

      try {
        const parsedValue = JSON.parse(message);
        const messageData = parsedValue.result || parsedValue;
        onMessage(fromJson(schema, messageData) as T);
      } catch (err) {
        logError("Error parsing message from stream", err, message);
      }
    }

    if (import.meta.env.DEV && slowStreamingMode) {
      await new Promise((resolve) => setTimeout(resolve, 500));
    }
  }
};