只有在订阅时才会发出的可观察的

Question

observable = Observable.defer(new Callable<ObservableSource<Long>>() {
        @Override
        public ObservableSource<Long> call() throws Exception {
            final AtomicInteger counter = new AtomicInteger();

            return Observable.create(new ObservableOnSubscribe<Long>() {

                @Override
                public void subscribe(ObservableEmitter<Long> e) throws Exception {
                    emitter = e;
                }
            }).doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(Disposable disposable) throws Exception {
                    counter.incrementAndGet();
                    startEmitting(emitter);
                }
            }).doOnDispose(new Action() {
                @Override
                public void run() throws Exception {
                    if (counter.decrementAndGet() == 0) {
                        stopEmitting(emitter);
                    }
                }
            });
        }
    });

有谁能推荐一个更好的解决方案吗？

Answer 1

我对rx-java2了解不多，但我有一些关于可观察模式的建议。

您可以创建一个名为ObservableEmitter的对象。在这个类中，您可以创建一个方法，比如这个subscribe（Subscriber Subscriber）和一个方法emit（）。

然后由您的订户实现订户接口。我将调用方法receive（Message Message）。

public interface Subscriber {
  void receive(Message msg);
}

public class ObservableEmitter {
 private List<Subscriber> subscribers = new ArrayList<Subscriber>();
 public subscribe(Subscriber sub) {
  subscribers.add(sub);
 }
 public void emit(Message msg) {
   for(Subscriber sub : subscribers) {
     sub.receive(msg);
   }
 }
}