Search documentation

Search by topic, API name, or import path.

GitHub ↗

variable / @typed/fx

Push.exhaustLatestMap

Runs one inner Fx at a time and retains only the latest value received while busy.

A value while idle starts immediately. While its inner runs, newer outer values replace a single pending slot. f(value) is evaluated and an inner Fx is constructed before every replacement; a superseded pending Fx is never run. After completion, only the latest pending Fx starts. Accepted inners preserve their own order and all values. The input Sink is unchanged.

Package version
2.0.0-beta.7
Category
Concurrent output work
Since
1.0.0
Member of
Push

Import

import { Push } from "@typed/fx";

Access Push.exhaustLatestMap through the imported Push export. Its declaration below describes the member.

This public exposure is a re-export. Its import path is supported; declaration documentation is shared with the other public exposures below.

Signatures

export declare const exhaustLatestMap: {
    <B, C, E3, R3>(f: (b: B) => Fx.Fx<C, E3, R3>): <A, E, R, E2, R2>(push: Push<A, E, R, B, E2, R2>) => Push<A, E, R, C, E2 | E3, R2 | R3 | Scope.Scope>;
    <A, E, R, B, E2, R2, C, E3, R3>(push: Push<A, E, R, B, E2, R2>, f: (b: B) => Fx.Fx<C, E3, R3>): Push<A, E, R, C, E2 | E3, R2 | R3 | Scope.Scope>;
};

Why

This is useful when work must not overlap but the newest requested state must eventually be processed.

Ownership and lifetime

The output Scope owns one active inner fiber and an in-memory latest slot. It joins the final inner before completion and interrupts it with the subscription. Inner failures and services join the output channels.

Examples

import { Effect } from "effect"
import * as Fx from "@typed/fx/Fx"
import * as Push from "@typed/fx/Push"
import * as Sink from "@typed/fx/Sink"

const constructed: Array<number> = []
const started: Array<number> = []
const finalized: Array<number> = []
const push = Push.make(
  Sink.make(() => Effect.void, (_: string) => Effect.void),
  Fx.fromIterable([1, 2, 3])
)
const latest = Push.exhaustLatestMap(push, (id) => {
  constructed.push(id)
  return Fx.make<number>((sink) =>
    Effect.gen(function* () {
      yield* Effect.sync(() => started.push(id))
      yield* Effect.sleep("5 millis")
      yield* sink.onSuccess(id * 10 + 1)
      yield* Effect.sleep("5 millis")
      yield* sink.onSuccess(id * 10 + 2)
    }).pipe(Effect.ensuring(Effect.sync(() => finalized.push(id))))
  )
})
const program = Fx.collectAll(latest).pipe(
  Effect.map((values) => ({ values, constructed, started, finalized })),
  Effect.scoped
)
// => values [11, 12, 31, 32]; constructed [1, 2, 3]; started/finalized [1, 3].

Other public imports

These import paths expose the same declaration. Each page retains its own public name and signature.

Source