Flutter – How to divide event stream effectively with Cubit

I would like to thank PurplePolyhedron for his answer, which set me off in the right direction but didn't quite get me where I wanted to go.

The crux of the issue seems to be that I was bringing Cubit in at too low of a level. At the service level, I really needed to be dealing with streams themselves and then introducing cubit when I needed to bridge to the presentation layer.

To do this, I first changed my TreadmillControlService to have two StreamControllers set up to broadcast(). One for treadmill status changes and one for workout status changes. In my characteristic listener handler, I can then write to these two streams separately.

// Irrelevant methods redacted for brevity
class TreadmillControlService implements Disposable {
  final StreamController<WorkoutStatus> _workoutStatusStreamController;
  final StreamController<TreadmillState> _treadmillStateStreamController;

  late Stream workoutStatusStream;
  late Stream treadmillStateStream;

  TreadmillControlService()
      : _workoutStatusStreamController = StreamController<WorkoutStatus>.broadcast(),
        _treadmillStateStreamController = StreamController<TreadmillState>.broadcast() {
    workoutStatusStream = _workoutStatusStreamController.stream;
    treadmillStateStream = _treadmillStateStreamController.stream;
    FlutterBluePlus.setLogLevel(LogLevel.warning, color: false);
  }

  void _processStatusUpdate(List<int> value) {
    final workoutStatus = WorkoutStatus.fromBytes(value);
    // Make a factory for this or something
    final treadmillState = TreadmillState(
        speedState: workoutStatus.speedInKmh == _requestedSpeed
            ? SpeedState.steady
            : workoutStatus.speedInKmh < _requestedSpeed
                ? SpeedState.increasing
                : SpeedState.decreasing,
        connectionState: _device!.isConnected ? ConnectionState.connected : ConnectionState.disconnected,
        requestedSpeed: _requestedSpeed,
        currentSpeed: workoutStatus.speedInKmh);
    _workoutStatusStreamController.add(workoutStatus);
    _treadmillStateStreamController.add(treadmillState);
  }
}

This has the added benefit of being able to listen to these streams in other parts of the business logic (hence the broadcast).

Then I have two separate cubits for presenting these streams to the UI:

Treadmill State:

class TreadmillStateCubit extends Cubit<TreadmillState> {
  final TreadmillControlService _treadmillControlService;

  TreadmillStateCubit(this._treadmillControlService) : super(TreadmillState.initial()) {
    _treadmillControlService.treadmillStateStream.listen((state) {
      emit(state);
    });
  }

  void connect() => _treadmillControlService.connect();

  void start() => _treadmillControlService.start();

  void stop() => _treadmillControlService.stop();

  void speedUp() => _treadmillControlService.speedUp();

  void speedDown() => _treadmillControlService.speedDown();
}

Workout State:

class WorkoutStatusCubit extends Cubit<WorkoutStatus> {
  final TreadmillControlService _treadmillControlService;

  WorkoutStatusCubit(this._treadmillControlService) : super(WorkoutStatus.zero()) {
    _treadmillControlService.workoutStatusStream.listen((state) {
      emit(state);
    });
  }
}

I also had to implement Equatable on both WorkoutStatus and TreadmillState to enable bloc to be able to accurately determine state changes. I don't think this would have been necessary for the stream coming from the real treadmill control service, as new objects are created each time. However, I have a simulation that mutates the state between emits.

I then had to wrap my controls in a MultiBlocProvider:

class ControlPage extends StatelessWidget {
  final TreadmillControlService treadmillControlService;
  const ControlPage(this.treadmillControlService, {Key? key}) : super(key: key);

  @override
  Widget build(BuildContext context) {
    return MultiBlocProvider(
      providers: [
        BlocProvider<WorkoutStatusCubit>(
          create: (ctx) => WorkoutStatusCubit(treadmillControlService),
        ),
        BlocProvider<TreadmillStateCubit>(
          create: (ctx) => TreadmillStateCubit(treadmillControlService),
        ),
      ],
      child: const Column(
        crossAxisAlignment: CrossAxisAlignment.stretch,
        children: [
          WorkoutStatusPanel(),
          TreadmillControls(),
        ],
      ),
    );
  }
}

And moved the relevant sub-widgets into their own classes, each with a BlocBuilder for their relevant Cubit:

WorkoutStatusPanel:

class WorkoutStatusPanel extends StatelessWidget {
  const WorkoutStatusPanel({super.key});

  @override
  Widget build(BuildContext context) {
    return BlocBuilder<WorkoutStatusCubit, WorkoutStatus>(builder: (ctx, state) {
      return Column(crossAxisAlignment: CrossAxisAlignment.stretch, children: [
        Row(
          children: [
            UnitQuantityCard(state.timeInSeconds, "s", 0),
            UnitQuantityCard(state.distanceInKm, "km", 2),
          ],
        ),
        Row(
          children: [
            UnitQuantityCard(state.indicatedCalories, "kCal", 0),
            UnitQuantityCard(state.steps, "steps", 0),
          ],
        )
      ]);
    });
  }
}

TreadmillControls:

class TreadmillControls extends StatelessWidget {
  const TreadmillControls({super.key});

  @override
  Widget build(BuildContext context) {
    final controlCubit = BlocProvider.of<TreadmillStateCubit>(context);
    return BlocBuilder<TreadmillStateCubit, TreadmillState>(builder: (ctx, state) {
      return Column(children: [
        Text("Connection: ${state.connectionState.name}"),
        Text("Speed ${state.speedState.name} (${state.currentSpeed}/${state.requestedSpeed})"),
        ElevatedButton(
          onPressed: () {
            controlCubit.connect();
          },
          child: const Text('Connect'),
        ),
        ElevatedButton(
          onPressed: () {
            controlCubit.start();
          },
          child: const Text('Start'),
        ),
        ElevatedButton(
          onPressed: () {
            controlCubit.stop();
          },
          child: const Text('Stop'),
        ),
        ElevatedButton(
          onPressed: () {
            controlCubit.speedUp();
          },
          child: const Text('Speed Up'),
        ),
        ElevatedButton(
          onPressed: () {
            controlCubit.speedDown();
          },
          child: const Text('Speed Down'),
        ),
      ]);
    });
  }
}

This now works nicely and we have a good division of responsibilities.