可取消的任务

app/src/main/java/com/miekir/newmvp/ui/home/HomeActivity.java

@@ -34,7 +34,7 @@ public class HomeActivity extends BaseMvpActivity implements View.OnClickListene
                 break;
             case R.id.btn_task:
                 // 点击之后，后台执行耗时操作
-                mArticlePresenter.doProgressWork();
+                mArticlePresenter.doStickyWork();
                 break;
             default:
                 break;

app/src/main/java/com/miekir/newmvp/ui/home/article/presenter/SimpleArticlePresenter.java

@@ -2,13 +2,16 @@ package com.miekir.newmvp.ui.home.article.presenter;
 
 import com.miekir.common.utils.LogTool;
 import com.miekir.mvp.presenter.BasePresenter;
-import com.miekir.network.widget.consumer.LoadingObserver;
+import com.miekir.network.widget.observe.cancelable.CancelConsumer;
+import com.miekir.network.widget.observe.cancelable.common.CancelableObserver;
+import com.miekir.network.widget.observe.sticky.common.LoadingStickyObserver;
 import com.miekir.newmvp.ui.home.IArticleView;
 
 import io.reactivex.Observable;
 import io.reactivex.ObservableOnSubscribe;
 import io.reactivex.android.schedulers.AndroidSchedulers;
 import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.observers.LambdaObserver;
 import io.reactivex.schedulers.Schedulers;
 
 /**
@@ -53,8 +56,23 @@ public class SimpleArticlePresenter extends BasePresenter<IArticleView> {
 
     /**
      * 直接在Presenter执行任务
+     * 有进度，可以手动取消的任务
      */
     public void doProgressWork() {
+        LambdaObserver<String> disposable = new CancelableObserver<String>(this, new CancelConsumer<String>(this) {
+            @Override
+            public void onResult(String s) {
+                LogTool.d("MainPresenter", "onNext thread is: " + Thread.currentThread().getName());
+                post(view -> view.onArticleResult("result"));
+            }
+        }, new CancelConsumer<Throwable>(this) {
+            @Override
+            public void onResult(Throwable throwable) {
+                LogTool.d("MainPresenter", "onNext thread is: " + Thread.currentThread().getName());
+                post(view -> view.onArticleResult("result"));
+            }
+        }).getCancelableObserver();
+
         Observable.create((ObservableOnSubscribe<String>) emitter -> {
             try {
                 Thread.sleep(5000);
@@ -66,8 +84,27 @@ public class SimpleArticlePresenter extends BasePresenter<IArticleView> {
         })
         .subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread())
-        .subscribe(new LoadingObserver<String>(this) {
+        .subscribe(disposable);
 
+        mProgressDisposableList.add(disposable);
+    }
+
+    /**
+     * 直接在Presenter执行任务
+     */
+    public void doStickyWork() {
+        Observable.create((ObservableOnSubscribe<String>) emitter -> {
+            try {
+                Thread.sleep(5000);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            emitter.onNext("");
+            emitter.onComplete();
+        })
+        .subscribeOn(Schedulers.io())
+        .observeOn(AndroidSchedulers.mainThread())
+        .subscribe(new LoadingStickyObserver<String>(this) {
             @Override
             public void onSuccess(String result) {
                 post(view -> view.onArticleResult("result"));
@@ -75,7 +112,7 @@ public class SimpleArticlePresenter extends BasePresenter<IArticleView> {
 
             @Override
             public void onFailure(int code, Throwable e, String errMsg) {
-
+                post(view -> view.onArticleResult("result"));
             }
         });
     }

network/src/main/java/com/miekir/network/base/BaseResponse.java

@@ -4,7 +4,6 @@ package com.miekir.network.base;
  * 统一返回封装
  * @author Miekir
  */
-
 public class BaseResponse<T> {
     private int code;
     private String message;

network/src/main/java/com/miekir/network/widget/consumer/LoadingObserver.java

@@ -1,51 +0,0 @@
-package com.miekir.network.widget.consumer;
-
-import com.miekir.mvp.presenter.BasePresenter;
-import com.miekir.network.constant.Code;
-import com.miekir.network.utils.ExceptionUtil;
-import com.miekir.network.widget.observe.common.StickyObserver;
-
-import io.reactivex.annotations.NonNull;
-import io.reactivex.disposables.Disposable;
-
-/**
- * 进度条观察者
- * 需要进度条时，调用onSubscribe同时显示进度条，
- * 流程走到onComplete/onError时隐藏进度条。
- * @author Miekir
- * 变为发送通知增加dialog显示次数
- */
-
-public abstract class LoadingObserver<T> extends StickyObserver<T> {
-    private BasePresenter<?> mPresenter;
-
-    public LoadingObserver(BasePresenter<?> presenter) {
-        mPresenter = presenter;
-    }
-
-    @Override
-    public void onSubscribe(@NonNull Disposable d) {
-        if (d.isDisposed()) {
-            return;
-        }
-
-        mPresenter.showProgress();
-    }
-
-    @Override
-    public void onNext(T t) {
-        mPresenter.hideProgress();
-        onSuccess(t);
-    }
-
-    @Override
-    public void onComplete() {
-
-    }
-
-    @Override
-    public void onError(@NonNull Throwable e) {
-        mPresenter.hideProgress();
-        onFailure(Code.COMMON, e, ExceptionUtil.exceptionHandler(e));
-    }
-}

network/src/main/java/com/miekir/network/widget/observe/BackupObserver.java

@@ -4,7 +4,7 @@ import android.content.Context;
 
 import com.miekir.network.R;
 import com.miekir.network.widget.CustomProgressDialog;
-import com.miekir.network.widget.observe.net.NetStickyObserver;
+import com.miekir.network.widget.observe.sticky.net.NetStickyObserver;
 
 import io.reactivex.annotations.NonNull;
 import io.reactivex.disposables.Disposable;

network/src/main/java/com/miekir/network/widget/observe/cancelable/CancelConsumer.java

@@ -0,0 +1,29 @@
+package com.miekir.network.widget.observe.cancelable;
+
+import com.miekir.mvp.presenter.BasePresenter;
+
+import io.reactivex.functions.Consumer;
+
+/**
+ * Copyright (C), 2019-2020, Miekir
+ *
+ * @author Miekir
+ * @date 2020/11/21 17:40
+ * Description:
+ */
+public abstract class CancelConsumer<T> implements Consumer<T> {
+
+    private BasePresenter<?> mPresenter;
+
+    public CancelConsumer(BasePresenter<?> presenter) {
+        this.mPresenter = presenter;
+    }
+
+    @Override
+    public void accept(T t) throws Exception {
+        mPresenter.hideProgress();
+        onResult(t);
+    }
+
+    public abstract void onResult(T t);
+}

network/src/main/java/com/miekir/network/widget/observe/cancelable/common/CancelableObserver.java

@@ -0,0 +1,30 @@
+package com.miekir.network.widget.observe.cancelable.common;
+
+import com.miekir.mvp.presenter.BasePresenter;
+import com.miekir.network.widget.observe.cancelable.CancelConsumer;
+
+import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.observers.LambdaObserver;
+
+/**
+ * Copyright (C), 2019-2020, Miekir
+ *
+ * @author Miekir
+ * @date 2020/11/21 17:25
+ * Description:
+ * 可见可手动取消的任务
+ */
+public class CancelableObserver<T> {
+    private CancelConsumer<? super T> onNext;
+    private CancelConsumer<? super Throwable> onError;
+
+    public CancelableObserver(BasePresenter<?> presenter, CancelConsumer<? super T> onNext, CancelConsumer<? super Throwable> onError) {
+        presenter.showProgress();
+        this.onNext = onNext;
+        this.onError = onError;
+    }
+
+    public LambdaObserver<T> getCancelableObserver() {
+        return new LambdaObserver<T>(onNext, onError, Functions.EMPTY_ACTION, Functions.emptyConsumer());
+    }
+}

network/src/main/java/com/miekir/network/widget/observe/common/LoadingStickyObserver.java

@@ -1,4 +1,4 @@
-package com.miekir.network.widget.observe.common;
+package com.miekir.network.widget.observe.sticky.common;
 
 import com.miekir.mvp.presenter.BasePresenter;
 
@@ -11,7 +11,7 @@ import io.reactivex.disposables.Disposable;
  * 流程走到onComplete/onError时隐藏进度条。
  * @author Miekir
  * 变为发送通知增加dialog显示次数
- * 可见，但不可手动取消的网络任务
+ * 可见，但不可手动取消的任务
  */
 
 public abstract class LoadingStickyObserver<T> extends StickyObserver<T> {

network/src/main/java/com/miekir/network/widget/observe/common/StickyObserver.java

@@ -1,4 +1,4 @@
-package com.miekir.network.widget.observe.common;
+package com.miekir.network.widget.observe.sticky.common;
 
 import com.miekir.network.constant.Code;
 import com.miekir.network.utils.ExceptionUtil;
@@ -10,7 +10,7 @@ import io.reactivex.disposables.Disposable;
 /**
  * 返回结果处理
  * @author Miekir
- * 用于不可见，不可取消的任务
+ * 用于不可见，不可手动取消的任务
  */
 
 public abstract class StickyObserver<T> implements Observer<T> {

network/src/main/java/com/miekir/network/widget/observe/net/LoadingNetStickyObserver.java

@@ -1,4 +1,4 @@
-package com.miekir.network.widget.observe.net;
+package com.miekir.network.widget.observe.sticky.net;
 
 import com.miekir.mvp.presenter.BasePresenter;
 

network/src/main/java/com/miekir/network/widget/observe/net/NetStickyObserver.java

@@ -1,4 +1,4 @@
-package com.miekir.network.widget.observe.net;
+package com.miekir.network.widget.observe.sticky.net;
 
 import com.miekir.network.base.BaseResponse;
 import com.miekir.network.constant.Code;