flutter How to process bloc events in parallel?

Notice the main idea of using Bloc is predictability - you will lose that predictability to some degree (depending on your concrete implementation). If you are using flutter_bloc you could follow this suggestion and override the default event stream processing on your bloc.

@override
Stream<Transition<MyEvent, MyState>> transformEvents(
   Stream<MyEvent> events, transitionFn) {
      return events.flatMap(transitionFn);
}

You could also look into isolates and maybe especially flutters compute which lets you spin up an isolate to run your code. I found this to be a good source.

While I'm very sure there is a better way to do this I came up with the following. I've cut out some of your logic for it to be a little more generic.

I'm not familiar with the performance details of compute and isolate in dart, so I want to make the disclaimer that this might not be a best practice approach, but maybe it helps you getting started.

import 'package:bloc/bloc.dart';
import 'package:flutter/foundation.dart';
import 'package:flutter/material.dart';
import 'dart:async';

void main() {
  runApp(ExampleApp());
}

class ExampleApp extends StatelessWidget {
  static ExampleBloc bloc = ExampleBloc();
  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      home: TextButton(
        onPressed: () => bloc.add(ExampleStartingEvent()),
        child: Text("Trigger"),
      ),
    );
  }
}

// Top level function that is computed in isolate

Future<void> _mockRequest(String body) async {
  // Do your async request here and await response
  Future.delayed(Duration(seconds: 5));
  ExampleBloc.successfulCompute("Successful!");
}

// Bloc

class ExampleBloc extends Bloc<ExampleEvent, ExampleState> {
  ExampleBloc() : super(ExampleStateInitial());

  static successfulCompute(String response) {
    ExampleApp.bloc.add(ExampleEventSuccess(response));
  }

  @override
  Stream<ExampleState> mapEventToState(
    ExampleEvent event,
  ) async* {
    if (event is ExampleEventSuccess) {
      print(event.response);
      yield ExampleStateSuccess(event.response);
    }
    if (event is ExampleStartingEvent) {
      compute(_mockRequest, "body");
    }
  }
}

// Events

class ExampleEvent {}

class ExampleStartingEvent extends ExampleEvent {}

class ExampleEventSuccess extends ExampleEvent {
  final response;
  ExampleEventSuccess(this.response);
}

// States

class ExampleState {}

class ExampleStateInitial extends ExampleState {}

class ExampleStateSuccess extends ExampleState {
  final response;
  ExampleStateSuccess(this.response);
}

class ExampleStateError extends ExampleState {}

Just to show a solution based on @kohjakob 's proposal but with:

• no static methods
• complete error handling routines

The idea is basically to wrap the repository call into an async method (_sendClick(...)) and call it non-blocking (i.e. without await) while the status update on the sending state is done synchronously.

The _sendClick(...) waits for the repository and adds a ColorSendSuccess or ColorSendFailed event to the bloc once it's done. These events are then handle in their own run of the mapEventToState(...) routine.

class ColorsBloc extends Bloc<ColorsEvent, ColorsState> {
  final ColorRepository colorRepository;

  ColorsBloc({required this.colorRepository}) : super(ColorsState.initial());

  @override
  Stream<ColorsState> mapEventToState(
      ColorsEvent event,
      ) async* {
    if (event is ColorsFetchRequested) {
      yield ColorsState.loading();
      try {
        final colors = await colorRepository.getColors();
        yield ColorsState.success(colors);
      } catch (e) {
        yield ColorsState.error();
      }
    } else if (event is ColorCounted) {
      yield* _mapColorCountedToState(event);
    } else if (event is ColorSendSuccess) {
      yield _mapColorSendSuccessToState(event);
    } else if (event is ColorSendFailed) {
      yield _mapColorSendFailedToState(event);
    }
  }

  Stream<ColorsState> _mapColorCountedToState(ColorCounted event) async* {
    yield state.copyWith(
      sendingByColorId: {...state.sendingByColorId, event.colorId},
    );
    // non-blocking <----------------
    _sendClick(Color(
      colorId: event.colorId,
      timestamp: DateTime.now().millisecondsSinceEpoch,
    ));
    final colors = await colorRepository.getColors();
    yield state.copyWith(
      status: Status.success,
      colors: colors,
      sendingByColorId: {...state.sendingByColorId}..remove(event.colorId),
    );
  }

  Future<void> _sendClick(Color color) async {
    try {
     int newId = await colorRepository.storeColor(color);
     Color storedColor = color.copyWith(id: () => newId);
     add(ColorSendSuccess(color: storedColor));
    } on StoreColorClickException catch (_) {
      add(ColorSendFailed(color: color));
    }
  }

  ColorsState _mapColorSendSuccessToState(ColorCounted event) async* {
    return state.copyWith(
      colors: [...state.colors]
        // replace the local color-click with the stored one
        ..removeWhere((element) => element.localId == event.color.localId)
        ..add(event.color.copyWith(localId: () => null)),
      sendingByColorId: {...state.sendingByColorId}..remove(event.color.id),
    );
  }

  ColorsState _mapColorSendFailedToState(ColorCounted event) async* {
    return state.copyWith(
      colors: [...state.colors]
        // remove the color that has not been stored
        ..removeWhere((element) => element.localId == event.color.localId),
      sendingByColorId: {...state.sendingByColorId}..remove(event.color.localId),
      // mark the color as failed
      errorByColorId: {...state.errorByColorId, event.color.localId},
    );
  }
}