I have the following code structure
Service
public Flowable entryFlow()
{
return Flowable.fromIterable(this::getEntries)
}
消费者
void start()
{
disposable = service
.entryFlow()
.observeOn(Schedulers.computation())
.subscribeOn(Schedulers.computation())
.subscribe(
entry -> ...,
this::onError,
this::subscriptionFinished);
}
void stop()
{
disposable.dispose();
}
private void onError(Throwable e)
{
subscriptionFinished();
}
private void subscriptionFinished()
{
//
}
当调用stop方法时,我需要一种停止flowable获取和发射数据的方法。
通过以下操作,我注意到doOnCancel lambda并不总是被调用。
void start()
{
disposable = service
.entryFlow()
.observeOn(Schedulers.computation())
.subscribeOn(Schedulers.computation())
.doOnCancel(this::snapshotFinished)
.subscribe(
entry -> ...,
this::onError,
this::subscriptionFinished);
}
void stop()
{
disposable.dispose();
}
替代方案可以是:
volatile stopped;
void start()
{
disposable = service
.entryFlow()
.observeOn(Schedulers.computation())
.subscribeOn(Schedulers.computation())
.takeUntil(x -> stopped)
.subscribe(
entry -> ...,
this::onError,
this::subscriptionFinished);
}
void stop()
{
stopped = true;
}
什么样的开始和结束实现方式被推荐,以便让Flowable停止发出事件并调用onComplete或类似方法(例如doOnCancel操作)?
稍后编辑:
为了让我的用户案例更简短,
只需要调用disposable.dispose来停止从iterable获取数据并将数据发送到source吗?我只有一个订阅者,并且需要在Flowable结束时无论原因如何都调用onComplete/onError/其他回调。
其他回调指的是doOnCancel/doFinally等之一。
谢谢
doOnCancel
不被调用的唯一原因是流已经在那个时候完成了。为什么你只需要对取消做出反应?你可以考虑使用doFinally
来对终止和取消做出反应。 - akarnokd