随时接受更多 observables 的 rxjava -百家乐凯发k8

问题描述

我目前正在使用 rx-java 2，并且有一个用例，即单个 camel route 订阅者需要使用多个 observable.使用此百家乐凯发k8的解决方案作为参考，我有一个部分可行的百家乐凯发k8的解决方案.rxjava - 随时接受更多 observable 的合并 observable?

i'm currently using rx-java 2 and have a use case where multiple observables need to be consumed by single camel route subscriber. using this solution as a reference, i have a partly working solution. rxjava - merged observable that accepts more observables at any time?

我打算使用 publishprocessor 订阅一个骆驼反应流订阅者，然后维护一个 concurrenthashset>我可以在其中动态添加新的 observable.
我目前不知道如何使用 publishprocessor 添加/管理 flowable 实例?我对 rx java 真的很陌生，所以任何帮助表示赞赏！这是我到目前为止所拥有的:

i'm planning to use a publishprocessor that will be subscribed to one camel reactive stream subscriber and then maintain a concurrenthashset> where i can dynamically add new observable.
i'm currently stuck on how can i add/manage flowable instances with publishprocessor? i'm really new to rx java, so any help is appreciated! this is what i have so far :

publishprocessor publishprocessor = publishprocessor.create();
camelreactivestreamsservice camelreactivestreamsservice = 
camelreactivestreams.get(camelcontext);
subscriber subscriber = 
     camelreactivestreamsservice.streamsubscriber("t-class",t.class);
}
set> flowableset = collections.newsetfrommap(new concurrenthashmap, boolean>());
public void add(flowable flowableorder){
    flowableset.add(flowableorder);
}
public void subscribe(){
    publishprocessor.flatmap(x -> flowableset.foreach(// todo)
    }) .subscribe(subscriber);
}

推荐答案

您可以拥有一个 processor 并订阅多个可观察流.您需要在添加和删除 observable 时通过添加和删除订阅来管理订阅.

you can have a single processor and subscribe to more than one observable stream. you would need to manage the subscriptions by adding and removing them as you add and remove observables.

publishprocessor publishprocessor = publishprocessor.create();
map, disposable> subscriptions = new concurrenthashmap<>();
void addobservable( flowable flowable ) {
  subscriptions.computeifabsent( flowable, fkey -> 
    flowable.subscribe( publishprocessor ) );
}
void removeobservable( flowable flowable ) {
  disposable d = subscriptions.remove( flowable );
  if ( d != null ) {
    d.dispose();
  }
}
void close() {
  for ( disposable d: subscriptions.values() ) {
    d.dispose();
  }
}

使用 flowable 作为地图的键，并添加或删除订阅.

use the flowable as the key to the map, and add or remove subscriptions.