packages/drive-store/utils/stream.ts

   1 import type { ReadableStreamDefaultReadResult } from 'web-streams-polyfill';
   2 import { ReadableStream, TransformStream } from 'web-streams-polyfill';
   3
   4 export const untilStreamEnd = async <T>(stream: ReadableStream<T>, action?: (data: T) => Promise<void>) => {
   5     const reader = stream.getReader();
   6
   7     const processResponse = async (result: ReadableStreamDefaultReadResult<T>): Promise<any> => {
   8         if (result.done) {
   9             return;
  10         }
  11
  12         await action?.(result.value);
  13
  14         return processResponse(await reader.read());
  15     };
  16
  17     return processResponse(await reader.read());
  18 };
  19
  20 export const streamToBuffer = async (stream: ReadableStream<Uint8Array>) => {
  21     const chunks: Uint8Array[] = [];
  22     await untilStreamEnd(stream, async (chunk) => {
  23         chunks.push(chunk);
  24     });
  25     return chunks;
  26 };
  27
  28 export const bufferToStream = (buffer: Uint8Array[]): ReadableStream<Uint8Array> => {
  29     return new ReadableStream({
  30         start(controller) {
  31             buffer.forEach((chunk) => controller.enqueue(chunk));
  32             controller.close();
  33         },
  34     });
  35 };
  36
  37 export class ObserverStream extends TransformStream<Uint8Array, Uint8Array> {
  38     constructor(fn?: (chunk: Uint8Array) => void) {
  39         super({
  40             transform(chunk, controller) {
  41                 fn?.(chunk);
  42                 controller.enqueue(chunk);
  43             },
  44         });
  45     }
  46 }