[Blog][Node.js] stream.Readable

0_setup.ts

interface Post {
  title: string;
}

interface FetchPostsArgs {
  skip: number;
  limit: number;
}

const BATCH_SIZE = 2;

const posts: Post[] = [
  { title: "foo" },
  { title: "bar" },
  { title: "buz" },
];

function fetchPosts({
  skip,
  limit,
}: FetchPostsArgs): Promise<Post[]> {
  return Promise.resolve(posts.slice(skip, skip + limit));
}

1-a_create_readable.ts

import { Readable } from "stream";

// Add `read` option when initializing `Readable`.
function createPostsStream(batchSize: number = BATCH_SIZE): Readable {
  const limit = batchSize;
  return new Readable({
    objectMode: true,

    async read(_size: number) {
      let skip = 0;

      let data = await fetchPosts({ skip, limit });
      while (data.length) {
        this.push(data);
        skip += limit;
        data = await fetchPosts({ skip, limit });
      }

      this.push(null); // Notify the stream ends.
    },
  });
}

// or create a subclass of Readable with _read() method

1-b_create_readable.ts

import { Readable } from "stream";

// Read.from with (async) generator func.
async function* _createPostsStreamGenerator(limit: number) {
  let skip = 0;

  let data = await fetchPosts({ skip, limit });
  while (data.length) {
    yield data;
    skip += limit;
    data = await fetchPosts({ skip, limit });
  }
}

function createPostsStream(batchSize: number = BATCH_SIZE): Readable {
  return Readable.from(_createPostsStreamGenerator(batchSize));
}

2-a_consume_readable.ts

import { Writable } from "stream";
import { finished, pipeline } from "stream/promises";

// Create Writable and piped
function createReceiverStream() {
  return new Writable({
    objectMode: true,
    write(chunk: Post[], _encoding: BufferEncoding, callback: Function) {
      console.log("type:", typeof chunk);
      console.log("data:", chunk);
      callback();
    },
  });
}

(async () => {
  const src = createPostsStream();
  const dest = createReceiverStream();
  await finished(src.pipe(dest));

  // or
  // await pipeline(src, dest);
})()
  .catch(console.error);

2-b_consume_readable.ts

import { finished, pipeline } from "stream/promises";


(async () => {
  const src = createPostsStrem();

  // Subscribe events
  src.on("data", (chunk) => {
    console.log(typeof chunk); // object
    console.log(chunk); // [{ title: "foo" }, ...]
  });

  src.on("end", () => {
    console.log("Stream END");
  });

  src.on("error", (err) => {
    console.error(err);
  });

  await finished(src);
})()
  .catch(console.error);

2-c_consume_readable.ts

(async () => {
  // for await
  for await (const chunk of createPostsStream()) {
    console.log(chunk);
  }
})()
  .catch(console.error);