feat: Add emitForEach / emitOnEach Methods (or something similar) to Cubit

[jtdLab]

Description

When working with a Cubit that consume some other Stream you need to maintain a StreamSubscription during the life of the Cubit and close it etc. This comes with some extra work which is handled well in Bloc via emit.forEach / emit.onEach but no similar solution exists for that when using Cubit.

Desired Solution

Adding the methods emitForEach and emitOnEach with same signatures and functionality as their counter part in Emitter.

Alternatives Considered

Maybe give a Cubit its own Emitter.

[jyardin]

I think the alternative solution you proposed of exposing an Emitter would be better, as it would give a more consistent experience between Bloc and Cubit.

It would be even cooler if the Emitter is returned by an emit property, replacing the emit(State) method.
It would be a breaking change, not sure how big a breaking change though, as the main usage of emit (just calling emit(newState)) would still work as intended.

[vgtle]

This would be a great feature. Are there any plans of implementing this? Or any ways to contribute to this?

[zamaniafshar]

+1

[zamaniafshar]

It would be great if Bloc supported this. Until then, the code below might be useful; I just made some changes to the Emitter class.

base class StreamableCubit<State> extends Cubit<State> {
  StreamableCubit(super.initialState);

  final _completer = Completer<void>();
  final _disposables = <FutureOr<void> Function()>[];

  Future<void> get future => _completer.future;

  @override
  Future<void> close() async {
    for (final disposable in _disposables) {
      disposable.call();
    }
    _disposables.clear();
    if (!_completer.isCompleted) _completer.complete();
    super.close();
  }

  Future<void> onEach<T>(
    Stream<T> stream, {
    required void Function(T data) onData,
    void Function(Object error, StackTrace stackTrace)? onError,
  }) {
    final completer = Completer<void>();
    final subscription = stream.listen(
      onData,
      onDone: completer.complete,
      onError: onError ?? completer.completeError,
      cancelOnError: onError == null,
    );
    _disposables.add(subscription.cancel);
    return Future.any([future, completer.future]).whenComplete(() {
      subscription.cancel();
      _disposables.remove(subscription.cancel);
    });
  }

  Future<void> forEach<T>(
    Stream<T> stream, {
    required State Function(T data) onData,
    State Function(Object error, StackTrace stackTrace)? onError,
  }) {
    return onEach<T>(
      stream,
      onData: (data) => emit(onData(data)),
      onError: onError != null
          ? (Object error, StackTrace stackTrace) {
              emit(onError(error, stackTrace));
            }
          : null,
    );
  }
}

[felangel]

I've thought about this a bit and have come to the conclusion that the bloc/emitter APIs don't really make sense in the context of cubit because the API doesn't provide a way for developers to control the lifetime of the subscription.

For example, if multiple public methods on the cubit used emit.forEach and subscribed to the same stream what should happen if they are called back-to-back? Currently, both subscriptions would stay alive for the lifetime of the cubit which would mean that whenever emit.forEach was used (as long as the underlying stream was open) it would continue to emit new states. This could be very confusing and bloc handles this by closing the underlying subscription whenever a new event is processed. We can't do the same in cubit because there are no hooks for when a public method gets called in cubit so there's no way to intercept or detect that a method has finished executing.

I think the best we could do is offer an API that just registers a subscription for the lifetime of the cubit and the cubit could then handle canceling all subscriptions when it is closed but I'm not sure that is what folks would want:

class MyCubit extends Cubit<MyState> {
  void doSomething() {
    // This would be equivalent to managing the subscription 
    // and canceling the subscription when the cubit is closed
    emit.onEach(stream, ...);
  }
  
  void doSomethingElse() {
    // This would register another subscription and calling `doSomething` and `doSomethingElse`
    // would potentially result in two subscriptions that are active simultaneously.
    emit.onEach(stream, ...);
  }
}

I'm guessing that's not what most people would want and instead they'd want the first subscription to be canceled when a new public method is called. At that point it doesn't seem worth it to introduce new APIs when the StreamSubscription API is designed exactly for this use-case.

[felangel]

I've spent a bit more time playing around with this and I actually do think we can support this -- I have a proof of concept and will try to get a PR up shortly 🎉

[vgtle]

That would be amazing to have

[zamaniafshar]

I'm guessing that's not what most people would want and instead they'd want the first subscription to be canceled when a new public method is called. At that point it doesn't seem worth it to introduce new APIs when the StreamSubscription API is designed exactly for this use-case.

If people want full control over their methods, they should use Bloc, which is specifically designed for event management. Cubit is better suited for simpler use cases. We don’t always need event management, but sometimes, even in Cubit, we may need to listen to multiple streams in the constructor and dispose of them when the Cubit is closed.

[felangel]

I have a draft PR up at #4293. The PR refactors cubit to share the Emitter and exposes onEach and forEach. The behavior in the PR is whenever emit.onEach or emit.forEach are called within a Cubit any existing subscriptions are canceled. The behavior in Bloc is unchanged. If you want long-running subscriptions across multiple method calls in a Cubit, then you'd still want to manage the subscription manually.

I'm still not convinced this is worth adding to Cubit so I'm leaning heavily on feedback from the community -- please let me know what you think, thanks! 💙

[vgtle]

We have 30+ Flutter apps and millions lines of code and we never really go for Bloc, since Cubits are way more convenient to use. There are only a few use cases where Blocs are the right choice. Therefore, I am really looking forward for the feature to be released

[zamaniafshar]

I have a draft PR up at #4293. The PR refactors cubit to share the Emitter and exposes onEach and forEach. The behavior in the PR is whenever emit.onEach or emit.forEach are called within a Cubit any existing subscriptions are canceled. The behavior in Bloc is unchanged. If you want long-running subscriptions across multiple method calls in a Cubit, then you'd still want to manage the subscription manually.

I'm still not convinced this is worth adding to Cubit so I'm leaning heavily on feedback from the community -- please let me know what you think, thanks! 💙

Thanks for the PR! However, this implementation isn't exactly what I'm looking for. I believe it may confuse users, as it could lead to unexpected behavior without clear explanations.

As I mentioned, if developers need to listen to a stream upon a user tap, they should either use Bloc, which is designed for handling events, or handle it manually.

Additionally, there's a way to add a tag in the emit.forEach or emit.onEach method in Cubits. This allows you to detect whether the provided stream is already being listened or not and implement the logic accordingly. Using this approach, we can achieve both behaviors similar to Bloc. However, the implementation in codes might be slightly more complex.

That said, as I mentioned, I'm not a big fan of this approach because if we need event management, we should use the right tool for it, which is Bloc.

However, in very common cases where we simply want to listen to one or multiple streams in the constructor of a Cubit and cancel the subscriptions in the close method, it’s better and simpler to use emit.onEach or emit.forEach. These handle cancellation for us, making the code cleaner and more consistent across the project, similar to how it is in Blocs.

[felangel]

After thinking about this some more and chatting with more folks I don't feel this change provides a ton of value since the functionality in this implementation cancels any pending streams whenever emit.forEach/emit.onEach is used (regardless of the method it's used in). This differs from the bloc implementation since emit.forEach/emit.onEach is scoped to the event handler so emit.forEach in event handler A won't cancel the subscription for emit.forEach in event handler B. This isn't possible to do with Cubit (at least not with the current public API) and I don't think this change is worth it since I fear it won't solve real problems folks have and it'll lead to confusion since the API is named the same but functions differently across bloc and cubit.