The Flow API provides functionalities for building asynchronous data processing pipelines. It offers a concise and expressive way to chain operations on streams of data.
Key Features:
Flow Operators: Extension methods for the Flow<T> class, offering functionalities like:
- map Applies a transformation function to each element in the flow, resulting in a flow with elements of type
U. - flatMap Applies a transformation function to each element in the flow, potentially creating new flows. The resulting flows are then flattened into a single stream of values.
- conflateMap Like
map, but coalesces overlapping work — keeps at most one transform call in flight, processing only the newest pending value once the current call finishes. - asStream Converts this flow into a
Stream<T>. - filter Filters elements emitted by the flow based on a provided predicate function.
- cache Creates a new Flow that applies a caching strategy using the provided
CacheFlowandCacheStrategyobjects. - catchError Handles errors that occur within the flow.
- onStart Executes an action before the flow starts collecting data.
- onCompletion Executes an action upon flow completion (needs improvement).
- retryWhen Implements retry logic based on a provided function to handle temporary errors.
- retryWith Implements retry logic based on a provided [RetryPolicy].
- onEach Returns a flow that invokes the given [action] before each value of theupstream flow is emitted downstream.
- onEmpty Creates a new flow that executes the provided action ([action]) only if the original flow emits no events (i.e., is empty).
- distinctUntilChangedA Function that returns a flow where all subsequent repetitions of the same value are filtered out
Flow Builders: Functions that create a new Flow<T>:
- channelFlow Creates a flow that can launch concurrent work (e.g. a one-shot fetch) alongside a value it forwards indefinitely (e.g. a live stream), without either closing the other early.
Lifecycle Integration:
- launchIn Starts collecting a flow, stopping automatically once a
FlowLifecycle(e.g. a ViewModel) is disposed — no manualStreamSubscription/cancel bookkeeping required.
Getting Started:
-
Install the Flow package:
pub add flow
-
Import the Flow library in your Dart code:
import 'package:flow/flow.dart';
To create flows, use the flow builder APIs. The flow builder function creates a new flow where you can manually emit new values into the stream of data using the flow collector emit function.
flow<String>((collector) async {
collector.emit('Flow API');
})Create a Flow from an iterable collection of elements.
flowOf([1, 2, 3, 4])
.map<String>((number) => (number * 2).toString())
.catchError((error, collector) => print("Error: $error"))
.collect(print); // Prints: 2, 4, 6, 8A plain flow<T>((collector) async {...}) closes as soon as its callback's returned future completes — any fire-and-forget work still running is orphaned, not awaited. channelFlow instead hands the callback a ProducerScope, which can launch concurrent work without the flow closing early: it doesn't finish until both the callback returns and every launched action has completed. That makes it safe to concurrently launch a one-shot operation (e.g. a network fetch) alongside a line that awaits something that never completes on its own (e.g. forwarding a live database watch stream).
An uncaught error thrown by a launched action is routed to scope.addError — it doesn't terminate the scope or any other launched action.
Flow<DashboardData> getDashboardData(String userId) {
return channelFlow<DashboardData>((scope) async {
// Runs concurrently with the line below — an uncaught error here
// routes to scope.addError, it doesn't close the scope.
scope.launch(() => fetchAndSyncFreshData(userId));
// Never returns on its own — keeps the channelFlow open, forwarding
// every value the watch stream emits (including ones caused by the
// write above) for as long as it's subscribed.
await database.watchDashboardData(userId).collect(scope.emit);
});
}[scope] : A ProducerScope<T> responsible for emitting values of type T into the flow, either directly (scope.emit) or from concurrent work started via scope.launch.
Applies a transformation function to each element in the flow, resulting in a flow with elements of type U.
final flow = flowOf([1,2,3,4])
.map((value) => value * 3)
.collect(print);
//Output 3,6,9,12Applies a transformation function and flattens the resulting streams.
This function is similar to map but allows transforming each element in the flow into a new flow. The resulting flows are then flattened into a single stream of values.
flowOf([1, 2, 3])
.flatMap((value) => flowOf([4, 5, 6]))
.collect(print);
// Output:
// 4 -> 5 -> 6Like map, but coalesces overlapping work instead of running transform concurrently for every value.
Plain map has no backpressure: if the flow emits several values before an earlier transform call finishes, each one still starts its own concurrent call — results can complete (and so emit) out of order, and any side effect inside transform runs once per value regardless of how close together they arrived. conflateMap instead keeps at most one call to transform in flight: a value that arrives while a previous call is still running replaces whatever's pending rather than starting a new concurrent call, and once the in-flight call finishes, transform runs exactly once more against only the newest pending value — not once per value that arrived in between. Equivalent to Kotlin's flow.conflate().map { ... }: conflate the source, then process whatever gets through sequentially.
This is deliberately not mapLatest/switchMap: it never cancels an in-flight transform call. A value superseding the one currently running still waits for that call to finish and emit before its own (coalesced) turn starts — only not-yet-started work is ever skipped.
Useful when several independent upstream sources can all trigger the same expensive rebuild in quick succession (e.g. several database tables watched together, all written by the same transaction) and transform itself has a side effect that shouldn't repeat once per source — running it once per settled batch instead of once per individual trigger is the point.
tableChangeFlow.conflateMap((_) => rebuildExpensiveSnapshot())
.collect(print);[transform] : A function that takes a value of type T and returns a FutureOr<U>. Only the most recently arrived value is guaranteed to eventually get its own transform call once the previous call settles.
Converts this flow into a Stream<T>.
This allows you to use Stream-based operators and functionalities on your flow.
flowOf([1, 2, 3, 4]).asStream()Filters elements emitted by the flow based on a provided predicate function.
This function allows you to selectively emit elements from the flow. The provided action function takes a single argument, the current value (T) emitted by the flow.
It should return a FutureOr<bool>. If the action function returns true, the value is emitted by the resulting flow. Otherwise, the value is discarded.
flowOf([1, 2, 3, 4]).filter((value) => value % 2 == 0)
.collect(print); // This will print only even numbers (2, 4)Creates a new Flow that applies a caching strategy using the provided CacheFlow and CacheStrategy objects.
This function allows you to integrate caching logic into your data flows. It takes a CacheFlow object that defines the caching behavior (e.g., reading, writing from cache), a CacheStrategy object that determines the specific caching strategy to employ (e.g., FetchOrElseCache, CacheThenFetch), and a FlowCollector to emit data downstream.
The provided CacheStrategy handles the interaction between the source Flow (this in the context of the function) and the CacheFlow toimplement the desired caching behavior.
// Example CacheFlow implementation (simplified)
class InMemoryCache<T> implements CacheFlow<T> {
// ... cache implementation details
}
// Example CacheStrategy implementation (simplified)
class FetchOrElseCache<T> implements CacheStrategy<T> {
@override
FutureOr<void> handle(CacheFlow<T> cacheFlow, Flow<T> sourceFlow,
FlowCollector collector) async {
// ... implementation to fetch or read from cache
}
}
// Usage
flowOf([1, 2, 3])
.cache(InMemoryCache<int>(), FetchOrElseCache<int>())
.collect(print);[cacheFlow] : An object implementing the CacheFlow interface that provides caching functionalities (read, write, etc.)
[strategy] : An object implementing the CacheStrategy interface that defines the specific caching strategy to be used with the CacheFlow.
Returns: A new Flow that incorporates the caching logic defined by the provided CacheStrategy and CacheFlow objects.
Implements retry logic based on a provided function.
This function allows you to define a retry strategy for handling errors within the flow. The provided action function takes the exception and the current attempt number as arguments. It should return true if the flow should be retried and false otherwise.
flow((collector) {
collector.emit('A');
throw Exception('502');
}).retryWhen((cause, attempts) {
if (cause.toString().contains('502') && attempts < 2) {
return true;
}
return false;
}).collect(print);[action] : A function that takes an Exception and an int (the current attempt number) as arguments. It determines whether to retry the flow based on the value returned(true|false) by [action]
Implements retry logic based on a provided [RetryPolicy].
This approach offers more flexibility by allowing you to define a custom retry policy class that encapsulates various retry strategies. The provided action function takes the encountered exception as an argument and should return a concrete implementation of the RetryPolicy interface. This policy object then dictates the retry behavior based on factors like the number of attempts, elapsed time, or specific error types.
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.exponentialBackOff())
.collect(print);[action] : A function that takes an Exception as an argument. It should return a concrete implementation of the RetryPolicy interface, defining the retry strategy for the flow in case of errors.
- ExponentialBackOffRetryPolicy
- CircuitBreakerRetryPolicy
- FixedIntervalRetryPolicy
- DecorrelatedJitterRetryPolicy
- LinearBackoffRetryPolicy
- RandomisedBackoffRetryPolicy
- NoRetryPolicy
The ExponentialBackOff retry policy mitigates transient failures by strategically increasing the delay between retry attempts. This approach helps in managing system load and improving the chances of subsequent attempts succeeding. Read More
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.exponentialBackOff())
.collect(print);The CircuitBreakerRetryPolicy prevents flow from performing operations that are likely to fail. If failures reach a certain threshold, the circuit breaker trips, and further attempts are blocked for a configured time period. After the timeout expires, the circuit breaker allows a limited number of test requests to pass through. If these are successful, normal operation is resumed. Read More
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.circuitBreaker())
.collect(print);This policy retries tasks at fixed intervals, regardless of the number of attempts made. It is simple and predictable, making it suitable for situations where the expected time for the issue to be resolved is known.
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.fixedInterval())
.collect(print);The DecorrelatedJitterRetryPolicy adds a randomized delay between retries to prevent thundering herd problems, which can occur when many clients retry a failed request simultaneously. This approach combines both random and fixed delay strategies to provide a balance between retrying quickly and avoiding overwhelming the server or resource. Read More
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.decorrelatedJitter())
.collect(print);The LinearBackOff retry policy increases the delay between retry attempts by a fixed increment. Unlike exponential backoff, which doubles the delay with each attempt, linear backoff provides a constant increase in the retry interval. This approach is straightforward and predictable, making it suitable for scenarios where a steady progression in retry attempts is preferred.
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.linearBackoff())
.collect(print);The RandomizedBackOff retry policy introduces randomness into the delay between retry attempts. Instead of following a deterministic pattern like linear or exponential backoff, randomized backoff adds a random component to the retry intervals. This randomness helps prevent synchronization among multiple clients attempting retries simultaneously, reducing contention and the likelihood of overwhelming the target system.
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.randomisedBackoff())
.collect(print);As the name suggests, the NoRetryPolicy implements the retry interface but does not attempt retries. It is useful as a default policy or in situations where retries are not desired.
flow((collector) {
collector.emit('A');
throw Exception('Something went wrong');
}).retryWith((cause) => RetryPolicy.noRetry())
.collect(print);To create your custom retry policy, implement the RetryPolicy interface and override the retry method. Your implementation should define the logic to determine if a retry should occur based on the number of [attempts] already made and the nature of the failure.
class MyCustomRetryPolicy implements RetryPolicy {
@override
FutureOr<bool> retry(int attempt) {
// Implement your custom retry logic here
// Return true to retry, false to not retry
if (attempt < 3) {
return true; // Retry on specific error up to 3 times
}
return false; // Do not retry for other errors or after 3 attempts
}
}Executes an action upon flow completion (needs improvement).
This function allows you to perform actions or cleanup tasks after the flow has finished processing data. The provided action function receives an optional exception (null if no exception occurred) and the FlowCollector as arguments. Note: Currently, the handling of completion errors within the context of the flow needs improvement.
flowOf([1, 2, 3]).onCompletion((exception, collector){
//Perform action
});[action] : A function that takes an optional Exception and a FlowCollector<T> as arguments. It can be used for post-processing, cleanup, or handling any errors that might occur during completion.
Executes an action before the flow starts collecting data.
This function allows you to perform setup tasks or initializations before the actual flow processing begins. The provided action function receives the FlowCollector as an argument.
flow((collector) => collector.emit('World!'))
.onStart((collector) => collector.emit('Hello,'));
.collect(stdout.write)
// Outputs:
// Hello, World![action] : A function that takes a FlowCollector<T> as an argument. It can be used for pre-processing or any actions needed before collecting data in the flow.
Returns a flow that invokes the given [action] before each value of the upstream flow is emitted downstream.
This function allows you to perform actions on each individual value that flows through the pipeline, potentially performing side effects before the value is sent further downstream.
flowOf([1, 2, 3])
.onEach((value) => print('Emitting value: $value'))
.collect(print);
// Output:
// Emitting value: 1
// 1
// Emitting value: 2
// 2
// Emitting value: 3
// 3[action] : A function that takes a value of type T and potentially performs asynchronous operations. This function is called for each value emitted by the source Flow.
Creates a new flow that executes the provided action ([action]) only if the original flow emits no events (i.e., is empty).
This function is useful for scenarios where you want to perform specific logic when a flow is empty. For example, you might want to emit a default value, fetch data from another source, or trigger some side effect when no data is available in the original flow.
Flow<int> numbers = Flow.from([1, 2, 3]);
// This action will NOT be executed because the original flow is not empty
Flow<int> withEmptyHandling = numbers
.onEmpty((collector) => collector.emit(0));
withEmptyHandling.collect(print); // Output: 1, 2, 3
Flow<String> emptyStringFlow = Flow.empty();
// This action WILL be executed because the original flow is empty
Flow<String> withEmptyAction = emptyStringFlow
.onEmpty((collector) => collector.emit("No data available"));
withEmptyAction.collect(print); // Output: No data available[action] : A function that accepts a FlowCollector<T> as its parameter. The provided action will be executed only if the original flow doesn't emit any values. Otherwise, the original flow's events are simply forwarded downstream without any modification.
A Function that returns a flow where all subsequent repetitions of the same value are filtered out.
flow<DummyClass>((collector) {
collector.emit(DummyClass(foo: 1));
collector.emit(DummyClass(foo: 2));
collector.emit(DummyClass(foo: 3));
collector.emit(DummyClass(foo: 2));
collector.emit(DummyClass(foo: 4));
})
.distinctUntilChanged(
keySelector: (value) => value.foo,
areEquivalent: (previousKey, nextKey) => (previousKey ?? 0) > nextKey!,
)
.collect((value) => print(value.foo));
// Output: 1,2,3,4A Flow that merges multiple source Flows into a single Flow, collecting values from each Flow sequentially.
This Flow collects values from each source Flow one at a time in the order they were provided. If any source Flow emits an unhandled error, collection stops and the error is propagated.
Example:
final flow1 = flowOf([1, 3]);
final flow2 = flowOf([2, 4]);
final merged = Flow.merge([flow1, flow2]);
merged.collect(print); // prints 1, 2, 3, 4
// Marge values from both flows using an extension method
final merged2 = flow1.mergeWith(flow2);
merged.collect(print); // prints all values from both flows as they arriveA Flow that combines the latest values from multiple source Flows into a single value.
This Flow waits for all source Flows to emit at least one value before emitting any combined values. When any source Flow emits a new value, it combines the latest values from all Flows using the provided combiner function and emits the result.
If any source Flow emits an unhandled error, collection stops and the error is propagated.
Example:
final flow1 = flowOf([1, 2, 3]);
final flow2 = flowOf(['a', 'b', 'c']);
final combined = Flow.combineLatest([flow1, flow2], (values) => '${values[0]}-${values[1]}');
combined.collect(print); // prints "1-a", "2-a", "2-b", "3-b", "3-c"
// Combine the latest values from both flows using an extension method
final combined = flow1.combineLatestWith(flow2, (a, b) => '$a-$b');
combined.collect(print); // prints: "1-a", "2-a", "2-b", "3-b", "3-c"Given two or more source Flows, emits all values from only the first Flow that emits a value. After the first Flow emits, all other Flows are ignored.
If the provided list of Flows is empty, the resulting Flow completes immediately without emitting any values.
If any source Flow emits an unhandled error, collection stops and the error is propagated.
Example:
final flow1 = flow<String>((collector) async {
await Future.delayed(const Duration(milliseconds: 200));
collector.emit("A");
collector.emit("C");
});
final flow2 = flow<String>((collector) async {
collector.emit("C");
});
final race = Flow.race([flow1, flow2]);
race.collect(print); // prints "B"
// Race using extension method
final raced = flow1.raceWith(flow2);
raced.collect(print); // prints values from the first flow to emitCreates a new flow that emits the specified value before emitting values from this flow. Example:
final flow = flowOf([1, 2, 3]);
final prefixed = flow.startWith(0);
prefixed.collect(print); // prints 0, 1, 2, 3Starts collecting this flow, stopping automatically once a lifecycle is disposed — no StreamSubscription/cancel bookkeeping required at the call site. Mirrors Kotlin's Flow.launchIn(scope).
Configure the pipeline with onEach/catchError first, then call launchIn last — it takes no callback of its own, whatever onEach attached upstream is what actually runs per value.
someFlow
.onEach((value) => print(value))
.catchError((e, _) => print('error: $e'))
.launchIn(viewModel); // viewModel implements FlowLifecycleCancellation cascades cooperatively: any nested flow collected while this call is active (e.g. an upstream flow this one wraps) tears itself down too, once the lifecycle ends.
launchIn returns the CancellationSignal backing this collection — mirrors Kotlin's launchIn returning a Job. Call .cancel() on it to stop this specific collection early, without disposing the lifecycle — useful when the same instance needs to restart collection (e.g. for a different argument) rather than tearing the whole thing down. The lifecycle disposing cancels it too, same as before; either path converges on the same signal, so there's one teardown mechanism, not two racing ones.
final signal = someFlow.onEach(handle).launchIn(viewModel);
// ...later, without disposing viewModel:
signal.cancel();[lifecycle] : A FlowLifecycle — anything with a well-defined "end" (a ViewModel's dispose, a widget's State.dispose, a request scope, etc.). It needs an isDisposed getter and an addCleanup(void Function()) method; most ViewModel base classes already have an equivalent shape, so satisfying this is typically a zero-new-code addition.
The minimal contract launchIn needs to know when to stop collecting on its own, without exposing a raw StreamSubscription to the caller.
abstract class FlowLifecycle {
bool get isDisposed;
void addCleanup(void Function() action);
}Implement this on anything that has a well-defined "end". The two members are named to match the shape most ViewModel base classes already have, so satisfying this interface is typically a zero-new-code addition on the implementing side.
A cancel-once, notify-listeners primitive returned by launchIn and threaded through the current Zone so nested flow operations can cascade their own teardown without the code that created them having to check anything explicitly.
isCancelled— whethercancel()has already been called.cancel()— cancels the signal, synchronously notifying every registered listener. Idempotent — calling it more than once has no further effect.onCancel(listener)— registerslistenerto run whencancel()is called. If the signal is already cancelled,listenerruns immediately.
You won't usually construct one directly — launchIn returns it, and Flow internals read it via currentCancellationSignal() to cascade teardown through nested Flow operations automatically.
Benefits:
- Asynchronous Processing: Efficiently handles streams of data with asynchronous operations.
- Concise Syntax: Provides a readable and easy-to-use API for building data pipelines.
- Composable Operators: Allows chaining various operations together for complex data processing workflows.
Further Documentation:
Contributing:
We welcome contributions to the Flow API. Please refer to the CONTRIBUTING.md file for guidelines.
License
Copyright 2024 Moniepoint, Inc.
Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.