Flutter : stream two Streams into a single screen?

Question

I have two streams fetching from two different api.

Stream<Month> get monthOutStream => monthOutController.stream;
Stream<MySchedule> get resultOutStream => resultController.stream;

I am fetching these data at two different state of the application, result at the begining and Months after some events from user.

MyScheduleBloc(){
  initialData();
}

Future initialData() async {
  MySchedule mySchedule = await myScheduleViewModel.importMySchedule(now.id);
  resultController.add(mySchedule);
}

My screen has a streambuilder as

Widget build(BuildContext context) {
final webCalenderBloc = WebCalenderBloc();
return StreamBuilder(
  stream: webCalenderBloc.resultOutStream,
  builder: (BuildContext context , snapdata){
    if(!snapdata.hasData){
      return Center(
        child: CircularProgressIndicator(),
      );
    }
    return body(snapdata.data);
   },
 );
}

Since the main widget build method took the StreamBuilder Widget with resultoutstream as a stream.Where do i fetch the other stream monthoutStream. Can i fetch stream inside a stream? Do i missing anything while handling two stream.I dont want to build any widget from monthoutstream but want to check the data inside it.

Answer 1

Well in my case, I prefer merging more than one stream into one, if they are of the same type. So you can use:

import 'package:async/async.dart' show StreamGroup;
...
StreamGroup.merge([stream1,stream2]);

Answer 2

error: was showing null check operator used on a null value. then i put it under try{}catch(e){} block & both streams work fine.. check the picenter image description here

Answer 3

multiple_stream_builder package:

Widget build(BuildContext context) {
  return StreamBuilder3<int, int, int>(
    streams: Tuple3(stream1, stream2, stream3),
    initialData: Tuple3(0, 0, 0),
    builder: (context, snapshots) {
      return Text(
        'stream1: ${snapshots.item1.data} - stream2: ${snapshots.item2.data} - stream3: ${snapshots.item3.data}',
      );
    },
  );
}

multi_stream_builder package:

Widget build(BuildContext context) {
  return MultiStreamBuilder(
    streams: [stream1, stream2],
    builder: (context, dataList) {
      final stream1Value = dataList[0];
      final stream2Value = dataList[1];
      return Text('$stream1Value, $stream2Value');
    },
  );
}

Answer 4

I'm using a sort of BLoC where I have broadcast Stream<Null>s that just notify listeners when certain things change. Kind of like Qt's signals & slots. Anyway for the case where I want to listen to more than one stream I made this class. It's basically StreamBuilder but you can listen to more than one stream, and it discards any data from the streams.

import 'dart:async';

import 'package:flutter/widgets.dart';

typedef MultiStreamWidgetBuilder<T> = Widget Function(BuildContext context);

// A widget that basically re-calls its builder whenever any of the streams
// has an event.
class MultiStreamBuilder extends StatefulWidget {
  const MultiStreamBuilder({
    required this.streams,
    required this.builder,
    Key? key,
  }) : super(key: key);

  final List<Stream<dynamic>> streams;
  final MultiStreamWidgetBuilder builder;

  Widget build(BuildContext context) => builder(context);

  @override
  State<MultiStreamBuilder> createState() => _MultiStreamBuilderState();
}

class _MultiStreamBuilderState extends State<MultiStreamBuilder> {
  final List<StreamSubscription<dynamic>> _subscriptions = [];

  @override
  void initState() {
    super.initState();
    _subscribe();
  }

  @override
  void didUpdateWidget(MultiStreamBuilder oldWidget) {
    super.didUpdateWidget(oldWidget);
    if (oldWidget.streams != widget.streams) {
      // Unsubscribe from all the removed streams and subscribe to all the added ones.
      // Just unsubscribe all and then resubscribe. In theory we could only
      // unsubscribe from the removed streams and subscribe from the added streams
      // but then we'd have to keep the set of streams we're subscribed to too.
      // This should happen infrequently enough that I don't think it matters.
      _unsubscribe();
      _subscribe();
    }
  }

  @override
  Widget build(BuildContext context) => widget.build(context);

  @override
  void dispose() {
    _unsubscribe();
    super.dispose();
  }

  void _subscribe() {
    for (final s in widget.streams) {
      final subscription = s.listen(
        (dynamic data) {
          setState(() {});
        },
        onError: (Object error, StackTrace stackTrace) {
          setState(() {});
        },
        onDone: () {
          setState(() {});
        },
      );
      _subscriptions.add(subscription);
    }
  }

  void _unsubscribe() {
    for (final s in _subscriptions) {
      s.cancel();
    }
    _subscriptions.clear();
  }
}

Example use:

class AppWidget extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return MultiStreamBuilder(
      streams: [appState.repoListChanged, appState.selectedRepoChanged],
      builder: _buildMain,
    );
  }

  Widget _buildMain(BuildContext context) {
    return Scaffold(
      body: Row(
...

I only just wrote it, so I haven't tested it much. I think in theory you could make one that gives you the states, though I'm not sure Dart's type system is advanced enough to let you do it without resorting to dynamic all over the place.

Answer 5

Observable.combineLatest2(
        aStream,
        bStream,
        (a, b, c) =>
        a != '' && b != '');

combineLatestN returns a combined stream

Answer 6

You can nest StreamBuilder if needed. Nothing prevents you from doing the following:

StreamBuilder(
  stream: stream1,
  builder: (context, snapshot1) {
    return StreamBuilder(
      stream: stream2,
      builder: (context, snapshot2) {
        // do some stuff with both streams here
      },
    );
  },
)

Another solution if this makes sense for you is: Streams are designed to be mergeable/transformed. You could make a third stream that is a merge of the two later streams.

Ideally for complex stream operations you'll want to use rxdart as it provides a few useful transformer.

Using rxdart, the fusion of two Observable (which are subclass of Stream) would be the following:

Observable<bool> stream1;
Observable<String> stream2;

final fusion = stream1.withLatestFrom(stream2, (foo, bar) {
  return MyClass(foo: foo, bar: bar);
});