Connecting blocs through domain layer: How to listen to one stream from multiple blocs?

Question

I have a method in repository that gets some data from api when it is called. this method can get called from two, three or more blocs. so to make the api calls optimized, this method should only get called once and these three blocs should get the result.

According to bloc documentation and the example provided here , I have changed the repository method to a stream like this:

 Stream<UserInfo> getUserInfo(
  {required String userKey, required String caller}) async* {
print("Repo Caller is $caller");

  UserInforesponse =
      await provider.getUserInfo(
    patientKey: patientKey,
  );

  yield response;
  }

And in blocs that need this method's result I am doing like this:

await emit.forEach(
      repo.getUserInfo(userKey: userKey,caller:"bloc1"),
      onData: (UserInfo response) {
        print("bloc1 got data");
        return state.copyWith( ...  );
      },
    );

And the same method in bloc 2 ,3 ...:

await emit.forEach(
      repo.getUserInfo(userKey: userKey,caller:"bloc2"),
      onData: (UserInfo response) {
        print("bloc2 got data");
        return state.copyWith(  ...  );
         
      },
    );

But the Problem is that these blocs are not listening to one Stream. They are creating their own individual stream. So as the result this is printed in the output:

 Repo Caller is bloc1
 Repo Caller is bloc2

So the api service is call twice and if I need this api result in 5 blocs, it will get called 5 times not one time. How can I achive the expected behavior as the example in the documents is not covering this case.

Answer 1

From the docs of forEach method in emmiter:

  /// Subscribes to the provided [stream] and invokes the [onData] callback
  /// when the [stream] emits new data and the result of [onData] is emitted.
  /// .....
  Future<void> forEach<T>(...

Next question is: how do you create repo instance and how do you provide/inject it to the blocs - is it the same instance? As an option, in your blocs you could do something common like this:

class Bloc1 extends Bloc<SomeEvent, SomeState> {
  final YourRepository _yourRepository
  final StreamSubscription _subscription;

      Bloc1(YourRepository yourRepository) : 
        _yourRepository = yourRepository,
        super(UserState.initial()) {
        ...
        _subscription = _yourRepository.userInfoStream.listen((userInfo){
            emit(state.copyWith(...));
          }, onError: (error) {
            emit(state.copyWith(...));
          });
        ...
      }

      void getUserInfo(
             required String userKey, 
             required String caller,
      ){
         _yourRepository.getUserInfo(userKey, caller);
      }
    
      @override
      Future<void> close() {
        _subscription?.cancel();
        return super.close();
      }
    }
...

And in your repo something like this:

    class YourRepository{
      final _userInfoController = StreamController<UserInfo>.broadcast();
      Stream<UserInfo> get userInfoStream => _userInfoController.stream;
    
     void getUserInfo({
            required String userKey, 
            required String caller,
     }) async {
    
         final userInforesponse = await provider.getUserInfo(
           patientKey: patientKey,
         );
    
         _userInfoController.add(userInforesponse);
     }
    
     void dispose() {
       _userInfoController.close();
     }
   }

Long story short: in your repo you have one stream that is visible from the outside, all blocs just subscribe and listen this stream.