diff --git a/lib/counter/bloc/counter_bloc.dart b/lib/counter/bloc/counter_bloc.dart
index de82a71..c0f954d 100644
--- a/lib/counter/bloc/counter_bloc.dart
+++ b/lib/counter/bloc/counter_bloc.dart
@@ -12,26 +12,28 @@ class CounterBloc extends Bloc<CounterEvent, CounterState> {
       : _counterRepository = counterRepository,
         super(const CounterState()) {
     on<CounterStarted>(_onCounterStarted);
-    on<_CounterConnectionStateChanged>(_onCounterConnectionStateChanged);
-    on<_CounterCountChanged>(_onCounterCountChanged);
     on<CounterIncrementPressed>(_onCounterIncrementPressed);
     on<CounterDecrementPressed>(_onCounterDecrementPressed);
   }
 
   final CounterRepository _counterRepository;
-  StreamSubscription<int>? _countSubscription;
-  StreamSubscription<ConnectionState>? _connectionSubscription;
 
-  void _onCounterStarted(
+  Future<void> _onCounterStarted(
     CounterStarted event,
     Emitter<CounterState> emit,
-  ) {
-    _countSubscription = _counterRepository.count.listen(
-      (count) => add(_CounterCountChanged(count)),
+  ) async {
+    final countEmitter = emit.forEach<int>(
+      _counterRepository.count,
+      onData: (count) =>
+          state.copyWith(count: count, status: CounterStatus.connected),
+    );
+
+    final connectionEmitter = emit.forEach<ConnectionState>(
+      _counterRepository.connection,
+      onData: (state) => this.state.copyWith(status: state.toStatus()),
     );
-    _connectionSubscription = _counterRepository.connection.listen((state) {
-      add(_CounterConnectionStateChanged(state));
-    });
+
+    await Future.wait([countEmitter, connectionEmitter]);
   }
 
   void _onCounterIncrementPressed(
@@ -47,27 +49,6 @@ class CounterBloc extends Bloc<CounterEvent, CounterState> {
   ) {
     _counterRepository.decrement();
   }
-
-  void _onCounterConnectionStateChanged(
-    _CounterConnectionStateChanged event,
-    Emitter<CounterState> emit,
-  ) {
-    emit(state.copyWith(status: event.state.toStatus()));
-  }
-
-  void _onCounterCountChanged(
-    _CounterCountChanged event,
-    Emitter<CounterState> emit,
-  ) {
-    emit(state.copyWith(count: event.count, status: CounterStatus.connected));
-  }
-
-  @override
-  Future<void> close() {
-    _connectionSubscription?.cancel();
-    _countSubscription?.cancel();
-    return super.close();
-  }
 }
 
 extension on ConnectionState {
diff --git a/lib/counter/bloc/counter_event.dart b/lib/counter/bloc/counter_event.dart
index 8b5e97d..8d863c0 100644
--- a/lib/counter/bloc/counter_event.dart
+++ b/lib/counter/bloc/counter_event.dart
@@ -15,15 +15,3 @@ class CounterIncrementPressed extends CounterEvent {
 class CounterDecrementPressed extends CounterEvent {
   const CounterDecrementPressed();
 }
-
-class _CounterCountChanged extends CounterEvent {
-  const _CounterCountChanged(this.count);
-
-  final int count;
-}
-
-class _CounterConnectionStateChanged extends CounterEvent {
-  const _CounterConnectionStateChanged(this.state);
-
-  final ConnectionState state;
-}