A lightweight TypeScript library providing reactive primitives for building event-driven applications. Designed as a minimal abstraction layer for reactive programming with excellent type safety.
The original goal behind this library was to create a re-usable set of helpers that can be easily combined with each other to hooking up various kinds of synchronous or asynchronous data sources to React.
npm install @janv/streamsA stream is a simple event emitter function that can be subscribed to:
type Source<T> = (handler: (value: T) => void) => () => void;Streams are:
- Lazy - nothing happens until a handler subscribes
- Lightweight - passing streams around is essentially "free"
- Composable - operators transform streams into new streams
Streams make no guarantees about asynchronicity — a handler may be called synchronously or asynchronously depending on the implementation. This is intentionally different from Promises, which always resolve asynchronously. Certain streams can choose to make such guarantees based on their use case, but the concept itself does not require it.
From the perspective of a consumer, it is impossible to distinguish between a direct source (such as returned from makeSource) and a source that has been extended with operators.
import { makeSource } from '@janv/streams';
const source = makeSource<number>();
// Subscribe to the stream
const unsubscribe = source((value) => {
console.log('Received:', value);
});
// Emit values
source.emit(1);
source.emit(2, 3, 4);
// Cleanup
unsubscribe();Streams are one-directional. Channels extend streams with bidirectional communication: a subscribe method (a stream), an optional update function, and an optional synchronous value getter.
interface Channel<Out, In = Out> {
subscribe: Source<Out>; // Stream for reading values
update?: (v: In) => void; // Method to send values back
value?: () => Out; // Synchronous getter for current value
}The value getter is provided as a function (not a plain value) so that a channel can be passed around and re-used while always returning an up-to-date current value. It is optional because there are scenarios where a synchronous default makes no sense or cannot be determined. If an initial value can be provided synchronously, the source can simply emit it on the stream once ready.
The update function is also optional — there are cases where a read-only channel with a default value is all that's needed. For the common read-write case, ReadWriteChannel makes update required.
It is not strictly necessary for the input and output types to be identical, though it is common in practice.
import { makeValueChannel } from '@janv/streams';
const channel = makeValueChannel(() => 'initial');
// Subscribe to changes
const unsubscribe = channel.subscribe((value) => {
console.log('New value:', value);
});
// Get current value synchronously
console.log(channel.value()); // 'initial'
// Update the value
channel.update('updated');
unsubscribe();BufferedChannel is a stateful ReadWriteChannel that synchronizes values between the stream and the update method to provide an optimized synchronous UI experience, even when the underlying source is asynchronous.
Values sent through update are immediately emitted back to the stream, unless updating the source causes it to synchronously emit a value (avoiding double-emit). A BufferedChannel also supports updater functions in addition to plain values. Because it caches the current value, the input and output types must be identical.
Operators transform streams. Use the pipe function to compose them:
import { pipe, filter, map, memo } from '@janv/streams';
const processed = pipe(
source,
filter((n) => n > 0),
map((n) => n * 2),
memo() // deduplicate consecutive identical values
);| Operator | Description |
|---|---|
filter(predicate) |
Remove values that don't match the predicate |
map(fn) |
Transform values using a mapping function |
mapAsync(fn) |
Async map supporting promises; always delivers results asynchronously |
startWith(value) |
Guarantee handler invocation with initial value |
memo(equals?) |
Deduplicate consecutive identical values |
fan |
Efficient fan-out for multiple subscribers (subscribes to source only once) |
switchAll |
Flatten nested streams (stream of streams to stream) |
merge(...sources) |
Combine multiple sources into one |
combineLatest(...sources) |
Combine latest values from multiple sources |
tap(onValue?, onSubscribe?, onUnsubscribe?) |
Side effect hooks |
effect(afterFirstSubscribe) |
Resource management (runs on first subscription, cleanup on last unsubscribe) |
A common pattern for working with dynamic stream sources:
import { pipe, startWith, map, switchAll } from '@janv/streams';
// Given a stream of objects that each contain a stream
const modeSource = pipe(
sceneStream,
startWith(() => getScene()),
map((scene) => startWith(() => getCurrentMode())(scene.mode)),
switchAll
);Use fan when multiple handlers need to subscribe to the same processed stream:
Without fan:
+--> Filter --> Map --> Consumer
Source |
+--> Filter --> Map --> Consumer
With fan:
+--> Consumer
Source --> Filter --> Map --> fan
+--> Consumer
Different channel interfaces for various use cases:
| Type | subscribe |
update |
value |
|---|---|---|---|
Channel<Out, In> |
Yes | Optional | Optional |
ChannelSync<Out, In> |
Yes | Optional | Required |
ReadOnlyChannel<Out> |
Yes | No | Optional |
ReadWriteChannel<Out, In> |
Yes | Required | Optional |
ReadWriteChannelSync<Out, In> |
Yes | Required | Required |
Create a custom event source:
const source = makeSource<number>(
(emit) => emit(initialValue), // Called when handler subscribes
() => console.log('Unsubscribed')
);
source.emit(42); // Emit to all handlers
source.handlers; // Set of current handlersCreate a simple read-write channel around a single value:
const channel = makeValueChannel(() => 0);
channel.value(); // Get current value
channel.update(5); // Update and emitThe library is fully typed with comprehensive generic support:
import type {
Source,
Handler,
UnsubscribeFn,
Operator,
Channel,
ChannelSync,
ReadOnlyChannel,
ReadWriteChannel,
ReadWriteChannelSync,
} from '@janv/streams';import { useStream, useChannel } from '@janv/streams/react';Subscribes a component to a Source<T> and re-renders on each emission. Accepts an optional initial value that is used until the stream emits.
// Without initial value — returns T | undefined until first emission
const count = useStream(countSource);
// With initial value — always returns T
const count = useStream(countSource, 0);
const count = useStream(countSource, () => expensiveDefault());Subscribes to a Channel and returns [value, update?]. The hook understands the channel type hierarchy and narrows the return type accordingly:
// ReadWriteChannelSync — value and update guaranteed, no default needed
const [value, update] = useChannel(readWriteSyncChannel);
// ReadWriteChannel with explicit default — value and update guaranteed
const [value, update] = useChannel(readWriteChannel, 'default');
// ReadWriteChannel without default — value may be undefined
const [value, update] = useChannel(readWriteChannel);
// ReadOnlyChannelSync — value guaranteed, no update
const [value] = useChannel(readOnlySync);
// ReadOnlyChannel with explicit default — value guaranteed
const [value] = useChannel(readOnlyChannel, 'default');
// Channel may be falsy (null/undefined/false/0/'') — hook suspends
const [value, update] = useChannel(maybeChannel, 'default');Synchronous channels (value() present): useChannel always reads the current value directly from channel.value() rather than relying on React state. This avoids stale-value issues even when the channel updates synchronously outside of React's render cycle.
Channel identity: the hook re-subscribes whenever the channel reference changes, similar to how useEffect handles dependencies.
# Install dependencies
npm install
# Run tests
npm test
# Build
npm run buildMIT