Support rxjava2

Support rxjava2

nohttputils-r1/build.gradle → nohttputils-r2/build.gradle

@@ -8,8 +8,8 @@ android {
     defaultConfig {
         minSdkVersion 11
         targetSdkVersion 23
-        versionCode 1
-        versionName "1"
+        versionCode 2
+        versionName "2"
     }
     buildTypes {
         release {
@@ -24,7 +24,7 @@ dependencies {
     testCompile 'junit:junit:4.12'
     compile 'com.android.support:appcompat-v7:23.4.0'
     compile 'com.yanzhenjie.nohttp:okhttp:1.1.4'
-    compile 'io.reactivex:rxandroid:1.2.1'
-    compile 'io.reactivex:rxjava:1.1.9'
+    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
+    compile 'io.reactivex.rxjava2:rxjava:2.1.5'
     compile 'com.google.code.gson:gson:2.8.0'
 }

nohttputils-r1/src/main/java/com/liqi/nohttputils/nohttp/RxNoHttp.java → nohttputils-r2/src/main/java/com/liqi/nohttputils/nohttp/RxNoHttp.java

@@ -21,10 +21,14 @@
 import java.net.ConnectException;
 import java.net.ProtocolException;
 
-import rx.Observable;
-import rx.Subscriber;
-import rx.android.schedulers.AndroidSchedulers;
-import rx.schedulers.Schedulers;
+import io.reactivex.Observable;
+import io.reactivex.ObservableEmitter;
+import io.reactivex.ObservableOnSubscribe;
+import io.reactivex.Observer;
+import io.reactivex.android.schedulers.AndroidSchedulers;
+import io.reactivex.annotations.NonNull;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.schedulers.Schedulers;
 
 
 /**
@@ -58,28 +62,49 @@ <T> void request(final ProtocolRequest<?, T> request, DialogGetListener mDialogG
             }
         }
 
-        Observable.create(new Observable.OnSubscribe<Response<T>>() {
+        Observable.create(new ObservableOnSubscribe<Response<T>>() {
             @Override
-            public void call(Subscriber<? super Response<T>> subscriberOut) {
+            public void subscribe(@NonNull ObservableEmitter<Response<T>> subscriberOut) throws Exception {
                 // 最关键的就是用NoHttp的同步请求请求到response了，其它的都是rxjava做的，跟nohttp无关的。
                 Response<T> response = NoHttp.startRequestSync(request);
                 if (response.isSucceed() || response.isFromCache()) {
                     subscriberOut.onNext(response);
                 } else {
                     subscriberOut.onError(response.getException());
                 }
-
             }
         }).subscribeOn(Schedulers.io())
                 .observeOn(AndroidSchedulers.mainThread())
-                .subscribe(new Subscriber<Response<T>>() {
+                .subscribe(new Observer<Response<T>>() {
+                    private Disposable mDisposable;
+
+                    @Override
+                    public void onSubscribe(@NonNull Disposable d) {
+                        mDisposable = d;
+                    }
+
                     @Override
-                    public void onCompleted() {
+                    public void onNext(@NonNull Response<T> tResponse) {
+                        mDisposable.dispose();
 
+                        // 关闭dialog.
+                        if (null != dialog && dialog.isShowing()) {
+                            try {
+                                dialog.dismiss();
+                            } catch (Exception e1) {
+                                Logger.e("Dialog-关闭异常：由于Dialog已经关闭或者依赖的Context不存在");
+                            }
+                        }
+
+                        if (null != responseInterfa) {
+                            responseInterfa.onNext(tResponse.get());
+                        }
                     }
 
                     @Override
-                    public void onError(Throwable e) {
+                    public void onError(@NonNull Throwable e) {
+                        mDisposable.dispose();
+
                         // 关闭dialog.
                         if (null != dialog && dialog.isShowing()) {
                             try {
@@ -115,8 +140,8 @@ public void onError(Throwable e) {
                             }
                             if (TextUtils.isEmpty(anUnknownErrorHint)) {
                                 show(dialog, R.string.error_unknow);
-                            }else{
-                                show(dialog,anUnknownErrorHint);
+                            } else {
+                                show(dialog, anUnknownErrorHint);
                             }
                         }
 
@@ -127,19 +152,8 @@ public void onError(Throwable e) {
                     }
 
                     @Override
-                    public void onNext(Response<T> tResponse) {
-                        // 关闭dialog.
-                        if (null != dialog && dialog.isShowing()) {
-                            try {
-                                dialog.dismiss();
-                            } catch (Exception e1) {
-                                Logger.e("Dialog-关闭异常：由于Dialog已经关闭或者依赖的Context不存在");
-                            }
-                        }
+                    public void onComplete() {
 
-                        if (null != responseInterfa) {
-                            responseInterfa.onNext(tResponse.get());
-                        }
                     }
                 });
     }
@@ -155,6 +169,7 @@ private void show(Dialog dialog, int stringId) {
             Toast.makeText(context, context.getResources().getString(stringId), Toast.LENGTH_SHORT).show();
         }
     }
+
     /**
      * 土司提示
      *

nohttputils-r1/src/main/java/com/liqi/nohttputils/nohttp/RxPollNoHttpConfig.java → nohttputils-r2/src/main/java/com/liqi/nohttputils/nohttp/RxPollNoHttpConfig.java

@@ -9,8 +9,8 @@
 import com.liqi.nohttputils.nohttp.rx_poll.operators.OnObserverEventListener;
 import com.yanzhenjie.nohttp.rest.RestRequest;
 
-import rx.functions.Action1;
-import rx.functions.Func1;
+import io.reactivex.functions.Consumer;
+import io.reactivex.functions.Predicate;
 
 /**
  * NoHttp轮询配置类
@@ -29,15 +29,15 @@ public class RxPollNoHttpConfig<T> {
     /**
      * 设置数据拦截监听对象
      */
-    private Func1<RxInformationModel<T>, Boolean> mBooleanFunc1;
+    private Predicate<RxInformationModel<T>> mBooleanFunc1;
     /**
      * 被观察者产生的行为事件监听器
      */
     private OnObserverEventListener<RestRequest<T>, RxInformationModel<T>> mOnObserverEventListener;
     /**
      * 观察者根据被观察产生的行为做出相应处理监听器
      */
-    private Action1<RxInformationModel<T>> mRxInformationModelAction1;
+    private Consumer<RxInformationModel<T>> mRxInformationModelAction1;
     /**
      * 网络请求参数对象
      */
@@ -54,15 +54,15 @@ public long getPeriod() {
         return mPeriod;
     }
 
-    public Func1<RxInformationModel<T>, Boolean> getBooleanFunc1() {
+    public Predicate<RxInformationModel<T>> getBooleanFunc1() {
         return mBooleanFunc1;
     }
 
     public RxRequestOperate<T> getRxRequestOperate() {
         return mRxRequestOperate;
     }
 
-    public Action1<RxInformationModel<T>> getRxInformationModelAction1() {
+    public Consumer<RxInformationModel<T>> getRxInformationModelAction1() {
         return mRxInformationModelAction1;
     }
 
@@ -123,7 +123,7 @@ public ConfigBuilder<T> setPeriod(long period) {
          * @param booleanFunc1 设置数据拦截监听对象
          * @return 构建轮询配置类
          */
-        public ConfigBuilder<T> setBooleanFunc1(Func1<RxInformationModel<T>, Boolean> booleanFunc1) {
+        public ConfigBuilder<T> setBooleanFunc1(Predicate<RxInformationModel<T>> booleanFunc1) {
             mRxPollNoHttpConfig.mBooleanFunc1 = booleanFunc1;
             return this;
         }
@@ -134,7 +134,7 @@ public ConfigBuilder<T> setBooleanFunc1(Func1<RxInformationModel<T>, Boolean> bo
          * @param rxInformationModelAction1 观察者根据被观察产生的行为做出相应处理监听器
          * @return 构建轮询配置类
          */
-        public ConfigBuilder<T> setRxInformationModelAction1(Action1<RxInformationModel<T>> rxInformationModelAction1) {
+        public ConfigBuilder<T> setRxInformationModelAction1(Consumer<RxInformationModel<T>> rxInformationModelAction1) {
             mRxPollNoHttpConfig.mRxInformationModelAction1 = rxInformationModelAction1;
             return this;
         }