RxJava通过自动连接处理可连接的可观察对象

Question

我有一个上游，它以块的形式发出数据。这个流应该使用slttleFirst进行限制。此外，在所有油门计时器完成后，应该发出最后一个值。不幸的是，RxJava 2中没有slttleFierstBu的运算符，因此我实现了一个观察变压器：

upstream -> {
      Observable<T> autoConnectingUpstream =
          upstream //
              .publish()
              .autoConnect(2);

      return Observable.merge(
              autoConnectingUpstream.throttleFirst(message.updatelimit().get(), TimeUnit.MILLISECONDS, scheduler),
              autoConnectingUpstream.debounce(message.updatelimit().get(), TimeUnit.MILLISECONDS, scheduler))
          //if debounce and throttle emit the same item
          .distinctUntilChanged();
    }

除了处理之外，它工作得很好。在处理完生成的可观察对象后，我也想处理上游。我该怎么做？

我尝试使用autoConnect（2，一次性）访问一次性-

Observable.<T>create(
            emitter -> {
              Observable<T> autoConnectingUpstream =
                  upstream //
                      .publish()
                      .autoConnect(2, emitter::setDisposable);

              Observable.merge(
                      autoConnectingUpstream.throttleFirst(message.updatelimit().get(), TimeUnit.MILLISECONDS, scheduler),
                      autoConnectingUpstream.debounce(message.updatelimit().get(), TimeUnit.MILLISECONDS, scheduler))
                  //if debounce and throttle emit the same item
                  .distinctUntilChanged()
                  .subscribe(emitter::onNext, emitter::onError, emitter::onComplete);
            });

Answer 1

据我所知，我在这里回答我自己的问题，如果我错了，请告诉我。

根据akarnokd的评论，解决方案如下所示：

Observable.<T>create(
        emitter -> {
          Observable<T> autoConnectingUpstream =
              upstream //
                  .publish()
                  .autoConnect(2, emitter::setDisposable);

          Observable.merge(
                  autoConnectingUpstream.throttleFirst(message.updatelimit().get(), TimeUnit.MILLISECONDS, scheduler),
                  autoConnectingUpstream.debounce(message.updatelimit().get(), TimeUnit.MILLISECONDS, scheduler))
              //if debounce and throttle emit the same item
              .distinctUntilChanged()
              .subscribe(emitter::onNext, emitter::onError, emitter::onComplete);
        });

自动连接的第二个参数是一个动作，它表示由两个连接的观察者建立的连接。

这可以与emitter：：SetDisposal一起使用，以便在观察者处理生成的可观察对象时处理自动连接。