Show HN: A Node.js stream library that works with promises (opens in new tab)

(npmjs.com)

10 pointsreconbot7y ago3 comments

3 comments

Hi I built this little stream processing library so I could write things like;

  const stats = await pipe(
    Nodes.scan({ fields: true }),
    map(generateStats),
    tap(() => count++),
    reduce(mergeGraphStats, {})
  )

and have generateStats and mergeGraphStats be async functions and not have to worry about error handling and pushing more than one object at a time in a read stream. We use it to process billions of events and objects a day. It makes nodejs streams fun to use.

Hope you find it as useful as we do.

oliverx07y ago

Pretty cool. Could the following be used as a baseline to create the same functionality?

    async function * nextStreamEntry() { 
      while(true){
        const entry = yield;
        const result = await processEntry(entry);
      }
    }

    // Initialize the iterator
    const iter = nextStreamEntry();
    iter.next();

    // For every entry of the stream
    for (const monster of results) {
      iter.next(monster)
    }

reconbotOP7y ago

yes!

I would love to, but not all node versions and not all streams support async iteration yet! Also theres is a lot of utility to some of the stream types in bluestream. Concurrency for one thing. I've been working on an async iterator version of this but haven't yet started using it in production yet.

https://github.com/reconbot/streaming-iterables

j / k navigate · click thread line to collapse

3 comments

reconbotOP7y ago

Hi I built this little stream processing library so I could write things like;

  const stats = await pipe(
    Nodes.scan({ fields: true }),
    map(generateStats),
    tap(() => count++),
    reduce(mergeGraphStats, {})
  )

Hope you find it as useful as we do.

oliverx07y ago

Pretty cool. Could the following be used as a baseline to create the same functionality?

    async function * nextStreamEntry() { 
      while(true){
        const entry = yield;
        const result = await processEntry(entry);
      }
    }

    // Initialize the iterator
    const iter = nextStreamEntry();
    iter.next();

    // For every entry of the stream
    for (const monster of results) {
      iter.next(monster)
    }

reconbotOP7y ago

yes!

https://github.com/reconbot/streaming-iterables

j / k navigate · click thread line to collapse