RxJava；如何同步发射可见光

Question

我想同步发射两个可观测对象（它们是异步的），一个接一个地返回第一个发射的可观测对象。如果第一个失败，则不应发出第二个。

假设我们有一个Observable可以让用户登录，另一个Observable可以在登录后自动选择用户的帐户。

这就是我所尝试的：

public Observable<AccessToken> signInAndSelectAccount(String username, String password)
{

    Observable<AccessToken> ob1 = ...; // Sign in.
    Observable<Account> ob2 = ...; // Select account.


    return Observable.zip(
            ob1,
            ob2,
            new Func2<AccessToken, Account, AccessToken>() {
                @Override
                public AccessToken call(AccessToken accessToken, Account account)
                {
                     return accessToken;
                }
            });
}

不幸的是，这不适用于我的用例。它将发出/调用两个平行的观测值，从“ob1”开始。

是否有人遇到过类似的用例？或者有没有一个想法，如何让可观测对象以同步的方式等待彼此，在那里可以返回第一次发射的数据？

提前感谢。

Answer 1

在反应式编程中没有“等待”这样的术语。您需要考虑创建一个数据流，其中一个可观察的数据流可能会被另一个数据流触发。在您的情况下，在收到令牌后，您需要接收帐户。可能是这样的：

Observable<Account> accountObservable = Observable.create(new Observable.OnSubscribe<AccessToken>() {
    @Override public void call(Subscriber<? super AccessToken> subscriber) {
        subscriber.onNext(new AccessToken());
        subscriber.onCompleted();
    }
}).flatMap(accessToken -> Observable.create(new Observable.OnSubscribe<Account>() {
    @Override public void call(Subscriber<? super Account> subscriber) {
        subscriber.onNext(new Account(accessToken));
        subscriber.onCompleted();
    }
}));

Answer 2

您可以使用Single。阻止同步调用的Get

// example 
signIn(name,password).blockingGet() 

Answer 3

您还可以使用rx.observables.Blocking观察，例如：

BlockingObservable.from(/**/).single();