Package 

Interface StreamSingleFlightProcessor

  • 
public interface StreamSingleFlightProcessor

    A concurrency construction that implements the single-flight pattern (à la Go's singleflight): concurrent calls with the same keyshare one in-flight execution and all await the same result.

    • One in-flight per key: While a call for key is running, additional run calls with the same key do not start a new execution; they join the existing one.

    • Result-based API:run never throws for work failures; it returns a Result. (The only exception is if the awaiting caller is cancelled—then a CancellationException is thrown to that caller, while the shared work continues for others.)

    • Cancellation:cancel cancels the shared in-flight execution for key (if any); all joiners then observe Result.failure(CancellationException). clear only clears bookkeeping and does not cancel.

    • Deduplicate idempotent reads (HTTP GETs, token fetches).

    • Prevent duplicate work due to UI recomposition.

    fun getUser(userId: String) = singleFlight.run("GET:/users/$userId") {
      // This block runs once per in-flight key; additional callers await the same result.
      // Cancellation of one awaiting caller does not cancel the shared execution.
      httpClient.get("/users/$userId").body()
}

    Doing a multiple calls to getUser(101) will only make one network call for that user.

    Implementations must be safe for concurrent use.

    • Method Detail

      • run

         abstract <T extends Any> Result<T> run(StreamTypedKey<T> key, SuspendFunction0<T> block)

        Runs block, ensuring that at most one execution for key is in flight. Concurrent callers with the same key await the same shared execution and receive its Result.

        Parameters:
        key - A stable, deterministic key that identifies the logical request (e.g., "GET:/users/42").
        block - The suspending work to execute once per in-flight key.

      • cancel

         abstract <T extends Any> Result<Unit> cancel(StreamTypedKey<T> key)

        Cancels the current in-flight execution for key, if any. Joiners will receive Result.failure(CancellationException). No-op if there is no in-flight execution.

      • clear

         abstract Result<Unit> clear(Boolean cancelRunning)

        Clears internal bookkeeping. Useful if another component owns lifecycle/cancellation and you want to drop references.

      • stop

         abstract Result<Unit> stop()

        Stops all execution and shuts down the runner. No further jobs are accepted. Stop is a destructive action.