-
public interface StreamSingleFlightProcessorA 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 Summary
Modifier and Type Method Description 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. abstract <T extends Any> Booleanhas(StreamTypedKey<T> key)Checks if there is an in-flight execution for key. abstract <T extends Any> Result<Unit>cancel(StreamTypedKey<T> key)Cancels the current in-flight execution for key, if any. abstract Result<Unit>clear(Boolean cancelRunning)Clears internal bookkeeping. abstract Result<Unit>stop()Stops all execution and shuts down the runner. -
-
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.
-
has
abstract <T extends Any> Boolean has(StreamTypedKey<T> key)
Checks if there is an in-flight execution for key.
- Parameters:
key- The key to check.
-
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.
-
-
-
-