RxJava2：未达到订阅

Question

我创建了这个订阅，它创建了另一个订阅：

this.loggingInterval$
    .filter(interval -> interval > 0)
    .doAfterNext(
        interval ->
        {
            Observable.interval(interval, TimeUnit.SECONDS)
                .doOnNext(t -> this.readAdditionalConfigurationFile())
                .doOnError(t -> LoggerFactory.getLogger(this.getClass()).error(t.toString()))
                .takeUntil(this.destroy$)
                .subscribe();
        }
    )
    .takeUntil(this.destroy$)
    .subscribe();

因此，每次interval 更改并且大于 0 时，我都会在每个新的 interval 秒内构建另一个订阅读取配置文件。当destroy$ 发出true 时，它会被销毁。

问题是永远无法到达interval.doOnNext()。

this.destroy$ = BehaviorSubject.createDefault(false);
this.loggingInterval$ = BehaviorSubject.createDefault(10);

有什么想法吗？

Answer 1

在 observable 中创建订阅是一种反模式；有很多更简单的方法。

this
.loggingInterval$
.filter(interval -> interval > 0)
.distinctUntilChanged()
.switchMap(interval -> Observable.interval(interval, TimeUnit.SECONDS, Schedulers.io()))
.doOnNext(dummy -> this.readAdditionalConfigurationFile())
.doOnError(t -> LoggerFactory.getLogger(this.getClass()).error(t.toString()))
.takeUntil(this.destroy$)
.subscribe();

如果订阅被取消订阅，上述操作还具有终止整个链的好处（这是你的 destroy$ 应该做的，对吧？）。