RxJava2 onSubscribe和onComplete

我正在学习RxJava2,我有一个关于如何正确使用防止内存泄漏的问题。

我需要确保每次调用onSubscribe()时都调用onComplete()吗?

Observables.combineLatest(
    siteDataController
        .GetCurrentSelectionObservable()
        .distinctUntilChanged { site -> site.GetStid() },
    uiController
        .GetDisplayFieldsObservable()
        .observeOn(AndroidSchedulers.mainThread())
) { site, data -> Pair(site, data) }
.takeWhile{continueUpdates.get()}
.subscribeOn(Schedulers.computation())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(object : Observer<Pair<SiteDataController.ProcessedSite, UIController.DisplayFields>> {
     override fun onComplete() {
         Log.e("Lifespan", "Site / UI onComplete()")
     }
     override fun onSubscribe(d: Disposable) {
          Log.e("Lifespan", "Site / UI onSubscribe()")
                    disposable = d
     }

     override fun onNext(t: Pair<SiteDataController.ProcessedSite, UIController.DisplayFields>) {
          // use result to build out ui
     }

     override fun onError(e: Throwable) {
         e.printStackTrace()
     }
})

它在片段的onResume()中设置。另外,“continueUpdates”在onResume()中设置为true,在onPause()中设置为false。组成combineLast()流的两个可见项的根都是behaviorSubject。但是,尽管takeWhile条件被切换为false,并且即使在两个behaviorSubject上调用onComplete(),观测者上的onComplete()永远不会被调用。我做错什么了吗?