修改一下线程中断的逻辑

licheedev · licheedev · commit 056d093596fa · 2019-11-25T10:59:04.000+08:00
diff --git a/README.md b/README.md
@@ -25,7 +25,7 @@ allprojects {
 }
 
   dependencies {
-	        implementation 'com.github.licheedev:Modbus4Android:2.0.0'
+	        implementation 'com.github.licheedev:Modbus4Android:2.0.1'
 }
 
 ```
diff --git a/app/src/main/java/com/licheedev/demo/utils/RxUtilEx.java b/app/src/main/java/com/licheedev/demo/utils/RxUtilEx.java
@@ -0,0 +1,120 @@
+package com.licheedev.demo.utils;
+
+import android.support.annotation.NonNull;
+import io.reactivex.Observable;
+import io.reactivex.ObservableSource;
+import io.reactivex.ObservableTransformer;
+import io.reactivex.Scheduler;
+import io.reactivex.Single;
+import io.reactivex.SingleSource;
+import io.reactivex.SingleTransformer;
+import io.reactivex.android.schedulers.AndroidSchedulers;
+import io.reactivex.functions.Function;
+import io.reactivex.schedulers.Schedulers;
+import java.util.concurrent.TimeUnit;
+
+public class RxUtilEx {
+
+    public static <T> ObservableTransformer<T, T> rxIoMain() {
+        return new ObservableTransformer<T, T>() {
+            public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
+                return upstream.subscribeOn(Schedulers.io())
+                    .observeOn(AndroidSchedulers.mainThread());
+            }
+        };
+    }
+
+    public static <T> SingleTransformer<T, T> rxSingleIoMain() {
+        return new SingleTransformer<T, T>() {
+            public SingleSource<T> apply(@io.reactivex.annotations.NonNull Single<T> upstream) {
+                return upstream.subscribeOn(Schedulers.io())
+                    .observeOn(AndroidSchedulers.mainThread());
+            }
+        };
+    }
+
+    /**
+     * 重试和重复执行
+     *
+     * @param retryDelay
+     * @param repeatDelay
+     * @param <T>
+     * @return
+     */
+    public static <T> ObservableTransformer<T, T> retryRepeat(final long retryDelay,
+        final long repeatDelay) {
+        return new ObservableTransformer<T, T>() {
+            public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
+                return upstream.retryWhen(
+                    new Function<Observable<Throwable>, ObservableSource<?>>() {
+                        @Override
+                        public ObservableSource<?> apply(Observable<Throwable> throwableObservable)
+                            throws Exception {
+                            return throwableObservable.delay(retryDelay, TimeUnit.MILLISECONDS);
+                        }
+                    })
+                    // 一定时间后重新查询
+                    .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
+                        @Override
+                        public ObservableSource<?> apply(Observable<Object> objectObservable)
+                            throws Exception {
+                            return objectObservable.delay(repeatDelay, TimeUnit.MILLISECONDS);
+                        }
+                    });
+            }
+        };
+    }
+
+    /**
+     * 重试
+     *
+     * @param delay
+     * @param <T>
+     * @return
+     */
+    public static <T> ObservableTransformer<T, T> retry(final long delay) {
+        return new ObservableTransformer<T, T>() {
+            public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
+                return upstream.retryWhen(
+                    new Function<Observable<Throwable>, ObservableSource<?>>() {
+                        @Override
+                        public ObservableSource<?> apply(Observable<Throwable> throwableObservable)
+                            throws Exception {
+                            return throwableObservable.delay(delay, TimeUnit.MILLISECONDS);
+                        }
+                    });
+            }
+        };
+    }
+
+    /**
+     * 重复执行
+     *
+     * @param delay
+     * @param <T>
+     * @return
+     */
+    public static <T> ObservableTransformer<T, T> repeat(final long delay) {
+        return new ObservableTransformer<T, T>() {
+            public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
+                return upstream
+                    // 一定时间后重新查询
+                    .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
+                        @Override
+                        public ObservableSource<?> apply(Observable<Object> objectObservable)
+                            throws Exception {
+                            return objectObservable.delay(delay, TimeUnit.MILLISECONDS);
+                        }
+                    });
+            }
+        };
+    }
+
+    public static Scheduler io() {
+        return Schedulers.io();
+    }
+
+    public static Scheduler main() {
+        return AndroidSchedulers.mainThread();
+    }
+}
diff --git a/modbus4android/src/main/java/com/licheedev/modbus4android/IModbusWorker.java b/modbus4android/src/main/java/com/licheedev/modbus4android/IModbusWorker.java
@@ -1,10 +1,8 @@
 package com.licheedev.modbus4android;
 
-import android.os.Handler;
 import com.serotonin.modbus4j.ModbusMaster;
 import com.serotonin.modbus4j.exception.ModbusInitException;
 import io.reactivex.Observable;
-import io.reactivex.Scheduler;
 
 /**
  * Modbus工作接口，定义了Modbus初始化相关的方法
@@ -35,19 +33,19 @@ interface IModbusWorker {
      */
     boolean isModbusOpened();
 
-    /**
-     * 给RxJava用的，Modbus工作线程调度器
-     *
-     * @return
-     */
-    Scheduler getModbusScheduler();
-
-    /**
-     * Modbus工作线程的Handler
-     *
-     * @return
-     */
-    Handler getModbusHandler();
+    ///**
+    // * 给RxJava用的，Modbus工作线程调度器
+    // *
+    // * @return
+    // */
+    //Scheduler getModbusScheduler();
+    //
+    ///**
+    // * Modbus工作线程的Handler
+    // *
+    // * @return
+    // */
+    //Handler getModbusHandler();
 
     /**
      * [RX]初始化modbus
diff --git a/modbus4android/src/main/java/com/licheedev/modbus4android/ModbusWorker.java b/modbus4android/src/main/java/com/licheedev/modbus4android/ModbusWorker.java
@@ -1,8 +1,5 @@
 package com.licheedev.modbus4android;
 
-import android.os.Build;
-import android.os.Handler;
-import android.os.HandlerThread;
 import android.support.annotation.NonNull;
 import android.util.Log;
 import com.serotonin.modbus4j.ModbusMaster;
@@ -29,16 +26,17 @@
 import io.reactivex.ObservableEmitter;
 import io.reactivex.ObservableOnSubscribe;
 import io.reactivex.Observer;
-import io.reactivex.Scheduler;
 import io.reactivex.android.schedulers.AndroidSchedulers;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.CompositeException;
 import io.reactivex.functions.Action;
 import io.reactivex.plugins.RxJavaPlugins;
+import io.reactivex.schedulers.Schedulers;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 /**
  * ModbusWorker实现，实现了初始化modbus，并增加了线圈、离散量输入、寄存器的读写方法
@@ -50,22 +48,13 @@ public class ModbusWorker implements IModbusWorker {
     private static final String NO_INIT_MESSAGE = "ModbusMaster hasn't been inited!";
 
     private final ExecutorService mRequestExecutor;
-    protected final Scheduler mModbusScheduler;
-    private final HandlerThread mModbusThread;
-    private final Handler mModbusHandler;
 
     protected ModbusMaster mModbusMaster;
 
     public ModbusWorker() {
 
         // modbus请求用的单一线程池
         mRequestExecutor = Executors.newSingleThreadExecutor();
-
-        // 辅助用的工作线程
-        mModbusThread = new HandlerThread("modbus-working-thread");
-        mModbusThread.start();
-        mModbusHandler = new Handler(mModbusThread.getLooper());
-        mModbusScheduler = AndroidSchedulers.from(mModbusThread.getLooper());
     }
 
     /**
@@ -91,12 +80,6 @@ public synchronized void release() {
             mModbusMaster = null;
         }
 
-        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.JELLY_BEAN_MR2) {
-            mModbusThread.quitSafely();
-        } else {
-            mModbusThread.quit();
-        }
-
         mRequestExecutor.shutdown();
     }
 
@@ -110,14 +93,6 @@ public synchronized boolean isModbusOpened() {
         return getModbusMaster() != null;
     }
 
-    public Scheduler getModbusScheduler() {
-        return mModbusScheduler;
-    }
-
-    public Handler getModbusHandler() {
-        return mModbusHandler;
-    }
-
     @Override
     public void checkWorkingState() throws ModbusInitException, IllegalStateException {
 
@@ -144,8 +119,16 @@ private <T> T doSync(Callable<T> callable)
         throws InterruptedException, ModbusInitException, ModbusTransportException,
         ModbusRespException, ExecutionException {
 
+        Future<T> submit = null;
         try {
-            return mRequestExecutor.submit(callable).get();
+            submit = mRequestExecutor.submit(callable);
+            return submit.get();
+        } catch (InterruptedException e) {
+            if (submit != null) {
+                submit.cancel(true);
+            }
+            Thread.currentThread().interrupt();
+            throw e;
         } catch (ExecutionException e) {
             //e.printStackTrace();
             Throwable cause = e.getCause();
@@ -191,7 +174,7 @@ public void subscribe(ObservableEmitter<T> emitter) throws Exception {
                     }
                 }
             }
-        }).subscribeOn(mModbusScheduler);
+        }).subscribeOn(Schedulers.io());
     }
 
     /**

Original file line number	Diff line number	Diff line change
`@@ -25,7 +25,7 @@ allprojects {`
`25`	`25`	`}`
`26`	`26`
`27`	`27`	`dependencies {`
`28`		`- implementation 'com.github.licheedev:Modbus4Android:2.0.0'`
	`28`	`+ implementation 'com.github.licheedev:Modbus4Android:2.0.1'`
`29`	`29`	`}`
`30`	`30`
`31`	`31`	```