缓冲区为空时如何停止PublishSubject的发射

Question

我正在尝试使用RxJava创建一个顺序下载服务。用户可以批量添加项目（20、30等）或单个项目。这些项目将被添加到队列中，然后以10个批次的顺序下载。为此，我使用PublishSubject：

PublishSubject<Int> pubSubject = PublishSubject.create();

它发出用户添加的项（ID），然后将缓冲区操作符应用于批处理项。使用这些ID，可以在flatMap中下载项目，并在订阅的onNext中返回。

  pubSubject.buffer(1, TimeUnit.SECONDS, 10)
            .observeOn(Schedulers.io())
            .flatMap { idsBatch -> downloadByIds(idsBatch) }
            .subscribe(
                /* onNext */ { apiResponse -> handleResponse() },
                /* onError */ { handleError(it) },
                /* onComplete*/ { hideProgressBar() }
             )

代码大部分按预期工作。项目已成功批处理并下载，但即使在发出所有项目后，buffer仍会使用空列表调用flatMap，并且永远不会调用onComplete（）。

我想知道当缓冲区中没有更多项目时，RxJava中是否有任何方法或方法可以获得on完成回调。因为否则我的下载服务永远不会终止。

Answer 1

您可以使用takeWhile操作：

返回一个可观察的，只要每个项目满足指定条件，它就会发出源可观察的发出的项目，然后在不满足此条件时立即完成。

pubSubject.buffer(1, TimeUnit.SECONDS, 10)
          .observeOn(Schedulers.io())
          .takeWhile { idsBatch -> idsBatch.isNotEmpty() }
          .flatMap { idsBatch -> downloadByIds(idsBatch) }
          .subscribe(
              /* onNext */ { apiResponse -> handleResponse() },
              /* onError */ { handleError(it) },
              /* onComplete*/ { hideProgressBar() }
           )