@repeaterjs/repeater

What is @repeaterjs/repeater?

The @repeaterjs/repeater package provides a way to create and work with async iterators and generators in JavaScript. It allows for the creation of push-based streams that can be used with async/await and for..of loops, making it easier to handle asynchronous data flows.

What are @repeaterjs/repeater's main functionalities?

Creating Repeaters

This feature allows you to create a new Repeater instance. The constructor takes an executor function that receives two arguments: push and stop. You can push values to the repeater and stop it when needed.

const { Repeater } = require('@repeaterjs/repeater');

const repeater = new Repeater(async (push, stop) => {
  push('Hello, Repeater!');
  stop();
});

(async () => {
  for await (const item of repeater) {
    console.log(item);
  }
})();

Handling Asynchronous Events

This code sample demonstrates how to create a repeater that listens for and handles asynchronous events, such as DOM events. The repeater pushes click events to the consuming async iterator.

const { Repeater } = require('@repeaterjs/repeater');

const eventRepeater = new Repeater(async (push, stop) => {
  document.addEventListener('click', push);
  await stop;
  document.removeEventListener('click', push);
});

(async () => {
  for await (const event of eventRepeater) {
    console.log('Clicked at:', event.clientX, event.clientY);
  }
})();

Combining Multiple Sources

This example shows how to combine multiple asynchronous sources into a single repeater. In this case, a timeout is used to push a value after a delay, but other async sources could be combined in a similar manner.

const { Repeater } = require('@repeaterjs/repeater');

const combinedRepeater = new Repeater(async (push, stop) => {
  const timeoutId = setTimeout(push, 1000, 'Timeout!');
  try {
    await stop;
  } finally {
    clearTimeout(timeoutId);
  }
});

(async () => {
  for await (const item of combinedRepeater) {
    console.log(item);
  }
})();

Other packages similar to @repeaterjs/repeater

rxjs

RxJS is a library for reactive programming using Observables, to make it easier to compose asynchronous or callback-based code. It is more feature-rich and has operators for combining, creating, and transforming streams, which makes it more complex but also more powerful than @repeaterjs/repeater.

most

Most.js is a high-performance reactive programming library. It provides a rich set of stream combinators and transformation operators similar to RxJS. It is focused on high performance and low memory usage, which might make it a better choice for performance-critical applications compared to @repeaterjs/repeater.

async-iterators

The async-iterators package provides utilities for working with asynchronous iterators. It offers a simpler API compared to @repeaterjs/repeater but lacks some of the more advanced control mechanisms such as the ability to stop the iterator from the producer side.

README

@channel/channel

The missing constructor for creating safe async iterators

For more information, visit channel.js.org.

API

class Channel<T> implements AsyncIterableIterator<T> {
  constructor(executor: ChannelExecutor<T>, buffer?: ChannelBuffer<T>);
  next(value?: any): Promise<IteratorResult<T>>;
  return(value?: any): Promise<IteratorResult<T>>;
  throw(error: any): Promise<IteratorResult<T>>;
  [Symbol.asyncIterator](): this;
}

The Channel class implements the AsyncIterableIterator interface. Channels are designed to be indistinguishable from async generator objects.

type Push<T> = (value: PromiseLike<T> | T) => Promise<any | void>;

interface Stop extends Promise<any | void> {
  (error?: any): void;
}

type ChannelExecutor<T> = (
  push: Push<T>,
  stop: Stop
) => Promise<T | void> | T | void;

The ChannelExecutor is passed the arguments push and stop.

push is a function which allows you to enqueue values onto the channel. It synchronously throws an error if there are too many pending pushes on the channel (currently set to 1024). It returns a promise which resolves when it’s safe to push more values.

stop is a both a promise and a function. As a function, stop can be called to stop a channel. Calling stop without any arguments stops the channel without error, and passing an error both stops the channel and causes the final iteration to reject with that error.

As a promise, stop can be awaited to defer event handler cleanup, and it can also be used with Promise.race to abort pending promises. If you pass a value to Channel.prototype.return, stop will resolve to that value.

The value of the final interation of the channel will be the return value of the executor. If the executor throws an error or returns a promise rejection, the channel will be immediately stopped and the final iteration will throw.

interface ChannelBuffer<T> {
  full: boolean;
  empty: boolean;
  add(value: T): void;
  remove(): T | undefined;
}

class FixedBuffer<T> implements ChannelBuffer<T> {
  constructor(capacity: number);
}

class SlidingBuffer<T> implements ChannelBuffer<T> {
  constructor(capacity: number);
}

class DroppingBuffer<T> implements ChannelBuffer<T> {
  constructor(capacity: number);
}

The Channel constructor optionally takes a ChannelBuffer instance as its second argument. Buffers allow multiple values to be pushed onto channels without waiting. FixedBuffer allows channels to push a set number of values, DroppingBuffer drops the latest values when the buffer has reached capacity, and SlidingBuffer drops the earliest values when the buffer has reached capacity. You can define custom buffering behaviors by implementing the ChannelBuffer interface.