rstream recipes

@thi.ng/rstream recipes

Resettable countdown / timeout stream

Metastreams are a powerful tool to build various higher-order stream operators/constructs. Here's an example to create a resettable countdown sequence...

import { fromIterable, metaStream, trace } from "@thi.ng/rstream";
import { range } from "@thi.ng/transducers";

const countdown = (from: number, delay = 1000) =>
    metaStream(() => fromIterable(range(from, -1), delay));

// countdown from 10 @ 1 Hz
const a = countdown(10, 1000);
a.subscribe(trace());

// kick off
a.next();
// 10
// 9
// 8
// 7

// reset
a.next();
// 10
// ...
// 0

// can be re-triggered any time
a.next();
// 10
// 9
// ...

Alternatively one can similarly build a resettable timeout utility stream, e.g. as sidechain/control input for other stream constructs. Each time a value is being sent to the stream, the timeout resets and once triggered, emits the user defined value.

const timeout = <T>(value: T, delay = 1000) =>
  metaStream(() => fromIterable([value], delay));

const a = timeout("foo", 1000);
a.subscribe(trace());

// kick off
a.next();

// 1sec later
// foo

Interpolating stream values

The tween operator takes an existing stream src and attaches a new subscription which interpolates between incoming values from src using the given mix function. The returned construct produces values at a rate controlled by the clock stream or frequency. If omitted, clock defaults to fromRAF() (~60Hz). If given as number, creates a fromInterval(clock) or else uses the given clock stream directly. In general, the frequency of the clock should always be higher than that of src.

If stop is given as well, no values will be passed downstream if that function returns true. This can be used to limit traffic once the tween target value has been reached.

The returned subscription closes automatically when either src or clock is exhausted.

import { stream, tween, trace } from "@thi.ng/rstream";

// input stream
const val = stream<number>();

tween(
  // consume from `val` stream
  val,
  // initial start value to interpolate from
  0,
  // interpolation fn (LERP)
  (a, b) => a + (b - a) * 0.5,
  // stop emitting values if difference to previous result < 0.01
  (a, b) => Math.abs(a - b) < 0.01,
  16
).subscribe(trace());

// kick off
val.next(10)
// 5
// 7.5
// ...
// 9.98046875

val.next(100)
// 55
// 77.5
// ...
// 99.989013671875

State tracking of multiple keys

For games and other interactive applications needing to track the state of multiple keys on the keyboard, the following function provides a stream of key state objects representing the pressed state of selected keys (and only those). The returned construct is merging two keyboard event streams attached to target (e.g. window) and then immutably updates and emits a key state object with each qualifying keydown or keyup event.

import { fromEvent, merge } from "@thi.ng/rstream";
import { comp, filter, reducer, scan } from "@thi.ng/transducers";

const keyStateStream = (target: EventTarget, keys: Set<string>) =>
    merge({
        src: [
            fromEvent(target, "keydown"),
            fromEvent(target, "keyup")
        ],
        xform: comp(
            filter((e: KeyboardEvent) => keys.has(e.key.toLowerCase())),
            scan(
                reducer(
                    () => ({}),
                    (acc, e) => ({
                        ...acc,
                        [e.key.toLowerCase()]: e.type !== "keyup"
                    })
                )
            )
        )
    });

// create a stream of key states for the given keys
keyStateStream(new Set(["w", "a", "s", "d", "shift", "alt"]))
  .subscribe({ next: console.log });

// { shift: true }
// { shift: true, w: true }
// { shift: true, w: false }
// { shift: false, w: false }
// { shift: false, w: false, a: true }
// { shift: false, w: false, a: false }