Split Rx Observable dans de multiples flux et processus individuellement

Voici une photo de ce que je cherche à accomplir.

--a-b-c-a-bbb--un

divisé en

--a-----a-------a --> un flux

----b------bbb--- --> b flux

------c---------- --> c flux

Puis, être en mesure de

a.subscribe()
b.subscribe()
c.subscribe()

Jusqu'à présent, tout ce que j'ai trouvé a diviser le flux à l'aide d'un groupBy(), mais s'est ensuite effondrée tout remettre en un seul flux de données et de les traiter tous de la même fonction. Ce que je veux faire est de traiter chaque dérivés flux d'une manière différente.

La façon dont je suis en train de faire maintenant est de faire un tas de filtres. Est-il une meilleure façon de le faire?

InformationsquelleAutor Brandon Bil | 2015-03-04
  1. 15

    Vous n'avez pas à l'effondrement Observables de groupBy. Vous pouvez au lieu de s'y abonner.

    Quelque chose comme ceci:

        String[] inputs= {"a", "b", "c", "a", "b", "b", "b", "a"};

    Action1<String> a = s -> System.out.print("-a-");

    Action1<String> b = s -> System.out.print("-b-");

    Action1<String> c = s -> System.out.print("-c-");

    Observable
            .from(inputs)
            .groupBy(s -> s)
            .subscribe((g) -> {
                if ("a".equals(g.getKey())) {
                    g.subscribe(a);
                }

                if ("b".equals(g.getKey())) {
                    g.subscribe(b);
                }

                if ("c".equals(g.getKey())) {
                    g.subscribe(c);
                }
            });

    Si les états look un peu laid mais au moins, vous pouvez gérer chaque catégorie séparément. Peut-être il ya un moyen de les éviter.

    • Ouais, je voudrais éviter d'avoir les fi peut-être. Cependant, si il fonctionne, alors il va chercher un peu plus propre depuis ses tous dans un seul endroit, plutôt que de faire des filtres sur les flux d'origine. Merci!
    • A travaillé comme un charme!
    • Cool! Je vais mettre à jour ma réponse si je figure sur la façon de se débarrasser de if consolidés.
    • Vous pouvez utiliser un Dictionary<groupKey,action>, et votre groupe subscribe méthode aurait pour résoudre l'action du dictionnaire et de l'appeler.
    • Quelque chose comme Dictionary<String, Action1<String>> handlers = new Hashtable<>(); handlers.put("a", s -> System.out.println("-a-"));.... et .subscribe((g) -> { Action1<String> action = handlers.get(g.getKey()); g.subscribe(action); });
    • Brandon projet de Loi, j'ai dû relire votre question pour comprendre que vous étiez à l'origine à l'aide de filter. Par curiosité, ce qui vous fait préférer cette groupBy solution à la filter un? Pour moi filter semble plus simple, plus facile à comprendre, sans doute plus performants. Quels avantages voyez-vous dans groupBy?
    • J'ai remarqué certaines choses intéressantes pendant le débogage (remarque c'est à l'aide de Rx JS), là où un "groupe" a été associé à un gestionnaire, un autre exemple d'un élément du groupe ne serait pas aller à travers les grands s'abonner fonction indiquée ci-dessus. Au lieu de cela, il transmettra directement à l'Action, le groupe a souscrit. Alors qu'avec un filtre, les comparaisons se font pour chaque valeur de la source de l'observable. Donc, avec N divise, une possible N comparaisons seront faites pour chaque élément. Avec des groupes, des comparaisons sont effectuées uniquement lorsqu'un nouveau groupe est trouvé.

    InformationsquelleAutor ihuk
  2. 37

    Facile comme bonjour, il suffit d'utiliser filter

    Un exemple en scala

    import rx.lang.scala.Observable

val o: Observable[String] = Observable.just("a", "b", "c", "a", "b", "b", "b", "a")
val hotO: Observable[String] = o.share
val aSource: Observable[String] = hotO.filter(x ⇒ x == "a")
val bSource: Observable[String] = hotO.filter(x ⇒ x == "b")
val cSource: Observable[String] = hotO.filter(x ⇒ x == "c")

aSource.subscribe(o ⇒ println("A: " + o), println, () ⇒ println("A Completed"))

bSource.subscribe(o ⇒ println("B: " + o), println, () ⇒ println("B Completed"))

cSource.subscribe(o ⇒ println("C: " + o), println, () ⇒ println("C Completed"))

    Vous avez juste besoin de s'assurer que la source observable est chaud. Le plus simple est de share il.

    • Que faire si vous voulez ou initiale observables à froid?
    • utilisez simplement publish au lieu de share et invoquer connect lorsque tous subsribers sont abonnés.
    • Upvoted en raison de commentaire de @Krzysztof Skyrzynecki
    • Il n'y a pas de point d'en faire chaud à l'aide de share. Le code fourni effectivement souscrit à trois reprises pour chaque abonné, même comme il le ferait sans share. La bonne façon de procéder est décrite dans Krzysztof Skyrzynecki commentaire: utiliser publish au lieu de share et invoquer connect lorsque tous subsribers sont abonnés.
    • Je suis d'accord que le publish+connect méthode est plus propre et je voudrais l'utiliser chaque fois que possible (mais parfois, vous ne savez pas les abonnés à l'avance et ne se soucient pas que vous manquez de certains éléments). Cependant, je ne pense pas que votre déclaration au sujet de trois abonnements à l'origine de froid o est correct. Si vous pouvez prouver le contraire, veuillez share() 🙂 votre preuve avec nous.
    InformationsquelleAutor Tomáš Dvořák