RxJava; 如何同步发出可观察量

Question

我想同步发出两个Observable对象(它们是异步的),一个接一个地返回第一个发出的Observable对象.如果第一个失败,则不应发出第二个.

假设我们有一个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"开头并行发出/调用两个observable.

有人遇到类似的用例吗？或者有一个关于如何让observables以同步的方式等待彼此的想法,在那里可以返回第一个发出的？

提前致谢.

Answer 1

您还可以使用rx.observables.BlockingObservable,例如:

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

Answer 2

您可以使用Single.blockingGet进行同步调用

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

Answer 3

在反应式编程中没有“等待”这样的术语。您需要考虑创建一个数据流，其中一个Observable可以被另一个触发。在您的情况下，收到后token您需要收到account. 它可能看起来像这样：

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