IxJS (Interactive Extensions for JavaScript)

The AsyncIterable object is based on the ECMAScript Asynchronous Iterators proposal. It allows you to create asynchronous collections of Promises and apply operators like map and filter.

Iteration is performed using the for await ... of statement. You can also use .forEach() and .catch() for handling values and errors in an asynchronous stream.

// ES
import { from } from 'ix/asynciterable';
import { filter, map } from 'ix/asynciterable/operators';

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

const results = from(source()).pipe(
  filter(async x => x % 2 === 0),
  map(async x => x * x)
);

for await (let item of results) {
  console.log(`Next: ${item}`);
}

IxJS extends the AsyncIterable concept (ES2018) with a standard library of factories and operators.

1. Pipeable Operators: Import factories from 'ix/asynciterable' and operators from 'ix/asynciterable/operators'. Use the .pipe() method.
2. Fluent/Dot-notation: Import AsyncIterableX from 'ix/asynciterable' and add specific modules via 'ix/add/asynciterable/<name>' and 'ix/add/asynciterable-operators/<name>'.

Async iterables are consumed using for await ... of loops or the forEach method.

import { as } from 'ix/asynciterable';
import { map } from 'ix/asynciterable/operators';

const soureFactory = async function*() {
  yield 1;
  yield 2;
  yield 3;
  yield 4;
};

const source = as(sourceFactory());
const result = source.pipe(
  map(async x => x * x)
);

for await (const item of results) {
  console.log(`Next: ${item}`);
}

The Iterable class allows you to compose synchronous collections using Array#extras style methods like map, filter, and reduce. You can iterate over these collections using a standard for ... of loop or the provided forEach method.

To keep bundle sizes small, you can import specific operators from ix/iterable/operators and use the .pipe() method, or you can add operators directly to the IterableX prototype using the ix/add modules.

// Using pipe with specific operators
import { from } from 'ix/iterable';
import { filter, map } from 'ix/iterable/operators';

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

const results = from(source()).pipe(
  filter(x => x % 2 === 0),
  map(x => x * x)
);

for (let item of results) {
  console.log(`Next: ${item}`);
}

IxJS provides npm scripts to manage the build lifecycle. You can run these scripts against specific targets (the JavaScript version/environment) and modules (the module format).

Available Targets (-t or --targets):

• ix: The main Ix target module bundle.
• ts: TypeScript.
• es5, es2015, esnext: Various ECMAScript versions.
• all: Builds all targets (default).

Available Modules (-m or --modules):

• cjs: CommonJS.
• esm: ECMAScript Modules.
• umd: Universal Module Definition.
• all: All module formats (default).

Scripts:

• npm run clean: Cleans build targets.
• npm run build: Cleans and compiles all targets.
• npm test: Executes tests against built targets.

To install the full IxJS library, use the following command:

npm install ix

To use a fluent API for synchronous iterables, import IterableX and then import only the specific modules needed for the methods you want to use. This keeps your bundle size small.

• Factories: 'ix/add/iterable/<name>'
• Operators: 'ix/add/iterable-operators/<name>'

import { IterableX as Iterable } from 'ix/iterable';

// Add factory and operators
import 'ix/add/iterable/of';
import 'ix/add/iterable-operators/map';

const source = Iterable.of(1, 2, 3, 4, 5);
const result = source.map(x => x);

for (const item of result) {
  console.log(`Next: ${item}`);
}

To use the fluent API for asynchronous iterables, import AsyncIterableX and add specific modules:

• Factories: 'ix/add/asynciterable/<name>'
• Operators: 'ix/add/asynciterable-operators/<name>'

import { AsyncIterableX as AsyncIterable } from 'ix/asynciterable';
import 'ix/add/asynciterable/as';
import 'ix/add/asynciterable/last';
import 'ix/add/asynciterable-operators/map';
import 'ix/add/asynciterable-operators/withabort';

const sourceFactory = async function*() {
  yield 1;
  yield 2;
  yield 3;
  yield 4;
};

const controller = new AbortController();
const result = await as(sourceFactory())
  .withAbort(controller.signal)
  .map(async x => x * x)
  .last({ signal: controller.signal });

The bufferCountOrTime operator allows you to process items in batches based on either a maximum number of items or a maximum time interval, whichever comes first. This is useful for handling message streams where you want to ensure batches are processed promptly even if the message rate slows down, preventing items from being stuck in a buffer waiting for a specific count to be reached.

To use it, pipe your subscription into bufferCountOrTime(count, timeMs) and then handle the resulting batches (which are arrays of items).

await subscription.pipe(
  // emit when buffer hits 16 items, or every 100ms
  bufferCountOrTime(16, 100)
)
.forEach(handleBatch)

Instead of using .pipe(), you can add operators directly to the IterableX prototype to enable chainable syntax. This is useful for reducing bundle size by only importing what you need.

// ES
import { IterableX as Iterable } from 'ix/iterable';
import 'ix/add/iterable/of';
import 'ix/add/iterable-operators/map';

const results = Iterable.of(1,2,3)
  .map(x => x + '!!');