With package:bloc, when should a Bloc handler follow a repository stream with emit.forEach or emit.onEach, and what must it do with the returned Future?
answer
- subscribe from inside the handler
- await or return the Future
- forEach maps each value to State
- onEach runs a void callback
- ends with stream or cancellation
basics
~20 sUse emit.forEach when each stream value should become a state, and emit.onEach when each value needs custom handling; both subscribe for the handler's lifetime and must be awaited or returned, or bloc reports pending subscriptions.
solid answer
~50 s`emit.forEach(stream, onData: (data) => nextState)` subscribes and emits whatever `onData` returns; `emit.onEach(stream, onData: (data) {...})` only calls `onData`, leaving emitting to you. Both return a `Future<void>` that completes when the stream ends or the handler is cancelled — by `close()` or a transformer such as `restartable()` — and the subscription is cancelled for you, so there is no `StreamSubscription` field to manage. The handler must `await` or `return` that Future. Otherwise the handler completes immediately: in debug builds an assertion reports that it left pending subscriptions behind; in release builds completing the run simply cancels the subscription. Without `onError`, a stream error is thrown from the Future and ends the subscription; with `onError`, it is handled and listening continues. The alternative is a manual subscription that adds a private event, which can be paused and given its own transformer.
code
dart · 31 linesimport 'package:bloc/bloc.dart';
import 'package:bloc_concurrency/bloc_concurrency.dart';
sealed class RecentCitiesEvent {}
final class RecentCitiesSubscriptionRequested extends RecentCitiesEvent {}
class RecentCitiesState {
const RecentCitiesState({this.cities = const [], this.failed = false});
final List<String> cities;
final bool failed;
}
abstract interface class CityHistoryRepository {
Stream<List<String>> recentSearches();
}
class RecentCitiesBloc extends Bloc<RecentCitiesEvent, RecentCitiesState> {
RecentCitiesBloc(this._history) : super(const RecentCitiesState()) {
on<RecentCitiesSubscriptionRequested>(
(event, emit) => emit.forEach<List<String>>(
_history.recentSearches(),
onData: (cities) => RecentCitiesState(cities: cities),
onError: (_, __) => const RecentCitiesState(failed: true),
),
transformer: restartable(),
);
}
final CityHistoryRepository _history;
}go deeper
Recall that a handler can follow a stream with emit.forEach, and that the call must be awaited or returned.
Explain forEach versus onEach, when the returned Future completes, and what onError changes.
Show the failure modes: the unawaited forEach assertion, duplicate subscriptions without restartable, and when a manual subscription with a private event is the better design.
Decide a codebase convention for reacting to repository streams so subscriptions are owned, cancellable and observable the same way everywhere.
## The situation In a travel app, a `RecentCitiesBloc` should show the user's recently searched cities, and the repository exposes them as a `Stream<List<String>>` that updates whenever a search completes. The stimulus is the stream, not a tap. **bloc** 9.2 offers two ways to wire that up, both described in its FAQ. ## Option 1 — emit.forEach or emit.onEach inside a handler ```dart RecentCitiesBloc(this._cities) : super(const RecentCitiesState()) { on<RecentCitiesSubscriptionRequested>( _onSubscriptionRequested, transformer: restartable(), ); } Future<void> _onSubscriptionRequested( RecentCitiesSubscriptionRequested event, Emitter<RecentCitiesState> emit, ) { return emit.forEach<List<String>>( _cities.recentSearches(), onData: (cities) => RecentCitiesState(cities: cities), onError: (_, __) => const RecentCitiesState(failed: true), ); } ``` How the two methods differ: | | `emit.forEach` | `emit.onEach` | |---|---|---| | `onData` returns | a `State`, which is emitted | nothing (`void`) | | `onError` returns | a `State`, which is emitted | nothing (`void`) | | Use when | each value maps to one state | you filter, combine or emit conditionally | Shared behaviour: - Both return a `Future<void>` that completes when the stream **ends** or the handler is **cancelled**, whichever comes first. - Cancellation happens on `close()` and when a transformer such as `restartable()` replaces the run; the subscription is cancelled automatically. - Without `onError`, an error on the stream is thrown from the returned Future and the subscription is cancelled. With `onError`, errors are passed to it and listening continues. ## What you must do with the Future bloc treats the handler's own Future as the lifetime of the run. If the handler starts `emit.forEach(...)` but neither awaits nor returns it: 1. The handler returns at once and its Future completes. 2. bloc completes the run's Emitter, which cancels every subscription the Emitter still holds. 3. In debug builds an assertion fires before that cancellation: an event handler completed but left pending subscriptions behind, most likely an unawaited `emit.forEach` or `emit.onEach`. In release builds there is no assertion, so the stream updates simply stop almost immediately. Write `return emit.forEach(...)`, `await emit.forEach(...)` in an `async` handler, or an arrow handler `(event, emit) => emit.forEach(...)`. ## Option 2 — a manual subscription and a private event The Bloc subscribes in its constructor, adds a private event such as `_RecentCitiesChanged` for each value, and cancels the `StreamSubscription` in an overridden `close()`. ## Trade-offs, per the bloc FAQ - **forEach/onEach**: no internal event needed; no subscription to manage; you control when subscribing starts by adding a public start event. - **forEach/onEach costs**: you cannot easily pause or resume the subscription; a public start event must be added from outside; you cannot put a transformer on the individual stream updates. - **Manual subscription**: more code and a `close()` override to remember, but each update is an event with its own handler and transformer, visible to `onEvent` and in `Transition` logs. ## Practical guidance - Put `restartable()` on the start event, so adding it again cleanly resubscribes instead of stacking two subscriptions. - Always supply `onError` when the stream can fail and the screen should survive it. - Keep `onData` pure and fast: it runs for every value, and heavy work belongs in the repository.
- What happens when the Bloc is closed while emit.forEach is still listening?close() closes the event stream and cancels every active Emitter; cancelling the Emitter cancels its forEach subscription and completes the returned Future. close() then cancels the handler subscriptions and closes the state stream, so no extra cleanup code is needed for the forEach subscription.
- Why put restartable() on the event that starts the subscription?With the default concurrent transformer, adding the start event twice runs two handlers, each with its own forEach subscription, so every value is emitted twice. restartable() cancels the first run, and with it the first subscription, before starting the second.
saying these in an interview costs you the question
- emit.forEach keeps listening even if the handler does not await it
- onEach's onData must return the next state
- A stream error inside forEach is always swallowed silently
- You must cancel forEach's subscription yourself in close()
- forEach's Future completes only when the Bloc is closed