Displace transfer: for each input value, creates a new inner
async-pushable + subscribable transfer via a factory function,
subscribes to it, pushes the value into it via asyncPush(), and
forwards the inner's emissions to outer subscribers.
On each new push(), the previous inner subscription is unsubscribed
and the previous inner transfer is destroyed — the new inner
displaces the previous one. Only the latest inner transfer's
emissions reach the outer subscribers.
The factory receives no arguments — the input value is delivered to
the inner transfer via asyncPush(data), not passed to the factory.
This keeps the factory purely declarative: it creates a transfer,
DisplaceTransfer handles data delivery.
The outer push() is synchronous: it creates the inner, disposes the
previous one, subscribes, and calls inner.asyncPush(data) fire-and-
forget. The async work happens inside the inner transfer — its
results arrive via subscription callbacks.
push(data) — calls factory() to create a new inner transfer,
then disposes the previous inner (unsubscribe + destroy),
then subscribes to the new inner (forwarding emissions to outer subscribers),
then calls inner.asyncPush(data) to deliver the value (fire-and-forget)
subscribe(handler) — subscribes to the outer output
destroy() — disposes the current inner transfer, unsubscribes outer subscribers
Error handling:
If factory() throws an exception, onError is called.
With onError provided, the exception is suppressed (previous inner remains active).
Without onError, the exception is rethrown.
If inner.asyncPush(data) rejects, the rejection is unhandled (fire-and-forget).
Provide onError on the inner transfer to suppress internal errors.
Configuration (DisplaceTransferConfig):
factory: () => Transfer<TInput, TOutput, [AsyncPushable, Subscribable]>
— creates inner transfer per input (no arguments; data is pushed via asyncPush)
onError?: ErrorHandler — factory error handler
onDisplace?: (displaced: Transfer<TInput, TOutput, [AsyncPushable, Subscribable]>) => void
— called with the previous inner transfer before it is unsubscribed and destroyed.
Use for cleanup that must happen before destruction (e.g., aborting an in-flight request,
closing a WebSocket, cancelling a timer). If the callback throws, the exception is
rethrown (the inner is still destroyed). Not called on destroy() — only on displacement
by a new push(). Not called for the first push() (no previous inner exists).
Use cases:
switchMap semantics: displace previous inner stream on new input
Search-as-you-type: debounce → displace(factory) → latest result wins
Per-value async operations (fetch, readFile) where only the latest result matters
Per-value WebSocket/stream subscriptions with automatic cleanup
Custom cancellation logic via onDisplace (abort, close, cancel) before inner is destroyed
Displace transfer: for each input value, creates a new inner async-pushable + subscribable transfer via a factory function, subscribes to it, pushes the value into it via asyncPush(), and forwards the inner's emissions to outer subscribers.
On each new push(), the previous inner subscription is unsubscribed and the previous inner transfer is destroyed — the new inner displaces the previous one. Only the latest inner transfer's emissions reach the outer subscribers.
The factory receives no arguments — the input value is delivered to the inner transfer via asyncPush(data), not passed to the factory. This keeps the factory purely declarative: it creates a transfer, DisplaceTransfer handles data delivery.
The outer push() is synchronous: it creates the inner, disposes the previous one, subscribes, and calls inner.asyncPush(data) fire-and- forget. The async work happens inside the inner transfer — its results arrive via subscription callbacks.
Capabilities: isInput, isOutput, isPushable, isSubscribable
Mechanics:
Error handling:
Configuration (DisplaceTransferConfig):
Use cases: