smartstream/ts/smartstream.nodewebhelpers.ts

import * as plugins from './smartstream.plugins.js';

/**
 * Creates a Web ReadableStream from a file.
 *
 * @param filePath - The path to the file to be read
 * @returns A Web ReadableStream that reads the file in chunks
 */
export function createWebReadableStreamFromFile(filePath: string): ReadableStream<Uint8Array> {
  const fileStream = plugins.fs.createReadStream(filePath);

  return new ReadableStream({
    start(controller) {
      // When data is available, enqueue it into the Web ReadableStream
      fileStream.on('data', (chunk) => {
        controller.enqueue(chunk as Uint8Array);
      });

      // When the file stream ends, close the Web ReadableStream
      fileStream.on('end', () => {
        controller.close();
      });

      // If there's an error, error the Web ReadableStream
      fileStream.on('error', (err) => {
        controller.error(err);
      });
    },
    cancel() {
      // If the Web ReadableStream is canceled, destroy the file stream
      fileStream.destroy();
    }
  });
}

/**
 * Converts a Web ReadableStream to a Node.js Readable stream.
 *
 * @param webStream - The Web ReadableStream to convert
 * @returns A Node.js Readable stream that reads data from the Web ReadableStream
 */
export function convertWebReadableToNodeReadable(webStream: ReadableStream<Uint8Array>): plugins.stream.Readable {
  const reader = webStream.getReader();

  return new plugins.stream.Readable({
    async read() {
      try {
        const { value, done } = await reader.read();
        if (done) {
          this.push(null); // Signal end of stream
        } else {
          this.push(Buffer.from(value)); // Convert Uint8Array to Buffer for Node.js Readable
        }
      } catch (err) {
        this.destroy(err); // Handle errors by destroying the stream
      }
    }
  });
}

/**
 * Converts a Node.js Readable stream to a Web ReadableStream.
 *
 * @param nodeStream - The Node.js Readable stream to convert
 * @returns A Web ReadableStream that reads data from the Node.js Readable stream
 */
export function convertNodeReadableToWebReadable(nodeStream: plugins.stream.Readable): ReadableStream<Uint8Array> {
  return new ReadableStream({
    start(controller) {
      nodeStream.on('data', (chunk) => {
        controller.enqueue(new Uint8Array(chunk));
      });

      nodeStream.on('end', () => {
        controller.close();
      });

      nodeStream.on('error', (err) => {
        controller.error(err);
      });
    },
    cancel() {
      nodeStream.destroy();
    }
  });
}

/**
 * Converts a Web WritableStream to a Node.js Writable stream.
 *
 * @param webWritable - The Web WritableStream to convert
 * @returns A Node.js Writable stream that writes data to the Web WritableStream
 */
export function convertWebWritableToNodeWritable(webWritable: WritableStream<Uint8Array>): plugins.stream.Writable {
  const writer = webWritable.getWriter();

  return new plugins.stream.Writable({
    async write(chunk, encoding, callback) {
      try {
        await writer.write(new Uint8Array(chunk));
        callback();
      } catch (err) {
        callback(err);
      }
    },
    final(callback) {
      writer.close().then(() => callback()).catch(callback);
    },
    destroy(err, callback) {
      writer.abort(err).then(() => callback(err)).catch(callback);
    }
  });
}

/**
 * Converts a Node.js Writable stream to a Web WritableStream.
 *
 * @param nodeWritable - The Node.js Writable stream to convert
 * @returns A Web WritableStream that writes data to the Node.js Writable stream
 */
export function convertNodeWritableToWebWritable(nodeWritable: plugins.stream.Writable): WritableStream<Uint8Array> {
  return new WritableStream({
    write(chunk) {
      return new Promise((resolve, reject) => {
        nodeWritable.write(Buffer.from(chunk), (err) => {
          if (err) {
            reject(err);
          } else {
            resolve();
          }
        });
      });
    },
    close() {
      return new Promise((resolve, reject) => {
        nodeWritable.end((err) => {
          if (err) {
            reject(err);
          } else {
            resolve();
          }
        });
      });
    },
    abort(reason) {
      return new Promise((resolve, reject) => {
        nodeWritable.destroy(reason);
      });
    }
  });
}