Bloc: Allow to emit current state on bloc subscription

[RaviKavaiya]

Hello @felangel ...

Is your feature request related to a problem? Please describe.
This has happened due to v6.0.0 release. We are not emitting bloc's current state when subscribing from now. But, that has created big problems for us. We would have to refactor each and every subscription (they are around 100). I visited https://bloclibrary.dev/#/migration?id=rationale-1. But unfortunately, that solution won't work for our app.

Our use case is, if certain event happens, do a subscription on X bloc, according to its state, react...

Your proposed solution from Migration is:

final bloc = MyBloc();
print(bloc.state);
bloc.listen(print);

Now, the problem is, we don't know the current state of bloc, until we hit the bloc.state statement. Suppose, using bloc.state, I do a process, but I continuously want to know if this state comes again at later time, I want to do the same. In above scenario, the same process can be invoked repeatedly, which is undesirable.

Describe the solution you'd like
Since you have made this change, I can't ask you to undo, but can we have some parameter like, emitOnListen(something like that) so that we can use blocs just as earlier and don't hit much breaking changes?

Thanks.

[Jomik]

Hey!
Could you provide a simple example that shows this issue?
Pushing bloc.state to your listener before subscribing directly simulates what would happen if the bloc emitted when listening, so I am unable to see where the issue is, without more information.

[innocentalone]

@RaviKavaiya I am facing the same issue. This change has cause very heavy refactoring to my apps. I wanted to stay on previous version, but VS Code extension forces me to update. :shrug:

[Jomik]

I think you could simply do this?

extension ListenWithCurrent<State, Event> on Bloc<State, Event> {
  StreamSubscription<State>
      listenWithCurrent<B extends Bloc<dynamic, State>, State>(
    B bloc,
    void Function(State) onData, {
    Function onError,
    void Function() onDone,
    bool cancelOnError,
  }) {
    onData(bloc.state);

    return bloc.listen(
      onData,
      onError: onError,
      onDone: onDone,
      cancelOnError: cancelOnError,
    );
  }
}

final bloc = MyBloc();
bloc.listenWithCurrent(print);

[balesz]

This was the correct solution in a previous version of bloc: https://github.com/felangel/bloc/blob/ae4e14bbe5b401b574f08a761a45db3381e4fab0/packages/bloc/lib/src/bloc.dart#L82

It worked like a BehaviorSubject in RxDart or RxJS. It emitted the current value of the state on listen.

[RaviKavaiya]

@Jomik Ok, thanks for the reply. I guess, that (extension method) should work for me.
Lets see what are @felangel 's thoughts...

[felangel]

Hi @RaviKavaiya 👋
Thanks for opening an issue and sorry for the inconvenience!

This breaking change was made to enable interop with cubit, align with Dart Streams, as well to enable a lot of other features and improvements. I would highly recommend using the extension only as a temporary solution and refactor your blocs (at your own pace) to respect the new behavior. I'm happy to help with any cases that you have trouble refactoring and again don't feel pressured to upgrade/refactor all at once if it is overwhelming -- you can tackle it incrementally.

Hope that helps and again sorry for any inconvenience!

Closing this for now but feel free to comment with any additional questions/concerns and I'm happy to continue the conversation 👍

[balesz]

@felangel I think this BehaviorSubject like behavior of the Stream implementation is a key feature of the whole Cubit/Bloc package. The problem is that this behavior is not implemented on the Cubit class. In the previous cubit implementation (v0.1.2) the CubitStream class implenet this behavior and it works as expected, but the new merged bloc package (v6.0.x) is not working.

[balesz]

@felangel I just realized that this behavior is intent. Why? If someone don't want to get the current state from the stream on listen why not just use bloc.skip(1)?

[felangel]

@balesz this is intentional for the reasons I mentioned above and you can access the current state at any time via the bloc.state if you do need the current state. Also it caused many bugs in cubit and was removed in the 0.2.0-dev release.

Hope that helps 👍

[balesz]

@felangel I think this is a must have feature every Stream based state management library. But if you see it differently I have to accept it. This is a quick fix for those who also lack this feature. Add BehaviorSubjectBloc mixin to your Bloc class:

class CounterBloc extends Bloc<int, int> with BehaviorSubjectBloc {
  ...
}

mixin BehaviorSubjectBloc<TEvent, TState> on Bloc<TEvent, TState> {
  @override
  StreamSubscription<TState> listen(void Function(TState state) onData,
      {Function onError, void Function() onDone, bool cancelOnError}) {
    onData(this.state);
    return super.listen(onData,
        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
  }
}

[felangel]

@balesz this is not how Dart Streams work and also widgets like BlocBuilder and BlocListener previously had to workaround this behavior using skip(1) as well as the bloc_tests. As you mentioned you can always support this functionality if you want but I do not think it should be the default. Again, sorry for any inconvenience!

[marcin-jelenski]

@felangel thank you for finding a duplicate. I read the change rationale, but I don't get it. All stream functions - first, take are locked until we emit a second state. That's not intuitive. I'd expect a bloc not to extend a stream then. Also, what should be a second emmited state if a bloc doesn't perform any subscription in a constructor, and just responds to user actions?

Would you mind providing some samples where it's helpful or exact issues, which were caused by first-state emission?

https://api.flutter.dev/flutter/dart-async/Stream/Stream.value.html
Value streams doesn't skip any values. Neither do Rx behaviour subjects. What's the relation?

[felangel]

@marcin-jelenski a lot of issues have been raised regarding the need to skip(1) everywhere (BlocBuilder, BlocListener, BlocConsumer, bloc_test, emitsExactly, etc...) I don't have time to find each of those issues right now but can search for them later.

Regarding the cubit interop issues you can check out:

• https://github.com/felangel/cubit/issues/52
• https://github.com/felangel/cubit/issues/60#issuecomment-654375187

Why is it not intuitive? When using a plain Dart Stream and subscribing you will only receive data emitted after the subscription. Stream.value only emits a single value which is not at all like a Bloc and Rx BehaviorSubjects are designed to add this behavior on top of Dart Streams.

import 'dart:async';

void main() async {
  final controller = StreamController<int>.broadcast();
  controller.add(0);
  controller.stream.listen(print);
  controller..add(1)..add(2)..add(3);
}

Outputs:

1
2
3

It would be very helpful if you could provide some concrete use-cases where you feel it's necessary to have the previous behavior and I would be happy to take a look 👍

[marcin-jelenski]

@felangel thank you for the response 🏎️

I have two cases where previous behaviour was priceless and we can't find any migration paths for them (despite some workarounds like @balesz suggested).

1. Distinct

For performance reasons, we've been mapping state fields and setting a distinct operator on them. The same time we used stream builders instead of BlocBuilder. Providing a separate bloc for each field isn't an option as it would add too much overhead.

StreamBuilder(
        stream: bloc.map((state) => state.field)
                    .distinct()
                    .map((field) => field.heavyComputation()),
        builder: (context, snapshot) {
          if (snapshot.hasData) {
            return MapComponent(snapshot.data); // we don't want to rebuild a heavy component each time state changes
          } else if (snapshot.hasError) {
            return Text("Error");
          } else {
            return Text("Loading");
          }
        });

2. State-to-value mapping

Based on state type and fields we'd like to receive a stream, which is accessed in a view to show / hide a progress bar. Worth to mention, that's something we've been sharing between flutter and angular dart.

class Event {}

abstract class State {}
class Initial implements State {}
class Loading implements State {}
class Ready implements State {}
class Sending implements State {}
class Sent implements State {}

class SampleBloc extends Bloc<Event, State> {
  SampleBloc() : super(Initial());

  Stream<bool> isLoading => map((state) => state is Loading || state is Sending);
}

---

Also, bloc as a resource is more like a behaviour subject than a stream to me. If it emits a state, it could repeat the latest value to late-subscribers (subscribers that are made in the meantime, not on start). Isn't it a case, the race errors were caused by performing emissions before anything subscribed? Maybe just replying the latest state to the subscribers would be an option. I haven't seen any similar issue on other platforms using such subjects.

[RaviKavaiya]

@marcin-jelenski Exactly, I totally agree with your point no. 2 (state to value). Have the same use case here 🙂

[AntiqPeace]

Thank god I found the reason why my code isn't working 🙄. As @marcin-jelenski pointed last, I am also using late subscribers. Ofcourse, solutions are posted above, but they don't seem well to me, TBH.

But one question @marcin-jelenski , your second point (state-2-value mapping) doesn't work in latest version?

[felangel]

@marcin-jelenski thanks for providing some more detail 👍

1. I'm not sure I understand why this approach is necessary. Why is the field performing a heavy computation rather than the bloc itself. I think you should try to instead invert this so that the bloc is the one performing the heavy computation as needed and the state contains the result which is consumed in the UI. Also, I don't understand why you wouldn't use BlocBuilder instead of StreamBuilder in this case (for example snapshot.hasData, snapshot.hasError are unnecessary as they can be derived from the state) and BlocBuilder handles using the last state to seed the builder until new states are emitted.

BlocBuilder<Bloc, State>(
  buildWhen: (previous, current) => previous.field != current.field,
  builder: (context, state) {
    switch (state.status) {
      case Status.loading:
        return Text('Loading');
      case Status.success:
        return MapComponent(state.field);
      default:
        return Text('Error');
    }
  },
);

1. Again why isn't this just part of the bloc state? Why do you need to expose a separate stream for this?

class Event {}

abstract class State {}
class Initial implements State {}
class Loading implements State {}
class Ready implements State {}
class Sending implements Loading {} // extend Loading
class Sent implements State {}

class SampleBloc extends Bloc<Event, State> {
  SampleBloc() : super(Initial());
}

BlocBuilder(
  buildWhen: (previous, current) => previous.runtimeType != current.runtimeType,
  builder: (context, state) {
    if (state is Loading) return LoadingIndicator();
    ...
  },
);

Also, bloc as a resource is more like a behaviour subject than a stream to me. If it emits a state, it could repeat the latest value to late-subscribers (subscribers that are made in the meantime, not on start). Isn't it a case, the race errors were caused by performing emissions before anything subscribed? Maybe just replying the latest state to the subscribers would be an option. I haven't seen any similar issue on other platforms using such subjects.

I disagree because BlocListener, BlocBuilder, the BlocPipe, blocTest, emitsExactly all had to compensate for this behavior and skip the first state in order to prevent unintended side-effects such as unnecessary rebuilds with duplicate states, multiple listener invocations when navigating to/from a screen, etc...

I know the new behavior is quite a drastic change but I believe it is for the better and most of the problems you mentioned can be solved by eliminating the manual subscriptions and parallel streams and relying on BlocBuilder instead of StreamBuilder.

Thoughts?

[felangel]

@AntiqPeace can you provide some code? The following should work if you really want to have the previous behavior but I don't recommend it.

mixin BehaviorSubjectMixin<E, S> on Bloc<E, S> {
  @override
  StreamSubscription<S> listen(
    void Function(S state) onData, {
    Function onError,
    void Function() onDone,
    bool cancelOnError,
  }) {
    onData(state);
    return super.listen(
      onData,
      onError: onError,
      onDone: onDone,
      cancelOnError: cancelOnError,
    );
  }
}

class Event {}
abstract class State {}
class Initial implements State {}
class Loading implements State {}
class Ready implements State {}
class Sending implements State {}
class Sent implements State {}

class SampleBloc extends Bloc<Event, State> with BehaviorSubjectMixin {
  SampleBloc() : super(Initial());

  Stream<bool> get isLoading =>
      map((state) => state is Loading || state is Sending);

  @override
  Stream<State> mapEventToState(Event event) async* {}
}

void main() {
  SampleBloc().isLoading.listen(print); // false
}

[marcin-jelenski]

@felangel I'll try to elaborate

1. I'm not sure I understand why this approach is necessary. Why is the field performing a heavy computation rather than the bloc itself.

In the above example (as well as in our apps) heavy computation function was moved out of a bloc because it's platform specific. To be even more precise, on a flutter target we're building a list of "flutter map markers" for a stateless widget and on angular dart we have to provide a list of "leaflet map markers". Without providing a distinct method, a whole map widget would be rebuilt each time state fields changes (even if they're not related to markers) and buildWhen filter isn't as convenient as streams - we'd have to transform bloc state two times (first for comparison, second for emission) or write bunch of ifs, which is 🙅 .

1. Again why isn't this just part of the bloc state? Why do you need to expose a separate stream for this?

Please note in an example, there're multiple states which causes loading to be visible. There're cases where we're doing much more than simple checks - value formatting etc. Also, we still would like to share such method between angular and flutter. An alternative option would be writing an extension function on a state:

extensions StateExtensions on State {
  bool get isLoading => this is Loading || this is Sending
}

But that option still can't operate on distinction, and forces us to rebuild each widget each time. (Also doable with buildWhen but not convenient as in pt.1. and error-prone).

In all above examples performance is a key. We could build one widget with BlocBuilder, but that's a performance drain.

[felangel]

@marcin-jelenski couldn't you accomplish this then by providing an initialData to the StreamBuilder?

StreamBuilder(
        initialData: bloc.state,
        stream: bloc.map((state) => state.field)
                    .distinct()
                    .map((field) => field.heavyComputation()),
        builder: (context, snapshot) {
          if (snapshot.hasData) {
            return MapComponent(snapshot.data); // we don't want to rebuild a heavy component each time state changes
          } else if (snapshot.hasError) {
            return Text("Error");
          } else {
            return Text("Loading");
          }
        });

Regarding the second point, another option is to just make isLoading a generator:

import 'dart:async';

import 'package:bloc/bloc.dart';

class Event {}
abstract class State {}
class Initial implements State {}
class Loading implements State {}
class Ready implements State {}
class Sending implements State {}
class Sent implements State {}

class SampleBloc extends Bloc<Event, State> {
  SampleBloc() : super(Initial());

  Stream<bool> get isLoading async* {
    yield _isLoading(state);
    yield* map(_isLoading);
  }

  bool _isLoading(State state) => state is Loading || state is Sending;

  @override
  Stream<State> mapEventToState(Event event) async* {}
}

void main() {
  SampleBloc().isLoading.listen(print);
}

Thoughts?

[marcin-jelenski]

@felangel you're right, we can mitigate those distinction issues with such instructions. Thank you 🙏 🙏 🙏

[nerder]

The following should work if you really want to have the previous behavior but I don't recommend it.

Why you don't recommend the mixin approach? @felangel