@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

Repeater.js

The missing constructor for creating safe async iterators.

For more information, visit repeater.js.org.

Installation

Repeater.js is available on npm in the CommonJS and ESModule formats.

$ npm install @repeaterjs/repeater

$ yarn add @repeaterjs/repeater

Requirements

The core @repeaterjs/repeater module has no dependencies, but requires the following globals in order to work:

• Promise
• WeakMap
• Symbol
  • Symbol.iterator
  • Symbol.asyncIterator

In addition, repeaters are most useful when used via async/await and for await…of syntax. You can transpile your code with babel or typescript to support enviroments which lack these features.

Examples

Logging timestamps with setInterval

import { Repeater } from "@repeaterjs/repeater";

const timestamps = new Repeater(async (push, stop) => {
  push(Date.now());
  const interval = setInterval(() => push(Date.now()), 1000);
  await stop;
  clearInterval(interval);
});

(async function() {
  let i = 0;
  for await (const timestamp of timestamps) {
    console.log(timestamp);
    i++;
    if (i >= 10) {
      console.log("ALL DONE!");
      break; // triggers clearInterval above
    }
  }
})();

Creating a repeater from a websocket

import { Repeater } from "@repeaterjs/repeater";

const socket = new WebSocket("ws://echo.websocket.org");
const messages = new Repeater(async (push, stop) => {
  socket.onmessage = (ev) => push(ev.data);
  socket.onerror = () => stop(new Error("WebSocket error"));
  socket.onclose = () => stop();
  await stop;
  socket.close();
});

(async function() {
  for await (const message of messages) {
    console.log(message);
    if (message === "close") {
      console.log("Closing!");
      break; // closes the socket
    }
  }
})();

socket.onopen = () => {
  socket.send("hello"); // "hello"
  socket.send("world"); // "world"
  socket.send("close"); // "close", "Closing!"
};

Listening for the Konami Code and canceling if Escape is pressed

import { Repeater } from "@repeaterjs/repeater";

const keys = new Repeater(async (push, stop) => {
  const listener = (ev) => {
    if (ev.key === "Escape") {
      stop();
    } else {
      push(ev.key);
    }
  };
  window.addEventListener("keyup", listener);
  await stop;
  window.removeEventListener("keyup", listener);
});

const konami = ["ArrowUp", "ArrowUp", "ArrowDown", "ArrowDown", "ArrowLeft", "ArrowRight", "ArrowLeft", "ArrowRight", "b", "a"];

(async function() {
  let i = 0;
  for await (const key of keys) {
    if (key === konami[i]) {
      i++;
    } else {
      i = 0;
    }
    if (i >= konami.length) {
      console.log("KONAMI!!!");
      break; // removes the keyup listener
    }
  }
})();

Converting an observable to an async iterator

import { Subject } from "rxjs";
import { Repeater } from "@repeaterjs/repeater";

const observable = new Subject();
const repeater = new Repeater(async (push, stop) => {
  const subscription = observable.subscribe({
    next: (value) => push(value),
    error: (err) => stop(err),
    complete: () => stop(),
  });
  await stop;
  subscription.unsubscribe();
});

(async function() {
  try {
    for await (const value of repeater) {
      console.log("Value: ", value);
    }
  } catch (err) {
    console.log("Error caught: ", err);
  }
})();
observable.next(1);
// Value: 1
observable.next(2);
// Value: 2
observable.error(new Error("Hello from observable"));
// Error caught: Error: Hello from observable

Changelog

repeater@3.0.6 - 2024-05-08

• Fix promise returned from push function not resolving when using a buffer.