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

Question

我有一个使用依赖于 RX Java 2的FileStack 依赖项io.reactivex.rxjava2:rxjava:2.1.2的 Android 项目。具体来说， . 直到现在这还不是真正的问题，因为我一直无法弄清楚如何具体取消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();}我需要做什么/调用uploadFlowable 以便在用户通过单击按钮取消上传时我可以手动触发取消？我看到有人建议调用dispose，但在检查可用于 Flowable 的可用方法时我没有看到该选项。

Answer 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的帮助/留言