Obtenir la dernière valeur de l'Observable et l'émettre immeditely

Je vais essayer d'obtenir la dernière valeur d'une Observable et l'obtenir à émettre
immédiatement une fois qu'il est appelé. Étant donné le code ci-dessous à titre d'exemple:

return Observable.just(myObservable.last())
    .flatMap(myObservable1 -> {
        return myObservable1;
    })
    .map(o -> o.x) //Here I want to end up with a T object instead of Observable<T> object

Cela ne fonctionne pas, car en faisant cela, les flatMap émet myObservable1 qui aura
à émettre pour atteindre le map.
Je ne sais pas si une telle chose est encore possible. Faire quelqu'un a une idée sur la façon d'atteindre cet objectif? Merci

Ce observables parlez-vous? Qu'entendez-vous par plus tard?
myObservable est une chaude observables qui émettent par exemple: "1", "2", "3" à intervalles irréguliers. Ce que je veux faire est d'être en mesure d'obtenir myObservable dernière valeur ("3" dans notre cas) au moment où j'arrive à la return directive
Est myObservable chaud ou froid?
On dirait que vous vous êtes toujours penser en impératif(pas réactionnelle). Pourriez-vous nous donner une perspective de plus grande? Le Point est que, normalement, vous créez observables sur la base d'autres observables. Vous pouvez créer lastObservable basé sur myObservable, qui sera bloqué jusqu'à ce que myObservable se termine, puis émettent de la dernière valeur. Il n'y a pas moyen de sortir de ces flux de "tubes" à l'exception des points de terminaison comme vous abonner.
Une fois, j'ai posé une question similaire sur le RxJava Groupe Google - peut-être Ben des réponses utiles pour vous: groups.google.com/forum/#!searchin/rxjava/mihola/rxjava/...

OriginalL'auteur E-Kami | 2016-05-03

  1. 22

    last() méthode ne sera pas de toute l'aide ici car elle attend les Observables de résilier pour vous donner le dernier élément émis.

    En supposant que vous n'avez pas le contrôle sur l'émission observables vous pouvez créer simplement un BehaviorSubject et abonnez-vous à la observables qui émet les données que vous souhaitez écouter, puis de s'abonner à la création de l'objet. Depuis Subject est à la fois Observable et Subscriber vous obtiendrez ce que vous voulez.

    Je pense (ne pas avoir le temps de le vérifier maintenant), vous pourriez avoir à vous désinscrire manuellement à partir de l'original observables comme la BehaviorSubject une fois l'ensemble de ses abonnés se désabonner ne se désabonner automatiquement.

    Quelque chose comme ceci:

    BehaviorSubject subject = new BehaviorSubject();
hotObservable.subscribe(subject);
subject.subscribe(thing -> {
    //Here just after subscribing 
    //you will receive the last emitted item, if there was any.
    //You can also always supply the first item to the behavior subject
});

    http://reactivex.io/RxJava/javadoc/rx/subjects/BehaviorSubject.html

    OriginalL'auteur MatBos

  2. 1

    Dans RxJava, l'abonné.onXXX est appelé asynchrone.Cela signifie que si votre Observables émettent des éléments dans le nouveau thread, vous pouvez ne jamais obtenir le dernier élément avant de retourner, à l'exception de vous bloquer le thread et d'attendre pour l'élément.Mais si l'Observable émettent de l'élément de façon synchrone et vous n'avez pas " le changement c'est thread par subscribeOn et observOn,
    comme le code:

    Observable.just(1,2,3).subscribe();

    Dans ce cas, vous pouvez obtenir le dernier élément en faisant comme ceci:

    Integer getLast(Observable<Integer> o){
    final int[] ret = new int[1];
    Observable.last().subscribe(i -> ret[0] = i);
    return ret[0];
}

    C'est une mauvaise idée de faire comme ça.RxJava préférez vous faire travail asynchrone.

    ne travaillent la plupart du temps

    OriginalL'auteur dieyidezui

  3. 0

    Ce que vous voulez vraiment atteindre ici est de prendre une tâche asynchrone et de le transformer en synchrone.

    Il y a plusieurs façons d'y parvenir, chacun avec ses avantages et ses inconvénients:

    • Utilisation toBlocking() - cela signifie que ce thread va être BLOQUÉ, jusqu'à ce que le flux est fini, afin d'obtenir un seul élément de simplement utiliser (première) qu'il sera terminé une fois qu'un article est livré.
      disons que l'ensemble de votre flux est Observable<T> getData();
      ensuite, une méthode qui sera la dernière valeur immédiatement ressemblera à ceci:

    public T getLastItem(){
    return getData().toBlocking().first();
    }

    s'il vous plaît ne pas utiliser dernier (le) qu'il va attendre que le flux complet, et alors seulement, émettent le dernier élément.

    Si votre flux est un réseau à la demande et il n'a pas obtenu de point mais cela va bloquer votre fil!, donc, utiliser uniquement lorsque vous êtes sûr qu'il est un élément disponible immédiatement (ou si vous voulez vraiment un bloc...)

    • une autre option est de simplement mettre en cache le dernier résultat, quelque chose comme ceci:

      getData().abonnez-vous(t-> cachedT = t;) //quelque part dans le code et il va garder la sauvegarde de la dernière marchandise livrée
      T public getLastItem(){
      retour cachedT;
      }

    si il n'y avait pas tout article envoyé par le temps vous le demandez, vous obtiendrez la valeur null ou quelle que soit la valeur initiale que vous avez défini.
    le problème avec cette approche est que le abonnez-vous de phase peut se produire après l'obtenir et pourrait faire une course à condition, si elle est utilisée dans les 2 threads différents.

    OriginalL'auteur ndori