> For the complete documentation index, see [llms.txt](https://wong-coupon.gitbook.io/flutter/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://wong-coupon.gitbook.io/flutter/my-flutter/architecture-state/redux-epics-rxdart.md).

# Redux Epics and RxDart

Choose the correct concurrency strategy in Redux Epics to prevent duplicate requests and stale results from overwriting state

## Result

After defining a concurrency policy for each Epic, I no longer use one `asyncMap` for every request. Search uses `debounceTime` and `switchMap` so only the latest result can update state, submit uses `exhaustMap` to drop double taps, and tasks that must preserve order continue to use `asyncMap`.

With the same three input actions, every operator produces a different result:

```
Input:       A---B---C

asyncMap:    A-----------A'---B-----------B'---C-----------C'
             Runs sequentially; no action is dropped.

switchMap:   A---x---B---x---C-----------C'
             Only the latest inner stream remains subscribed.

exhaustMap:  A-----------A'
                 B and C are dropped while A is running.

flatMap:     A-----------A'
                 B-----------B'
                     C-----------C'
             Runs in parallel; output follows completion time.
```

The `x` symbol means the previous subscription was canceled. It does not necessarily mean the underlying HTTP request was aborted.

This article covers shared Dart Streams behavior. It has no Android- or iOS-specific implementation.

## Problem

When I started using Redux Epics, most code only filtered an action, called an API with `asyncMap`, and returned a success or failure action. This was easy to read and worked when actions arrived infrequently.

The problem appeared when the same kind of action was dispatched repeatedly:

* A search field dispatched an action after every character.
* A user tapped submit twice before the first request completed.
* A new refresh arrived before the previous refresh finished.
* Several independent items needed preloading without unlimited concurrency.
* A request error was not converted into a failure action, so loading state never closed.

If every case uses `asyncMap`, requests are queued. An old query can block a new query, while a duplicated submit still runs after the first request completes.

The important question is no longer “Where should I call the API?” but “Which action is allowed to win?”

| Situation                          | Expected behavior                                  | Strategy         |
| ---------------------------------- | -------------------------------------------------- | ---------------- |
| Updates must preserve order        | Wait for the current task before starting the next | Queue/sequential |
| Search should use the latest query | Discard the previous request result                | Latest wins      |
| A submit must not be duplicated    | Drop new actions while the first request runs      | First wins       |
| Several tasks are independent      | Run with bounded parallelism                       | Parallel         |

## Solution

### Treat an Epic as a stream policy

An Epic receives an action stream and returns another action stream:

```
Widget or service
       │ dispatch RequestAction
       ▼
Redux middleware chain
       │
       ├──► Reducer receives RequestAction
       │
       └──► Epic receives RequestAction
                 │
                 ▼
           Repository/API
                 │
                 ▼
       SuccessAction or FailureAction
                 │ dispatched again
                 └──────────────► Redux middleware chain
```

`redux_epics` calls `next(action)` before adding the action to the Epic stream. The original action still reaches the reducer; an Epic does not “swallow” it. Actions emitted by an Epic are dispatched again and pass through middleware and reducers like normal actions.

This article assumes the store is already split by feature and actions have clear types. If those boundaries do not exist yet, read the Redux store organization article first; this article does not repeat `AppState`, reducers, selectors, or `StoreConnector`.

{% content-ref url="/pages/flT55X8RgxKD0XpLXtUC" %}
[Redux/Flutter Redux at Scale](/flutter/my-flutter/architecture-state/redux-flutter-redux-large-app.md)
{% endcontent-ref %}

This direction of the link ensures that every Epic belongs to a feature. The reverse link from the Redux store article points here when a simple Future middleware can no longer describe debouncing, cancellation, or concurrency.

### Preparation

Add the packages:

```yaml
dependencies:
  redux: ^5.0.0
  redux_epics: ^0.15.2
  rxdart: ^0.28.0
```

Import them:

```dart
import 'package:redux_epics/redux_epics.dart';
import 'package:rxdart/rxdart.dart';
```

An Epic has this contract:

```dart
typedef Epic<State> = Stream<dynamic> Function(
  Stream<dynamic> actions,
  EpicStore<State> store,
);
```

The output must be an action. If an Epic emits the same action it filters for, that action is dispatched again and can create an infinite loop.

### Filter actions early

This example uses typed actions:

```dart
sealed class SearchAction {
  const SearchAction();
}

final class SearchRequested extends SearchAction {
  const SearchRequested(this.query);

  final String query;
}

final class SearchSucceeded extends SearchAction {
  const SearchSucceeded(this.query, this.items);

  final String query;
  final List<String> items;
}

final class SearchFailed extends SearchAction {
  const SearchFailed(this.query, this.error, this.stackTrace);

  final String query;
  final Object error;
  final StackTrace stackTrace;
}
```

Filter the action before reading its payload:

```dart
Stream<SearchRequested> selectSearchRequests(Stream<dynamic> actions) {
  return actions.whereType<SearchRequested>();
}
```

Data that determines the request should live in the action payload. If an event is queued, `store.state` may have changed by the time that event is processed.

### Use `asyncMap` when order matters

Dart's `Stream.asyncMap` waits for the current Future to complete before processing the next event.

```dart
Epic<AppState> createPreferenceEpic(
  PreferenceRepository repository,
) {
  return (actions, store) {
    return actions.whereType<PreferenceChanged>().asyncMap((action) async {
      try {
        await repository.save(action.value);
        return PreferenceSaved(action.value);
      } catch (error, stackTrace) {
        return PreferenceSaveFailed(error, stackTrace);
      }
    });
  };
}
```

I use `asyncMap` when every event must be processed and the order sent to the repository must match the action order. A slow request blocks the following event, so this is not the default choice for search-as-you-type.

### Use `debounceTime` and `switchMap` when the latest result must win

The repository is injected through an Epic factory so tests can replace it with a fake:

```dart
abstract interface class SearchRepository {
  Future<List<String>> search(String query);
}

Epic<AppState> createSearchEpic(SearchRepository repository) {
  return (actions, store) {
    return actions
        .whereType<SearchRequested>()
        .map((action) => action.query.trim())
        .where((query) => query.length >= 2)
        .distinct()
        .debounceTime(const Duration(milliseconds: 300))
        .switchMap(
          (query) => Stream.fromFuture(repository.search(query))
              .map<dynamic>((items) => SearchSucceeded(query, items))
              .onErrorReturnWith(
                (error, stackTrace) =>
                    SearchFailed(query, error, stackTrace),
              ),
        );
  };
}
```

In this pipeline:

* `distinct()` removes consecutive identical queries.
* `debounceTime` emits only after the user pauses typing.
* `switchMap` stops listening to the previous inner stream when a new query appears.
* Success and failure actions carry the query so the reducer can confirm that the result is still relevant.
* `onErrorReturnWith` converts a stream error into a failure action.

The previous result no longer reaches the Redux pipeline. However, if `repository.search` only returns a normal Future, the old network request may still continue underneath.

### Distinguish unsubscribing from canceling an HTTP request

To abort the actual request, the HTTP client must provide a cancellation API. This example uses Dio's `CancelToken`:

```dart
import 'package:dio/dio.dart';

abstract interface class CancellableSearchRepository {
  Future<List<String>> search(
    String query, {
    CancelToken? cancelToken,
  });
}

Stream<dynamic> searchRequest(
  CancellableSearchRepository repository,
  String query,
) {
  final cancelToken = CancelToken();

  return Stream.fromFuture(
    repository.search(query, cancelToken: cancelToken),
  )
      .map<dynamic>((items) => SearchSucceeded(query, items))
      .onErrorReturnWith(
        (error, stackTrace) => SearchFailed(query, error, stackTrace),
      )
      .doOnCancel(() {
        if (!cancelToken.isCancelled) {
          cancelToken.cancel('A newer search replaced this request');
        }
      });
}
```

Then use:

```dart
.switchMap((query) => searchRequest(repository, query))
```

`doOnCancel` connects inner-subscription cancellation to `CancelToken.cancel()`. The repository abstraction can still hide Dio from the rest of the application.

Cancellation caused by a new query is not an error that should be shown to the user. If the HTTP client emits an exception when canceled, classify it before creating `SearchFailed`.

### Use `exhaustMap` to drop double submits

```dart
Epic<AppState> createSaveProfileEpic(
  ProfileRepository repository,
) {
  return (actions, store) {
    return actions.whereType<ProfileSaveRequested>().exhaustMap(
      (action) => Stream.fromFuture(repository.save(action.profile))
          .map<dynamic>((profile) => ProfileSaveSucceeded(profile))
          .onErrorReturnWith(
            (error, stackTrace) =>
                ProfileSaveFailed(error, stackTrace),
          ),
    );
  };
}
```

While the first request is running, later submit actions are dropped. This works when a double tap must not create a second request and the first request is allowed to complete.

Do not use `exhaustMap` when new input must win. If a user changes the content while a save is running, the new action can be dropped.

### Use `flatMap` for independent tasks

```dart
Epic<AppState> createPreloadEpic(ItemRepository repository) {
  return (actions, store) {
    return actions.whereType<ItemPreloadRequested>().flatMap(
      (action) => Stream.fromFuture(repository.load(action.id))
          .map<dynamic>((item) => ItemPreloadSucceeded(action.id, item))
          .onErrorReturnWith(
            (error, stackTrace) =>
                ItemPreloadFailed(action.id, error, stackTrace),
          ),
      maxConcurrent: 4,
    );
  };
}
```

`flatMap` fits when each request has its own identity and results do not overwrite the same state slot. I always set `maxConcurrent` instead of letting the number of requests grow without a limit.

### Emit multiple actions through the output stream

One request may need to emit several actions in order:

```dart
Stream<dynamic> refreshCompletedFlow(RefreshSucceeded action) async* {
  yield action;
  yield const SummaryReloadRequested();
  yield const BadgeCountReloadRequested();
}
```

You can use `asyncExpand`, `async*`, or concatenation when one input creates several outputs. I avoid scattered `store.dispatch` calls inside an Epic because the output stream can describe the same behavior while preserving the complete action log.

### Keep errors inside the inner stream

If a Future fails without being mapped to an action, the reducer does not receive a failure action to close the loading state:

```dart
return actions.whereType<SearchRequested>().switchMap(
  (action) => Stream.fromFuture(repository.search(action.query)),
);
```

Place error handling inside the inner stream:

```dart
return actions.whereType<SearchRequested>().switchMap(
  (action) => Stream.fromFuture(repository.search(action.query))
      .map<dynamic>((items) => SearchSucceeded(action.query, items))
      .onErrorReturnWith(
        (error, stackTrace) =>
            SearchFailed(action.query, error, stackTrace),
      ),
);
```

Every request ends with a success or failure action, while the outer action stream continues receiving new requests.

### Bound retries

I retry only when the request is idempotent or has an idempotency key, the error is temporary, and the retry count is bounded. Timeouts, brief disconnections, or an unavailable server may qualify; authentication errors, invalid payloads, cancellations, and business errors do not.

RxDart provides `RetryWhenStream` to recreate a stream when a notifier emits an event. In production code, a retry helper should include at least:

* `maxAttempts`.
* `shouldRetry(error)`.
* Delay/backoff, optionally with jitter.
* A way to stop on logout, dispose, or a cancellation action.

Do not retry forever, and do not retry a side-effecting operation unless the server provides idempotency protection.

### Compose Epics by feature

Each feature exports its own Epic list:

```dart
List<Epic<AppState>> createCatalogEpics(
  SearchRepository searchRepository,
  ItemRepository itemRepository,
) {
  return [
    createSearchEpic(searchRepository),
    createPreloadEpic(itemRepository),
  ];
}
```

The root only combines those lists:

```dart
final appEpics = combineEpics<AppState>([
  ...createCatalogEpics(searchRepository, itemRepository),
  ...createProfileEpics(profileRepository),
]);
```

`combineEpics` merges the output of every Epic into one stream. Middleware placed before `EpicMiddleware` can block an action before it reaches the Epic stream, so middleware order is part of the architecture and must be tested.

### Test concurrency instead of only testing final state

A `switchMap` test must prove that the previous inner stream can no longer emit output:

```dart
test('switchMap only emits the latest result', () async {
  final queries = StreamController<String>();
  final first = StreamController<String>();
  final second = StreamController<String>();
  final values = <String>[];

  final subscription = queries.stream
      .switchMap((query) => query == 'fl' ? first.stream : second.stream)
      .listen(values.add);

  queries.add('fl');
  await pumpEventQueue();

  queries.add('flutter');
  await pumpEventQueue();

  first.add('old result');
  second.add('latest result');
  await pumpEventQueue();

  expect(values, ['latest result']);

  await subscription.cancel();
  await queries.close();
  await first.close();
  await second.close();
});
```

For a complete Epic, I also verify that:

* Search A completes after search B but does not emit a stale success action.
* A's cancellation token is called when B appears.
* One request failure does not prevent the Epic from handling the next request.
* `exhaustMap` calls the repository only once when submit is dispatched twice.
* `asyncMap` does not start B before A completes.
* Retry stops at `maxAttempts` and does not retry cancellations or business errors.

I use `Completer`, a fake repository, fake time, or stream matchers to control the timeline. A fixed `Future.delayed` followed by reading the final state does not prove which action was dropped or which request ran concurrently.

### Common errors and trade-offs

#### Using `asyncMap` for search

**Symptom:** an old query blocks a new query, and the UI displays data that is no longer relevant.

**Fix:** use `debounceTime` with `switchMap`, and include the query or request ID in the success action.

#### Treating `switchMap` as HTTP cancellation

**Symptom:** the UI does not receive the previous result, but the server still processes the request and the device still uses network resources.

**Fix:** connect `doOnCancel` to the HTTP client's cancellation token when the network request must actually be aborted.

#### Using `flatMap` for submit

**Symptom:** a double tap creates multiple concurrent requests.

**Fix:** use `exhaustMap`, disable the UI, or add an idempotency key according to the constraint. Do not rely only on the button's disabled state.

#### Catching errors outside the entire Epic

**Symptom:** a failed request does not create a failure action or affects later actions in the stream.

**Fix:** map errors inside each request's inner stream.

#### Reading `store.state` too late

**Symptom:** an action is queued but reads the state belonging to a newer action when it finally runs.

**Fix:** put request-defining data in the action; read the latest state only when that behavior is intentional.

RxDart makes a concurrency policy compact and visible in the pipeline, but the wrong operator can create a bug that is harder to notice than a sequence of `await` calls. A simple Future without debouncing, cancellation, or stream coordination can still use normal middleware.

When a new Epic replaces a Fish Redux `Effect`, its concurrency operator is only one part of the migration. The related article explains how I preserve route contracts, prevent an Effect and Epic from handling the same intent, and remove legacy code one vertical slice at a time.

{% content-ref url="/pages/1Z2MpTyQSwl5Y7CFKgbq" %}
[Incremental Fish Redux Migration](/flutter/my-flutter/architecture-state/incremental-fish-redux-migration.md)
{% endcontent-ref %}

### Verified versions

* Flutter: 3.41.2.
* Dart: 3.11 to before 4.0.
* `redux`: 5.0.0.
* `redux_epics`: 0.15.2.
* `rxdart`: 0.28.0.
* Platform: Shared.
* Verified on: 2026-08-23.

The reference app currently locks `redux_epics` 0.15.1 and `rxdart` 0.26.0. The examples use the current API; the app's older versions are not recommendations for a new project.

### References

* [redux\_epics — Dart API documentation](https://pub.dev/documentation/redux_epics/latest/redux_epics/)
* [redux\_epics — Epic](https://pub.dev/documentation/redux_epics/latest/redux_epics/Epic.html)
* [redux\_epics — EpicMiddleware](https://pub.dev/documentation/redux_epics/latest/redux_epics/EpicMiddleware-class.html)
* [redux\_epics — combineEpics](https://pub.dev/documentation/redux_epics/latest/redux_epics/combineEpics.html)
* [Dart Stream.asyncMap](https://api.dart.dev/dart-async/Stream/asyncMap.html)
* [RxDart — debounceTime](https://pub.dev/documentation/rxdart/latest/rx/DebounceExtensions.html)
* [RxDart — switchMap](https://pub.dev/documentation/rxdart/latest/rx/SwitchMapExtension.html)
* [RxDart — RetryWhenStream](https://pub.dev/documentation/rxdart/latest/rx/RetryWhenStream-class.html)
* [Dio — CancelToken](https://pub.dev/documentation/dio/latest/dio/CancelToken-class.html)

## Conclusion

Redux Epics are useful when a side effect is more than waiting for one Future and must define how several actions compete. `asyncMap`, `switchMap`, `exhaustMap`, and `flatMap` represent queue, latest wins, first wins, and parallel behavior respectively.

When the operator matches the intended behavior, errors become actions, and cancellation is connected to the HTTP client when necessary, the Redux flow remains traceable even when requests arrive continuously. If a side effect has no complex concurrency, a simple Future middleware remains easier to read and maintain.

[Buy Me a Coffee](https://buymeacoffee.com/ducmng12g) | [Support Me on Ko-fi](https://ko-fi.com/I2I81AEJG8)