Dart 流阻塞答案

【问题标题】：Dart Streams blockingDart 流阻塞
【发布时间】：2019-03-28 00:20:34
【问题描述】：

我正在为 Futter 应用程序使用 BLoC 模式，但我无法让两个侦听器都工作。如果我在register.listen() 之前调用loggedIn.listen()，register.listen() 将不再起作用，反之亦然。

AuthBloc({this.authService}) {
    Stream<CrendentialsDetail> detailStream = Observable.combineLatest2(
      emailStream,
      passwordStream,
      (email, password) {
        return CrendentialsDetail(email: email, password: password);
      },
    );

    Stream<AuthResponse> loggedIn ... code ...

    Stream<RegisterResponse> register ... code ...

    register.listen((RegisterResponse res) {
      Scaffold.of(_context).showSnackBar(
        new SnackBar(
          content: new Text(res.message),
        ),
      );
    }, onError: (error) {
      Scaffold.of(_context).showSnackBar(
        new SnackBar(
          content: new Text(error.message),
        ),
      );
    });

    loggedIn.listen((AuthResponse user) {
      currentUserSubject.add(user);
      _goToHome();
    }, onError: (error) {
      Scaffold.of(_context).showSnackBar(
        new SnackBar(
          content: new Text(error.message),
        ),
      );
    });
  }

任何帮助将不胜感激！

【问题讨论】：

正如所写的那样，没有什么可以阻止两个侦听器被调用。也许两个听众的副作用隐藏了其他听众的影响？您是否尝试过记录以查看他们是否都受到了打击？否则，错误可能存在于代码的其他细节中。你能提供一个最小的复制吗？
我尝试并更改了代码，以便在 Streams 上直接调用 .listen()，并将 detailStream 对象实例化为参数。它工作。不知道为什么。
不清楚“不起作用”是什么意思。您是否尝试在单一订阅流上收听两次？如果是这样，第二次听应该抛出。我不确定 RxDart 中的某些东西是否会吞下该异常或类似的东西。总而言之，这取决于您在此处未显示的许多细节。如果您可以缩小到最小的重现，我们也许可以进行诊断。

标签： stream dart

【解决方案1】：

这是我现在正在使用的解决方案。有人可以向我解释为什么有效而我的问题无效吗？

AuthBloc({this.authService}) {
    Observable(loginSubject.stream).withLatestFrom(
        Observable.combineLatest2(
          emailStream,
          passwordStream,
          (email, password) {
            return CrendentialsDetail(email: email, password: password);
          },
        ), (_, d) {
      return d;
    }).flatMap((d) {
      return Observable.fromFuture(authService.loginUser(d)).doOnListen(() {
        loadingSubject.add(true);
      }).doOnDone(() {
        loadingSubject.add(false);
      });
    }).listen((AuthResponse user) {
      currentUserSubject.add(user);
      _goToHome();
    }, onError: (error) {
      Scaffold.of(_context).showSnackBar(
        new SnackBar(
          content: new Text(error.message),
        ),
      );
    });

    Observable(signUpSubject.stream).withLatestFrom(
        Observable.combineLatest2(
          emailStream,
          passwordStream,
          (email, password) {
            return CrendentialsDetail(email: email, password: password);
          },
        ), (_, d) {
      return d;
    }).flatMap((d) {
      return Observable.fromFuture(authService.register(d)).doOnListen(() {
        loadingSubject.add(true);
      }).doOnDone(() {
        loadingSubject.add(false);
      });
    }).listen((RegisterResponse res) {
      Scaffold.of(_context).showSnackBar(
        new SnackBar(
          content: new Text(res.message),
        ),
      );
    }, onError: (error) {
      Scaffold.of(_context).showSnackBar(
        new SnackBar(
          content: new Text(error.message),
        ),
      );
    });
  }

【讨论】：