Stream

Stream: object

Defined in: Core/Stream.ts:42

Type Declaration

emit

emit: <S, K>(target, message) => void = emitStream

Emits a message payload to one or more target streams.

Uses a synchronous breadth-first trampoline queue to handle re-entrant emissions deterministically.

Type Parameters

S

S extends Record<string, unknown>

K

K extends string

Parameters

target

Stream<S> | readonly Stream<S>[]

message

WithKind<K> & WithValue<S[K]>

Returns

void

Example

Stream.emit(streamA, {
  kind: "userLoggedIn",
  value: { userId: "user-1" },
});

Stream.emit([streamA, streamB], {
  kind: "userLoggedIn",
  value: { userId: "user-1" },
});

forward

forward: <S>(options) => () => void = forwardStream

Forwards messages from one stream to another (or multiple).

Type Parameters

S

S extends Record<string, unknown>

Parameters

options

ForwardOptions<S>

Returns

() => void

Example

const stop = Stream.forward({
  from: authStream,
  to: analyticsStream,
  only: ["userLoggedIn"],
});

listen

listen: <S, K>(stream, events, options?) => ListenerBuilder<S> = listenStream

Initiates listener registration on a stream for specific event kind(s) or sequence.

Type Parameters

S

S extends Record<string, unknown>

K

K extends string

Parameters

stream

Stream<S>

events

K | readonly K[]

options?

SequenceOptions<S>

Returns

ListenerBuilder<S>

Example

const sub = Stream.listen(
  appStream,
  ["userLoggedIn", "checkoutStarted"],
  { ordered: true }
).reduce(
  (msg, state) => ({ count: state.count + 1 }),
  { count: 0 }
);

make

make: <S>(options?) => Stream<S> = makeStream

Constructs a new Stream instance.

Type Parameters

S

S extends Record<string, unknown>

Parameters

options?

Options

Returns

Stream<S>

Example

const stream = Stream.make<AppMessages>({ name: "app" });