flutter - RXDART CombinelastStream函数不起作用

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

获取所有数据库帐户流组是

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

删除数据库帐户组为

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

组合器结果类为

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

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

它的地雷在Android Studio上运行日志。

I/flutter (28323): accounts getting

I/flutter (28323): hasData : false

I/flutter (28323): hasData : false

I/flutter (28323): accounts get

流工作，但我没有得到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,
  });
}

该代码未经测试，因此可能有错别字或我想念的东西，但是您应该大致了解。