dart-flow

KO | EN

gitlite — search

#python #java #python3 #arduino #golang #machine-learning #rust #html #flask #javascript #seismology #nodejs

★ 19 Open GitHub ↗

An Asynchronous data stream that emits events sequentially.

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# Flow API

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<U>](#map) Applies a transformation function to each element in the flow, resulting in a flow with elements of type `U`.
* [flatMap](#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](#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](#asStream) Converts this flow into a `Stream<T>`.
* [filter](#filter)  Filters elements emitted by the flow based on a provided predicate function.
* [cache](#cache) Creates a new Flow that applies a caching strategy using the provided ```CacheFlow``` and ```CacheStrategy``` objects.
* [catchError](#catchError)  Handles errors that occur within the flow.
* [onStart](#onstart) Executes an action before the flow starts collecting data.
* [onCompletion](#onCompletion) Executes an action upon flow completion (**needs improvement**).
* [retryWhen](#retryWhen) Implements retry logic based on a provided function to handle temporary errors.
* [retryWith](#retryWith) Implements retry logic based on a provided [RetryPolicy].
* [onEach](#onEach) Returns a flow that invokes the given [action] before each value of theupstream flow is emitted downstream.
* [onEmpty](#onEmpty) Creates a new flow that executes the provided action ([action]) only if the original flow emits no events (i.e., is empty).
* [distinctUntilChanged](#distinctUntilChanged)A 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](#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](#launchin) Starts collecting a flow, stopping automatically once a `FlowLifecycle` (e.g. a ViewModel) is disposed — no manual `StreamSubscription`/cancel bookkeeping required.

**Getting Started:**

1. Install the Flow package:
   ```bash
   pub add flow
   ```

2. Import the Flow library in your Dart code:
   ```dart
   import 'package:flow/flow.dart';
   ```

## Creating a flow

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.

```dart
flow<String>((collector) async {
  collector.emit('Flow API');
})
```
Create a Flow from an iterable collection of elements.
```dart
flowOf([1, 2, 3, 4])
  .map<String>((number) => (number * 2).toString())
  .catchError((error, collector) => print("Error: $error"))
  .collect(print); // Prints: 2, 4, 6, 8
```

## channelFlow

A 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.

```dart
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`.

## Flow Operators

## map<U>
Applies a transformation function to each element in the flow, resulting in a flow with elements of type `U`.

```dart
final flow = flowOf([1,2,3,4])
  .map((value) => value * 3)
    .collect(print);
 //Output 3,6,9,12
```

## flatMap
Applies 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.

```dart
flowOf([1, 2, 3])
    .flatMap((value) => flowOf([4, 5, 6]))
    .collect(print);
// Output:
// 4 -> 5 -> 6
```

## conflateMap
Like `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.

```dart
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.

## asStream
Converts this flow into a `Stream<T>`.

This allows you to use `Stream`-based operators and functionalities on your flow.

```dart
     flowOf([1, 2, 3, 4]).asStream()
 ```

## filter
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.

```dart
     flowOf([1, 2, 3, 4]).filter((value) => value % 2 == 0)
      .collect(print); // This will print only even numbers (2, 4)
  ```

## cache
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.

```dart
  // 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.

## retryWhen
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.

```dart
   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]

## retryWith
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.

```dart
     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.

### At the moment, dart-flow supports 7 different retryPolcies, which are

1. ExponentialBackOffRetryPolicy
  2. CircuitBreakerRetryPolicy
  3. FixedIntervalRetryPolicy
  4. DecorrelatedJitterRetryPolicy
  5. LinearBackoffRetryPolicy
  6. RandomisedBackoffRetryPolicy
  7. NoRetryPolicy

### ExponentialBackOffRetryPolicy

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](https://www.baeldung.com/resilience4j-backoff-jitter)

### CircuitBreakerRetryPolicy

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](https://dantt.medium.com/circuit-breaker-and-retry-64830e71d0f6)

```dart
     flow((collector) {
       collector.emit('A');
       throw Exception('Something went wrong');
     }).retryWith((cause) =>  RetryPolicy.circuitBreaker())
       .collect(print);
   ```

### FixedIntervalRetryPolicy

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.

```dart
     flow((collector) {
       collector.emit('A');
       throw Exception('Something went wrong');
     }).retryWith((cause) =>  RetryPolicy.fixedInterval())
       .collect(print);
   ```

### DecorrelatedJitterRetryPolicy

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](https://www.baeldung.com/resilience4j-backoff-jitter)

```dart
     flow((collector) {
       collector.emit('A');
       throw Exception('Something went wrong');
     }).retryWith((cause) =>  RetryPolicy.decorrelatedJitter())
       .collect(print);
   ```

### LinearBackoffRetryPolicy

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.

```dart
     flow((collector) {
       collector.emit('A');
       throw Exception('Something went wrong');
     }).retryWith((cause) =>  RetryPolicy.linearBackoff())
       .collect(print);
   ```

### RandomisedBackoffRetryPolicy

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.

```dart
     flow((collector) {
       collector.emit('A');
       throw Exception('Something went wrong');
     }).retryWith((cause) =>  RetryPolicy.randomisedBackoff())
       .collect(print);
   ```

### NoRetryPolicy

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.

```dart
     flow((collector) {
       collector.emit('A');
       throw Exception('Something went wrong');
     }).retryWith((cause) =>  RetryPolicy.noRetry())
       .collect(print);
   ```

### Implementing Custom Retry Policies

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.

```dart
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
  }
}
```

## onCompletion
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.

```dart
  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.

## onStart
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.

```dart
   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.

## onEach
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.

```dart
    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.

## onEmpty
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.

```dart
 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.

## distinctUntilChanged
A Function that returns a flow where all subsequent repetitions of the same value are filtered out.

```dart
   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,4
   ```

## Combination Operators

## merge
A 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:
 ```dart
 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 arrive
  ```

## combineLatest
A 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:
 ```dart
 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"
 ```

## race
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:
```dart
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 emit
```

## startWith
Creates a new flow that emits the specified value before emitting values from this flow.
Example:
```dart
final flow = flowOf([1, 2, 3]);
final prefixed = flow.startWith(0);
prefixed.collect(print); // prints 0, 1, 2, 3
```

## Lifecycle Integration

## launchIn
Starts 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.

```dart
someFlow
  .onEach((value) => print(value))
  .catchError((e, _) => print('error: $e'))
  .launchIn(viewModel); // viewModel implements FlowLifecycle
```

Cancellation 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.

```dart
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.

## FlowLifecycle
The minimal contract `launchIn` needs to know when to stop collecting on its own, without exposing a raw `StreamSubscription` to the caller.

```dart
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.

## CancellationSignal
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` — whether `cancel()` 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)` — registers `listener` to run when `cancel()` is called. If the signal is already cancelled, `listener` runs 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:**

##### TODO

**Contributing:**

We welcome contributions to the Flow API. Please refer to the CONTRIBUTING.md file for guidelines.

License

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.

dart-flow

Explore Similar Repositories

Was this content helpful?

Recent Feedback

Download README