Typed streams#

@craft-ts/stream is a small, typed stream library built on the same ideas as the rest of craft-ts: every operator adds its dependencies and its exceptions to the type of the stream, so you can read them off where the pipeline is finally consumed.

Use it when you want operators (map, filter, scan, share, …) over something that emits over time — and you want the compiler, not a comment, to tell you which services it needs and which typed exceptions it may end with. Not for one-shot async work (that is a query loader) or plain state (that is state).

Import#

import { of, map, filter, catchTag, subscribe } from '@craft-ts/stream';

@craft-ts/stream depends on @craft-ts/core only. craft-ts itself has no RxJS dependency.

The CraftStream<A, Y> type#

interface CraftStream<A, Y = never> extends Subscribable<A> {
  pipe: CraftStreamPipe<A, Y>;
}

• A is the type of the values.

• Y is the same Yielded union a craftGen program carries: the service-dependency requests the stream's handlers yield*, and a marker per typed exception the stream may terminate with.

A stream is cold: every subscription re-runs it. Make it hot with share() / shareReplay(), or by starting from a core subject().

Three terminal notifications#

An Observable's error is a defect. Exceptions are the typed channel.

Sources#

of(1, 2, 3); // emits then completes
empty(); // completes immediately
fail(craftException({ _tag: 'NotFound' }, { id: 1 })); // typed exception

fromSubscribable(subject); // a core subject, an RxJS Observable, ...
fromSource(reset$); // a source$ — its service dependency stays tracked

concat(a$, b$); // one after the other
defer(() => of(Date.now())); // built afresh for each subscriber
throwError(() => new Error('x')); // a DEFECT (use fail(...) for a typed exception)
never(); // neither emits nor completes

from([1, 2, 3]); // an array, a Set… (also: a Promise, a Subscribable, another stream)
fromPromise(fetchUser()); // resolved value, then complete; a rejection is a defect
generate({ initialState: 1, condition: (n) => n < 100, iterate: (n) => n * 2 });
bindCallback(fs.exists)('/tmp'); // a callback-style function as a stream
fromFetch('/api/items'); // the Response (abort on unsubscribe); an HTTP error is a value

interval(1_000); // 0, 1, 2… every second (driven by the temporal runtime)
timer(500); // one value after 500 ms
timer(500, 100); // then every 100 ms
fromEvent(button, 'click'); // DOM events; the listener is removed on unsubscribe

A core subject<T, E>() declares its failure set E; fromSubscribable turns it into the stream's exception type.

Operators#

Operators go in .pipe(...), up to 14 per call (nest .pipe(...).pipe(...) beyond that). Each slot is inferred from the previous one.

const squares = of(1, 2, 3, 4).pipe(
  map((n) => n * n),
  filter((n) => n > 1),
  scan((total, n) => total + n, 0),
  take(2),
);

Creators beyond of/from/interval: range, iif (a branch per subscription), forkJoin (the last value of each input, as a tuple or record) and partition (two cold halves by a predicate; a type guard narrows both).

A stream that cannot give the value an operator promises ends with a typed defect, never a hang: EmptyStreamError (first, last, single, throwIfEmpty), StreamSequenceError (single found several), StreamOutOfRangeError (elementAt). min/max/forkJoin of an empty input simply complete without a value.

Handlers that yield services#

A generator callback can yield* a craft service. The service is resolved when the stream runs, and it is added to the stream's type:

const prices = of('EUR', 'USD').pipe(
  map(function* (currency) {
    const rates = yield* Rates();
    return rates.convert(currency);
  }),
); // CraftStream<number, ...the Rates dependency>

The stream resolves services from the injector it runs with: subscribe inside an injection context, or pass { injector } (see Terminals). Without one, a handler that yield*s a service fails with a clear error.

Handlers run one at a time, in arrival order. If one suspends (a craftSleep, a promise), later values wait; there is no hidden buffer and no concurrency. A handler that returns a craftException ends the stream with that exception.

Flattening#

A projection returns an inner stream; the inner values are flattened into one stream. The four operators differ in what they do when an outer value arrives while an inner stream is still running:

const results$ = query$.pipe(
  switchMap(function* (text) {
    const { Search } = yield* SearchService();
    return Search.results(text); // a stream: its service joins the type
  }),
);

The inner streams' dependencies and exceptions join the outer type, and the projection may be a generator whose own yield*s join it too. A projection may also return a plain Subscribable (an RxJS Observable, a core subject).

Nothing is buffered implicitly. concatMap and a bounded mergeMap keep a queue, unbounded unless you bound it:

concatMap(save, { buffer: 10, overflow: 'drop-oldest' });
// overflow: 'error' (default: a defect), 'drop-newest' or 'drop-oldest'
mergeMap(fetch, { concurrency: 4 });

Combining#

combineLatest([a$, b$]); // CraftStream<[A, B]>
combineLatest({ a: a$, b: b$ }); // CraftStream<{ a: A; b: B }>
merge(a$, b$); // CraftStream<A | B>
zip(a$, b$); // pairs by position (buffers the faster input — inherent to zip)
race(a$, b$); // mirrors whichever speaks first
a$.pipe(withLatestFrom(b$)); // CraftStream<[A, B]>

The result's type is the union of every input's: all their dependencies and exceptions. An exception or defect in any input ends the result. Inputs must be craft streams — wrap anything else with fromSubscribable(...).

Time#

Every timer goes through the temporal runtime: it is cancelled with the subscription and the owner's DestroyRef, and driven by the virtual clock in tests. There is no raw setTimeout.

input$.pipe(debounce(300)); // latest value after 300 ms of silence
input$.pipe(throttle(100, { leading: true, trailing: true }));
input$.pipe(auditTime(100)); // the latest value at the END of each window
input$.pipe(delay(50));
input$.pipe(timeout(5_000)); // raises a typed StreamTimeout exception
request$.pipe(retry({ times: 3, backoff: 'exponential', delayMs: 200 }));
poll$.pipe(repeat({ delayMs: 10_000 }));

• timeout raises a typed StreamTimeout exception ({ ms } payload) — catch it with catchTag('StreamTimeout', …) — rather than throwing a defect.

• retry re-runs the stream on a typed exception, with the same policy as the program retry (times, while on exception tags, backoff, delayMs, schedule). A defect is never retried.

• repeat re-runs on completion. A synchronous source repeated forever never yields control: bound it (times, take, takeUntil) or wait between runs.

• These operators add a RuntimeTemporalAwaitRequest to the stream's type: it is asynchronous, so a synchronous host cannot consume it in place.

Grouping and sampling#

input$.pipe(bufferCount(3)); // arrays of 3
input$.pipe(bufferTime(1_000)); // what arrived in each second
input$.pipe(buffer(flush$)); // until flush$ emits
input$.pipe(bufferWhen(() => timer(1_000))); // a fresh closing stream after each flush
input$.pipe(windowCount(10)); // a hot inner stream per 10 values
input$.pipe(sample(tick$)); // the latest value on each tick
input$.pipe(groupBy((event) => event.kind)); // a hot stream per key
input$.pipe(window(open$)); // a hot inner stream per window

groupBy and window hand out hot inner streams: subscribe to each one as soon as you receive it.

Catching typed exceptions#

catchTag is typed from the stream it is piped onto — the handler's parameter is the exception with that tag, and a tag the stream cannot produce is a compile error:

declare const user$: CraftStream<User, CraftGenExceptionMarker<NotFound | Forbidden>>;

const safe$ = user$.pipe(
  catchTag('NotFound', (e) => GUEST), // e: NotFound
);
// CraftStream<User | typeof GUEST, CraftGenExceptionMarker<Forbidden>>

The handler's result is emitted and the stream completes; NotFound leaves the exception union. catchTag.exhaustive takes a handler per remaining tag and rejects a missing or unreachable one:

user$.pipe(
  catchTag.exhaustive({
    NotFound: () => GUEST,
    Forbidden: () => GUEST,
  }),
); // exceptions: never

mapException rewrites exceptions; orElse continues with another stream.

Sharing#

const prices$ = source$.pipe(map(...), shareReplay(1));

share() connects to upstream on the first subscriber and disconnects when the last one leaves. shareReplay(n) replays the last n values to late subscribers and keeps the connection. Apply them in an injection context: the upstream run uses the Injector and DestroyRef captured when the operator is applied.

When the moment of connection matters, make it explicit:

const live = connectable(prices$, { connector: () => replaySubject(1) });
const view = live.pipe(map(format)); // subscribers attach first…
const connection = live.connect(); // …then the source starts
connection.unsubscribe(); // disconnects

publish(), publishReplay(n), publishBehavior(v) and multicast(subject) build a connectable stream; paired with refCount() they are share(). Note that .pipe(…) types its result as a plain stream, which hides connect — call the operator directly (publish()(source$)) or use connectable(...) when you need connect().

connect shares the source with a selector that uses it several times — the source is connected once the selector's stream is subscribed, so a synchronous source reaches every use:

source$.pipe(
  connect((shared) => merge(shared.pipe(take(2)), shared.pipe(skip(10)))),
);

Network sources#

ajax.getJSON<User[]>('/api/users'); // the parsed body, then complete
ajax.post('/api/users', { name: 'Ada' }); // a plain object is sent as JSON → AjaxResponse
ajax({ url: '/slow', timeout: 5_000 }); // aborts after 5 s (temporal runtime)

const socket = webSocket<Message>('wss://example.test/feed');
socket.pipe(map((message) => message.text)).subscribe({ next: render });
socket.next({ text: 'hi' }); // sends (queued until the socket is open)

An HTTP error status is an AjaxError defect (it carries status and response); unsubscribing aborts the request. A webSocket opens when its first subscriber arrives and closes when the last one leaves; a clean close completes the stream, an unclean one is a defect. Application HTTP belongs to query / mutation / server functions — they carry loading state, caching and typed exceptions; these are the stream-shaped primitives for the cases that really are a stream.

Notifications as values, and the clock#

source$.pipe(materialize()); // { kind: 'N', value } | { kind: 'X', exception } | { kind: 'E', error } | { kind: 'C' }
notes$.pipe(dematerialize()); // and back
source$.pipe(observeOn(0)); // re-deliver every notification on the next turn of the clock
source$.pipe(subscribeOn(0)); // subscribe on the next turn of the clock

'X' is a typed exception (RxJS has no such notion) and 'E' a defect. A materialized stream no longer raises: its exceptions left its type and became values. There is exactly one scheduler, the temporal runtime, so observeOn and subscribeOn take a delay, not a scheduler.

Terminals#

A terminal is where the type is consumed.

subscribe#

subscribe(
  user$,
  {
    next: (user) => render(user),
    exception: {
      NotFound: (e) => showNotFound(e.payload.id),
      Forbidden: () => redirectToLogin(),
    },
  },
  { injector },
);

• When the stream may end with typed exceptions, exception is required: a function receiving the union, or a { [tag]: handler } map covering every tag.

• When the stream depends on services, { injector } is required; the compile error names the missing services.

• error receives defects only.

As a craft program#

lastValueFrom, firstValueFrom, toArray and runForEach turn a stream into a program you yield*. Its dependencies and exceptions become the program's:

const latest = yield* lastValueFrom(prices$);

A stream that settles synchronously returns without suspending; otherwise the program suspends and the subscription is released if the program is aborted or its injector destroyed.

streamSignal#

Expose a stream as a named primitive of a craftService:

const { Feed } = craftService({ name: 'Feed', providedIn: 'global' }, function* () {
  yield* streamSignal('latest', prices$);
});

const feed = craftUse(Feed());
feed.latest.value(); // the latest value
feed.latest.status(); // 'idle' | 'running' | 'completed' | 'exception' | 'error'
feed.latest.exception(); // the typed exception, or undefined

The stream's service dependencies are folded into the service's verified dependency tree, and its exceptions into the host's exception types.

Observability#

Streams plug into the same cross-cutting mechanisms as the rest of CraftTS (see Observability), with no change to the pipeline.

Handlers go through provideFnWrapper. Every handler an operator runs — map, mergeMap, catchTag, tap, … plain function or generator — is executed through the injector's function wrappers, like a service method or a query loader. Correlation tracking, provideTakeAppSnapshot and your own wrappers therefore see stream work. With no wrapper installed the handler runs as is, and a synchronous pipeline stays synchronous.

provideStreamTrace observes root subscriptions. A root is where a stream is consumed: subscribe, a program terminal, streamSignal, an adapter. Inner streams (mergeMap, switchMap, …) are part of their root, not roots themselves.

import { provideStreamTrace } from '@craft-ts/stream';

provideStreamTrace((event, context) => {
  // context: { streamId, name, root, startCorrelationId }
  // event.kind: 'subscribe' | 'next' | 'exception' | 'error' | 'complete' | 'unsubscribe'
  console.debug(context.name ?? context.streamId, event.kind);
});

• traceStage('label') marks a point inside a pipeline: every notification that passes it is reported as a stage event (stage, notification, value), so a long pipeline shows where a value, an exception or a defect went through. It never alters the stream and is a plain pass-through when nothing traces.

• name comes from the { name } option of subscribe / captureStreamContext; streamSignal passes its own.

• startCorrelationId is the user gesture that was current when the stream started — the link between "the user clicked Save" and "this feed emitted four seconds later".

• unsubscribe is reported only when the consumer leaves before any terminal notification.

• Observers are passive: they run in the stream's injector, cannot alter the stream, and one that throws is ignored.

source$.pipe(
  traceStage('raw'),
  map(parse),
  traceStage('parsed'),
)

A defect takes an app snapshot. A stream that ends with a defect (error, never a typed exception) triggers provideTakeAppSnapshot, even when the defect comes from a source rather than a handler. A handler that already went through a snapshot wrapper is not snapshotted a second time. streamSignal also appears in app snapshots, tagged with its host.

All of it is development-only, like provideCraftHttpTrace: under provideCraftProduction() a root subscription runs unwrapped.

Time and testing#

Every timer in @craft-ts/stream — interval, timer, debounce, throttle, delay, timeout, retry, repeat, the buffers, ajax's timeout, observeOn / subscribeOn — goes through the temporal runtime of the stream's context, the same one craftSleep, asyncProcess and query use. There is no raw setTimeout in the library. Two consequences:

• One clock to drive. A test activates a VirtualCraftTemporalRuntime once and every stream timer — and every other craft timer — follows it:

  tsCopy

  const clock = new VirtualCraftTemporalRuntime();
  const restore = activateCraftTemporalRuntime(clock);
  
  const seen: number[] = [];
  interval(100).pipe(take(3)).subscribe({ next: (n) => seen.push(n) });
  
  await clock.advanceBy(300);
  expect(seen).toEqual([0, 1, 2]);
  restore();
  

• Cleanup comes with the subscription. A timer is a task on the runtime, cancelled when the subscription ends or the owner's DestroyRef is destroyed; clock.pendingTasks() shows what is still waiting, which is how a test proves nothing leaked.

The wiring to native timers is covered too: operators.realtime.spec.ts runs interval, timer, debounce, delay, timeout, retry and bufferTime on the real runtime with short delays, asserting ordering and lower bounds only, and that cancelled work leaves no task behind.

Operators that wait add RuntimeTemporalAwaitRequest to the stream's Y: a stream that may suspend cannot be consumed in place by a synchronous host.

timestamp() reads the runtime's civil time and timeInterval() its monotonic time, so both are deterministic under the virtual clock.

Effect's clock#

Effect has its own Clock: an Effect.sleep, a Schedule or a Stream.schedule inside an Effect program keeps real time unless told otherwise, and would need real waiting (or Effect's TestClock) in a test. @craft-ts/stream-effect provides a Layer that points it at the temporal runtime:

import { provideLayer } from '@craft-ts/effect';
import { craftTemporalClock } from '@craft-ts/stream-effect';

provideLayer(craftTemporalClock());

It is opt-in, so apply it where Effect time and craft time should be the same time (an app, or a route). Effect.sleep then runs as a cancellable task on the runtime — interrupting the fiber cancels it — and a VirtualCraftTemporalRuntime drives it. Effect resumes its fibers on its own scheduler, so a test lets a few macrotasks pass after each advanceBy.

fromEffectSchedule does not need the Layer: it only asks a Schedule for its next delay and waits on the temporal runtime itself.

Interop with RxJS#

Subscribable is structural: an RxJS Observable is assignable to it, so fromSubscribable(observable$) needs no adapter. Going the other way, wrap the Subscribable returned by toSubscribable(stream):

const exposed = toSubscribable(stream$);
const observable$ = new Observable((subscriber) => exposed.subscribe(subscriber));

Effect interop#

@craft-ts/stream-effect adapts between a craft stream and Effect's Stream, Queue, PubSub, SubscriptionRef and Schedule. It is a separate package, so using @craft-ts/effect never forces @craft-ts/stream on you.

import { fromStream, toStream } from '@craft-ts/stream-effect';

const prices$ = fromStream(priceStream); // Stream -> CraftStream
const effectStream = toStream(prices$); // CraftStream -> Stream

• A typed Effect failure becomes a craft exception tagged with the error's _tag (so catchTag works on it); a defect (die) stays a defect; an interruption is neither.

• fromStream resolves the Effect injector level (provideLayer) at subscription, flattens chunks, and interrupts the fiber when unsubscribed or when the owner is destroyed. A stream with requirements the type cannot see provided does not compile.

• toStream has a bounded buffer (64, sliding by default): a craft stream pushes and cannot be told to wait, so back-pressure is lost at this boundary by nature.

• fromQueue, fromPubSub and fromSubscriptionRef read Effect's primitives. There is deliberately no automatic two-way bridge between a subject and a queue: it would echo.

• toEffectSchedule(craftSchedule) and fromEffectSchedule(effectSchedule) convert schedules; the latter only accepts pure ones (it steps them synchronously) and raises CraftScheduleNotPure otherwise.

Migrating from RxJS#

craft-ts no longer depends on RxJS. The craft-migrate-streams codemod rewrites RxJS pipelines to @craft-ts/stream:

npx craft-migrate-streams --dry-run
npx craft-migrate-streams --write

It is all-or-nothing per file: a file is rewritten only when every rxjs symbol it imports has an equivalent and every .pipe(...) receiver can be recognised; otherwise it is left untouched and a diagnostic names what blocks it: a symbol that is in none of its tables (it names the symbol), an alias that would clash with another name in the file, a namespace import used as a plain value, or an operator in a form that cannot be expressed. Aliased imports (map as rxMap) and namespace imports (rx.of, rx.operators.map, rx.Observable) are expanded, renaming every reference; a .pipe whose receiver's type cannot be established is wrapped in from(...) with a review notice. from, scheduled, generate (positional form included), bindCallback, fromFetch, ajax, webSocket and every publish-family form (selector forms become connect(...)) are translated. Subjects become subject() / behaviorSubject() / replaySubject(), subscribe(fn) becomes subscribe({ next: fn }), mergeMap(project, 3) becomes mergeMap(project, { concurrency: 3 }), lastValueFrom / firstValueFrom move to @craft-ts/core, scheduler arguments (asyncScheduler, asapScheduler, queueScheduler, animationFrameScheduler) are dropped with a review notice — there is one scheduler, the temporal runtime — x.pipe(…, publish()) becomes connectable(x.pipe(…)) (and publishReplay / publishBehavior / multicast get their connector), and the operators and creators with the same name (interval, timer, fromEvent, defer, throwError, concat, first, last, reduce, auditTime, expand, …) move over unchanged.

isObservable becomes isObservableLike, Observer becomes StreamObserver, and SubscriptionLike / Unsubscribable / Subscribable move to @craft-ts/core. The operator-function types (OperatorFunction, MonoTypeOperatorFunction, ObservableInput, …) block their file with the replacement to write by hand, since a craft operator also carries a yielded type. New imports take the place of the rxjs ones; the rest of the import block is left exactly as it was.

The deprecated operators with no equivalent (retryWhen, repeatWhen, mergeScan, switchScan, windowTime, windowToggle, windowWhen, bufferToggle, using, onErrorResumeNext, publishLast, timeoutWith, sequenceEqual, animationFrames, pluck, …MapTo) block their file with a diagnostic that names the replacement. exhaust becomes exhaustAll.

Semantic changes are reported as review notices on a migrated file: catchError(() => x$) becomes orElse(() => x$), which continues on a typed exception of a craft stream and does not catch defects — prefer catchTag so the handler is typed per exception; timeout raises a typed exception; retry retries typed exceptions only; zip buffers without bound.