subject

push items to pulling iterators

Install

npm install @async-generators/subject --save
yarn add @async-generators/subject

This package's main entry points to a commonjs dist. The module entry points to a es2015 module dist. Both require native async-generator support, or be down-compiled with a webpack loader.

Api

Soupler()

Subject that provides three methods: next(item), error(err), and done() to push data and events. When [Symbol.asyncIterator] is called an internal subscription is created. pushed items are buffered (per iterator) until they are pulled by the consuming iterator. If the consuming iterator pulls items slower than the speed they are pushed to the subject, then the internal buffer will continue to grow in size. error(err) will cause the iterator to rethrow the given error to the consumer and dispose of the subject.

Example

example.js

const {Subject} = require('@async-generators/subject');

let subject = new Subject<number>();

async function* source(){
  yield 1; yield 2; yield 3; yield 4;
}

async function main() {
  let reader = async function () {
    for await (let item of subject) {
      console.log("PULL:", item);
    }
  };

  let readers = [reader(), reader()];
  
  let writer = async function () {
    for await (let item of source()){
      console.log("PUSH:", item);
      subject.next(item);
    }
    subject.done();
  }

  await writer();
  await Promise.all(readers);
}

main().catch(console.log);

Execute with the latest node.js 9:

node --harmony example.js

output:

PUSH: 1
PUSH: 2
PULL: 1
PULL: 1
PUSH: 3
PULL: 2
PULL: 2
PUSH: 4
PULL: 3
PULL: 3
PULL: 4
PULL: 4

Typescript

This library is fully typed and can be imported using:

import {Subject} from '@async-generators/subject');

It is also possible to directly execute your properly configured typescript with ts-node:

ts-node --harmony foo.ts