Не работает функция Rxdart comblaststream

Question

Я собираюсь объединить два потока. Но это не работает. В чем моя ошибка?

Моя функция сборки:

  @override
  Widget build(BuildContext context) {
    return StreamBuilder(
      stream: Observable.combineLatest2(
        getAllDBAccountsBloc.getAllDBAccountsStream,
        deleteDBAccountBloc.deleteDBAccountStream,
        (accountList, deleteAccountResultModel) {
          print("my account list : ${accountList == null}");
          return AccountsCombinerResult(
            deleteAccountResultBlocModel: deleteAccountResultModel,
            accountsList: accountList,
          );
        },
      ),
      builder: (context, snapshot) {
        print("hasData : ${snapshot.hasData}");
        if (snapshot.hasData) accountsCombinerResult = snapshot.data;
        if (snapshot.hasError) return Text(snapshot.error.toString());
        return _buildWidget;
      },
    );
  }

Получить все учетные записи БД Stream Blo c is

class GetAllDBAccountsBloc {
  final _getAllDBAccountsFetcher = PublishSubject<List<AccountDatabaseModel>>();

  Observable<List<AccountDatabaseModel>> get getAllDBAccountsStream => _getAllDBAccountsFetcher.stream;

  getAllDBAccounts() async {
    print("accounts getting");
    _getAllDBAccountsFetcher.sink.add(null);
    await new Future.delayed(const Duration(seconds: 1));
    _getAllDBAccountsFetcher.sink.add(await Repository.getAllDBAccounts());
    print("accounts get");
  }

  dispose() {
    _getAllDBAccountsFetcher.close();
  }
}

final getAllDBAccountsBloc = GetAllDBAccountsBloc();

Удалить учетную запись БД Blo c is

class DeleteDBAccountBloc {
  final _deleteDBAccountFetcher = PublishSubject<DeleteAccountResultBlocModel>();

  Observable<DeleteAccountResultBlocModel> get deleteDBAccountStream => _deleteDBAccountFetcher.stream;

  deleteDBAccount(DeleteAccountRequestBlocModel requestModel) async {
    _deleteDBAccountFetcher.sink.add(DeleteAccountResultBlocModel());
    await new Future.delayed(const Duration(seconds: 1));
    _deleteDBAccountFetcher.sink.add(await Repository.deleteDBAccount(requestModel));
  }

  dispose() {
    _deleteDBAccountFetcher.close();
  }
}

final deleteDBAccountBloc = DeleteDBAccountBloc();

Класс результата Combiner равен

class AccountsCombinerResult {
  final DeleteAccountResultBlocModel deleteAccountResultBlocModel;
  final List<AccountDatabaseModel> accountsList;

  AccountsCombinerResult({
    @required this.accountsList,
    @required this.deleteAccountResultBlocModel,
  });
}

его шахта Запустить журнал android studio ..

I / флаттер (28323): учетные записи получают

I / флаттер (28323): hasData: false

I / flutter (28323): hasData: false

I / flutter (28323): учетные записи получают

Поток работает, но я не получил данные результатов AccountsCombiner.

Этот метод сборки работает, но я не хочу его использовать ...

  @override
  Widget build(BuildContext context) {
    return StreamBuilder(
      stream: getAllDBAccountsBloc.getAllDBAccountsStream,
      builder: (context, getDbAccountsSnapshot) {
        return StreamBuilder(
          stream: deleteDBAccountBloc.deleteDBAccountStream,
          builder: (context, deleteDbAccountStreamSnapshot) {
            if (deleteDbAccountStreamSnapshot.hasData && getDbAccountsSnapshot.hasData) {
              print("qweqweq");
              accountsCombinerResult = AccountsCombinerResult(
                accountsList: getDbAccountsSnapshot.data,
                deleteAccountResultBlocModel: deleteDbAccountStreamSnapshot.data,
              );
            }
            if (getDbAccountsSnapshot.hasError) return Text(getDbAccountsSnapshot.error.toString());
            if (deleteDbAccountStreamSnapshot.hasError) return Text(deleteDbAccountStreamSnapshot.error.toString());
            return _buildWidget;
          },
        );
      },
    );
  }

Answer 1

Вы создаете новый поток каждый раз, когда вызывается метод build. Вам нужно сохранить ссылку на поток в состоянии.

StreamController<AccountsCombinerResult> _streamController = StreamController<AccountsCombinerResult>();

@override
void initState() {
    super.initState();
    _streamController.addStream(Observable.combineLatest2(
        getAllDBAccountsBloc.getAllDBAccountsStream,
        deleteDBAccountBloc.deleteDBAccountStream,
        (accountList, deleteAccountResultModel) {
          print("my account list : ${accountList == null}");
          return AccountsCombinerResult(
            deleteAccountResultBlocModel: deleteAccountResultModel,
            accountsList: accountList,
          );
        },
      ));
}

@override
void dispose() {
    super.dispose();
    _streamController.close();
}

@override
Widget build(BuildContext context) {
    return StreamBuilder(
      stream: _streamController.stream,
      builder: (context, snapshot) {
        print("hasData : ${snapshot.hasData}");
        if (snapshot.hasData) accountsCombinerResult = snapshot.data;
        if (snapshot.hasError) return Text(snapshot.error.toString());
        return _buildWidget;
      },
    );
}

Чтобы упростить эту задачу, вы можете использовать StreamProvider из пакета поставщика. https://pub.dev/packages/provider https://pub.dev/documentation/provider/latest/provider/StreamProvider-class.html

Он создает поток только один раз.

@override
Widget build(BuildContext context) {
    return StreamProvider<AccountsCombinerResult>(
      initialData: null, // not sure if this works, you can try []
      create: () => Observable.combineLatest2(
        getAllDBAccountsBloc.getAllDBAccountsStream,
        deleteDBAccountBloc.deleteDBAccountStream,
        (accountList, deleteAccountResultModel) {
          print("my account list : ${accountList == null}");
          return AccountsCombinerResult(
            deleteAccountResultBlocModel: deleteAccountResultModel,
            accountsList: accountList,
          );
        },
      ),
      catchError: (context, error) => AccountsCombinerResult(
          deleteAccountResultBlocModel: null,
          accountsList: null,
          error: error,
      ), 
      child: Builder(
        builder: (context) {
            final data = Provider.of<AccountsCombinerResult>(context);
            // maybe null check
            if (data.error != null) return Text(data.error.toString());
            accountsCombinerResult =data;
            return _buildWidget;
         },
      ),
    );
}

class AccountsCombinerResult {
  final DeleteAccountResultBlocModel deleteAccountResultBlocModel;
  final List<AccountDatabaseModel> accountsList;
  final dynamic error;

  AccountsCombinerResult({
    @required this.accountsList,
    @required this.deleteAccountResultBlocModel,
    this.error,
  });
}

Код не проверен, поэтому Может быть, опечатки или вещи, которые я пропустил, но вы должны получить общее представление.