如何手动取消/处置 RXJava2 Flowable？答案

【问题标题】：How do I manually cancel / dispose of an RXJava2 Flowable?如何手动取消/处置 RXJava2 Flowable？
【发布时间】：2019-10-17 12:14:05
【问题描述】：

我有一个 Android 项目正在使用依赖于 RX Java 2 的 FileStack dependency。具体来说，io.reactivex.rxjava2:rxjava:2.1.2。到目前为止，这并不是一个真正的问题，因为我无法弄清楚如何专门取消Flowable。

我已经实现了

下面是我的代码：

private Flowable<Progress<FileLink>> upload;

private void myMethod(){
    upload = new Client(myConfigOptionsHere)
      .uploadAsync(filePath, false, storageOptions);

        upload.doOnNext(progress -> {
                    double progressPercent = progress.getPercent();
                    if(progressPercent > 0){
                        //Updating progress here
                    }
                    if (progress.getData() != null) {
                        //Sending successful upload callback here
                    }
                })
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        //Logging he complete action here
                    }
                })
                .doOnCancel(new Action() {
                    @Override
                    public void run() throws Exception {
                        //Logging the cancel here
                    }
                })
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable t) throws Exception {
                        //Logging the error here
                    }
                })
                .subscribe();

}

public void cancelUpload(){
    //What do I do here to manually stop the upload Flowable? 
    //IE upload.cancel();
}

我需要做什么/调用 upload Flowable 以便在用户通过单击按钮取消上传时手动触发取消？我看到有人 recommending 调用dispose，但在检查可用于 Flowable 的可用方法时，我没有看到该选项。

【问题讨论】：

标签： java android rx-java2 filestack

【解决方案1】：

原来问题在于我试图处置/取消错误的对象。我将我的代码调整为以下内容：

private Disposable disposable;

private void myMethod(){
    Flowable<Progress<FileLink>> upload = new Client(myConfigOptionsHere)
      .uploadAsync(filePath, false, storageOptions);

        this.disposable = upload.doOnNext(progress -> {
                    double progressPercent = progress.getPercent();
                    if(progressPercent > 0){
                        //Updating progress here
                    }
                    if (progress.getData() != null) {
                        //Sending successful upload callback here
                    }
                })
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        //Logging he complete action here
                    }
                })
                .doOnCancel(new Action() {
                    @Override
                    public void run() throws Exception {
                        //Logging the cancel here
                    }
                })
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable t) throws Exception {
                        //Logging the error here
                    }
                })
                .subscribe();

}

public void cancelUpload(){
    if(this.disposable != null){
        this.disposable.dispose();
        this.disposable = null;
    }
}

并且能够让它正常工作。本质上，您需要针对dispose 对象调用dispose() 方法，而不是Flowable。

感谢您的帮助/留言jschuss

【讨论】：