RxJava; Comment émettre des observables de façon synchrone

Je veux synchrone émettent deux objets observés (qui sont asynchrones), l'un après l'autre où il renvoie la première émis l'objet observé. Si le premier tombe en panne, il ne devrait pas émettre le second.

Disons que nous avons un Observables que les signes d'un utilisateur, et un autre Observable qui sélectionne automatiquement le compte de l'utilisateur, après signature.

C'est ce que j'ai essayé:

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

Cela ne fonctionne malheureusement pas pour mon cas d'utilisation. Il émettra/appel à la fois observables parallèle, en commençant avec 'ob1'.

Quelqu'un a rencontrer un semblable cas d'utilisation? Ou a une idée sur comment faire des observables attendre eachother en mode synchrone, dans lequel le premier émis peuvent être retournés?

Merci d'avance.

OriginalL'auteur Thomas Neuteboom | 2016-01-28

  1. 4

    Il n'y a pas une telle terme "attendre" dans la programmation réactive. Vous avez besoin de réfléchir à la création d'un flux de données, où l'on Observable pourrait être déclenché par un autre. Dans votre cas, après réception de token vous avez besoin pour recevoir account. Il pourrait ressembler à ceci:

    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();
    }
}));
    Ce n'est pas correct. S'il est vrai que vous devriez éviter de bloquer les appels dans la plupart des cas, RxJava 1 fournit BlockingObservable et RxJava 2 fournit plusieurs .blocking*() méthodes de l'interface avec le code existant qui est non réactif.

    OriginalL'auteur eleven

  2. 6

    Vous pouvez également utiliser des rx.observables.BlockingObservable par exemple:

    BlockingObservable.from(/**/).single();
    BlockingSingle.from(/**/) est également une option, en particulier pour Single.

    OriginalL'auteur netgui

  3. 1

    Je ne sais pas Java, mais la solution à Scala serait probablement ce, j'espère que c'est lisible pour vous

    import rx.lang.scala.Observable

class AccessToken
class Account

case class TokenAndAccount(token: AccessToken, account: Account)

val accessTokenSource = Observable.just(new AccessToken)
val accountSource = Observable.just(new Account)

accessTokenSource
   .flatMap(token ⇒ accountSource.map(account ⇒ TokenAndAccount(token, account)))
   .subscribe(tokenAndAccount ⇒ println(tokenAndAccount))

    Fondamentalement flatMap assurez-vous que la accountSource.map... est utilisé seulement après que le jeton de accessTokenSource a été émise. À l'intérieur de la accountSource.map nous combinons le jeton de compte pour une utilisation ultérieure dans subscribe.

    flatMap est l'un des opérateurs les plus utiles, assurez-vous de le lire c'est docs et peut-être quelques tutoriels.

    Salut Tomáš, merci pour cette. Je vais le regarder!

    OriginalL'auteur Tomáš Dvořák

  4. 1

    Vous pouvez utiliser Unique.blockingGet pour appel synchrone

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

    OriginalL'auteur Phan Van Linh