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.

Pentosan

Answer 1

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);
});

Oslo

Thank you for the information about merging two stream or using one after another.At some point i go the condition where the first stream has a data "true" then it has to pop out the Navigation and if "false" then move on to another stream. The problem arise as the widget was building and has to pop out Navigation though the pop out worked but for a sec there exist an error saying widget was building returned a null.How can i solve that while using stream one after another. – Pentosan

pop out the Navigation should be in listener, not in streamBuilder. – Differ

Answer 2

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

combineLatestN returns a combined stream

Congruent

I have tried this one, but is not working. StreamBuilder is returning error message with 'Bad state: Stream has already been listened to.' – Rhombus

@NihadDelic you might have closed the stream after receiving data or you are not using streamController.broadcast() . – Complaint

Answer 3

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]);

Thirtyeight

According to the StreamGroup API merge<T>(Iterable<Stream<T>> streams) → Stream<T>, this only works if all streams that you want to merge have the same type T. Therefore, u cannot merge two streams with completely different types, for example Stream<A> and Stream<B> – Staffordshire

Exactly this is what I said – Thirtyeight

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.

Arica

Answer 5

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');
    },
  );
}

Unreality

Answer 6

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

Allsopp