Combine multiple BehaviorSubject streams into one?

Issue

The goal is for me to use multiple streams without having to nest 3+ StreamBuilders.

I’m already using rxdart and I looked into mergewith but I don’t really understand the proper syntax of using it?

I currently have a Bloc file titled InfoBloc. This is the code inside InfoBloc

  final _name = BehaviorSubject<String>();
  final _description = BehaviorSubject<String>();
  final _picture = BehaviorSubject<File>();

  Observable<String> get name => _name.stream.transform(_validateName);
  Observable<File> get picture => _picture.stream;
  Observable<String> get eventDescription =>
      _description.stream.transform(_validateMessage);

  final _validateMessage = StreamTransformer<String, String>.fromHandlers(
      handleData: (eventMessage, sink) {
    if (eventMessage.length > 0) {
      sink.add(eventMessage);
    } else {
      sink.addError(StringConstant.eventValidateMessage);
    }
  });

  var obs = Observable.merge([])
  final _validateName = StreamTransformer<String, String>.fromHandlers(
      handleData: (String name, sink) {
    if (RegExp(r'[!@#<>?":_`~;[\]\\|=+)(*&^%0-9-]').hasMatch(name)) {
      sink.addError(StringConstant.nameValidateMessage);
    } else {
      sink.add(name);
    }
  });

  void dispose() async {
    await _description.drain();
    _description.close();
    await _name.drain();
    _name.close();
    await _picture.drain();
    _picture.close();
  }

Now inside of a widget, I need the name, picture, and description snapshot. So I would normally do

 StreamBuilder(
   stream: _bloc.name,
   builder: (BuildContext context, AsyncSnapshot<String> snapshotName) {
     return StreamBuilder(
       stream: _bloc.eventDescription,
       builder: (BuildContext context, AsyncSnapshot<String> snapshotDescription) {
         return StreamBuilder(
           stream: _bloc.picture,
           builder: (BuildContext context, AsyncSnapshot<File> snapshotName) {

But there has got to be a better way of doing this.

The dream is that I can make something inside the InfoBloc file that can combine all these streams and I only have to use StreamBuilder once to stream that combined stream.

Solution

You can check the combineLatest() method for the Observable class. It merges the given Streams into one Observable sequence by using the combiner function such that the resulting Observable won’t emit unless all the Streams have emitted atleast one item.

There are a bunch of combineLatest() methods. But since you have 3 Streams you can use combineLatest3().

Example –

Observable.combineLatest3(
  new Observable.just("a"),
  new Observable.just("b"),
  new Observable.fromIterable(["c", "c"]),
  (a, b, c) => a + b + c)
.listen(print); //prints "abc", "abc"

You can check this link for more information.

UPDATE –

You can refer to the below example for using the funciton in your code.

Observable<String> get name => _name.stream.transform(_validateName);
Observable<File> get picture => _picture.stream;
Observable<String> get eventDescription => _description.stream.transform(_validateMessage);
//Add this line to combine your Streams
Observable<bool> readyToSubmit => Observable.combineLatest3(name, picture, eventdescription, (value1, value2, value3) => true);
//We simply return true from the combiner function since we don't want to perform any operation on the Streams when combining.

I have used a bool type Observable just for example. You can change the type to Map<String, dynamic>' and add the Stream values into theMapif you want to access the data from all three streams. And in yourStreamBuilder` widget you can use –

StreamBuilder(
   stream: _bloc.readyToSubmit,
   builder: //Your builder function here
)

Hope this helps!

Answered By – thedarthcoder

Answer Checked By – Clifford M. (FlutterFixes Volunteer)