+ 在tryNext中发生异常会回调tryError, 所有try函数都捕捉任何异常, 防止崩溃发生

+ 发生空指针异常默认提示 "数据为空"
| RxJava创建被观察者函数[from|emitter]取消订阅后异步阻塞发射事件崩溃
+ ResponseException 增加一个构造参数 tag, 用于错误回调中获取数据

README.md

@@ -58,7 +58,7 @@ allprojects {
 module 的 build.gradle
 
 ```groovy
-implementation 'com.github.liangjingkanji:Net:1.3.1'
+implementation 'com.github.liangjingkanji:Net:1.3.2'
 ```
 
 
@@ -448,7 +448,7 @@ fun <M> Observable<M>.page(
     block: PageObserver<M>.(M) -> Unit
 ) {
     subscribe(object : PageObserver<M>(pageRefreshLayout) {
-        override fun onNext(t: M) {
+        override fun tryNext(t: M) {
             block(t)
         }
     })
@@ -457,7 +457,9 @@ fun <M> Observable<M>.page(
 
 所有扩展订阅函数的都在`ObserverUtils`类中
 
+`try`系列函数都会捕捉异常
 
+`on`系列函数都是原始的Observer函数
 
 扩展函数都会返回对应的Observer对象可以进行手动取消订阅等操作
 
@@ -554,8 +556,21 @@ reset // 重置轮循器(包含计数器count和计时period)
 
 ## 快速创建被观察者
 
+使用本框架定义的创建被观察者函数可以避免异步阻塞时发射事件导致的空指针
+
+
+
 ```
-from { }.auto(this).subscribe { 
+from { 
+
+// 返回值为事件
+}.auto(this).subscribe { 
+	
+}
+
+shoot<Int> { 
+	it.onNext(2)
+}.auto(this).subscribe { 
 	
 }
 ```

net/src/main/AndroidManifest.xml

@@ -1,2 +1 @@
-<manifest xmlns:android="http://schemas.android.com/apk/res/android"
-    package="com.drake.net" />
+<manifest package="com.drake.net" />

net/src/main/java/com/drake/net/Net.kt

@@ -53,9 +53,7 @@ inline fun <reified M> get(
                     it.onNext(response.succeed())
                     it.onComplete()
                 } else {
-                    it.onError(
-                        ResponseException(response.failed(), response.code())
-                    )
+                    it.onError(ResponseException(response.failed(), response.code()))
                 }
             }
         } catch (e: java.lang.Exception) {
@@ -97,9 +95,7 @@ inline fun <reified M> post(
                     it.onNext(response.succeed())
                     it.onComplete()
                 } else {
-                    it.onError(
-                        ResponseException(response.failed(), response.code())
-                    )
+                    it.onError(ResponseException(response.failed(), response.code()))
                 }
             }
         } catch (e: java.lang.Exception) {
@@ -249,9 +245,7 @@ inline fun <reified M> syncPost(
                     it.onNext(response.succeed())
                     it.onComplete()
                 } else {
-                    it.onError(
-                        ResponseException(response.failed(), response.code())
-                    )
+                    it.onError(ResponseException(response.failed(), response.code()))
                 }
             }
         } catch (e: java.lang.Exception) {

net/src/main/java/com/drake/net/NetConfig.kt

@@ -35,24 +35,23 @@ object NetConfig {
     internal var onError: Throwable.() -> Unit = {
 
         val message = when (this) {
-            is NetworkError -> app.getString(R.string.network_error)
-            is URLError -> app.getString(R.string.url_error)
-            is HostError -> app.getString(R.string.host_error)
-            is ConnectTimeoutError -> app.getString(R.string.connect_timeout_error)
-            is ConnectException -> app.getString(R.string.connect_exception)
-            is WriteException -> app.getString(R.string.write_exception)
-            is ReadTimeoutError -> app.getString(R.string.read_timeout_error)
-            is DownloadError -> app.getString(R.string.download_error)
-            is NoCacheError -> app.getString(R.string.no_cache_error)
-            is ReadException -> app.getString(R.string.read_exception)
-            is ParseError -> app.getString(R.string.parse_error)
-            is RequestParamsException -> app.getString(R.string.request_error)
-            is ServerResponseException -> app.getString(R.string.server_error)
-            is ExecutionException -> app.getString(R.string.image_error)
+            is NetworkError -> app.getString(R.string.net_network_error)
+            is URLError -> app.getString(R.string.net_url_error)
+            is HostError -> app.getString(R.string.net_host_error)
+            is ConnectTimeoutError -> app.getString(R.string.net_connect_timeout_error)
+            is ConnectException -> app.getString(R.string.net_connect_exception)
+            is WriteException -> app.getString(R.string.net_write_exception)
+            is ReadTimeoutError -> app.getString(R.string.net_read_timeout_error)
+            is DownloadError -> app.getString(R.string.net_download_error)
+            is NoCacheError -> app.getString(R.string.net_no_cache_error)
+            is ReadException -> app.getString(R.string.net_read_exception)
+            is ParseError -> app.getString(R.string.net_parse_error)
+            is RequestParamsException -> app.getString(R.string.net_request_error)
+            is ServerResponseException -> app.getString(R.string.net_server_error)
+            is ExecutionException -> app.getString(R.string.net_image_error)
+            is NullPointerException -> app.getString(R.string.net_null_error)
             is ResponseException -> msg
-            else -> {
-                app.getString(R.string.other_error)
-            }
+            else -> app.getString(R.string.net_other_error)
 
         }
 
@@ -63,31 +62,27 @@ object NetConfig {
     internal var onStateError: Throwable.(view: View) -> Unit = {
 
         val message = when (this) {
-            is NetworkError -> app.getString(R.string.network_error)
-            is URLError -> app.getString(R.string.url_error)
-            is HostError -> app.getString(R.string.host_error)
-            is ConnectTimeoutError -> app.getString(R.string.connect_timeout_error)
-            is ReadException -> app.getString(R.string.read_exception)
-            is WriteException -> app.getString(R.string.write_exception)
-            is ConnectException -> app.getString(R.string.connect_exception)
-            is ReadTimeoutError -> app.getString(R.string.read_timeout_error)
-            is DownloadError -> app.getString(R.string.download_error)
-            is NoCacheError -> app.getString(R.string.no_cache_error)
-            is ParseError -> app.getString(R.string.parse_error)
-            is RequestParamsException -> app.getString(R.string.request_error)
-            is ServerResponseException -> app.getString(R.string.server_error)
-            is ExecutionException -> app.getString(R.string.image_error)
+            is NetworkError -> app.getString(R.string.net_network_error)
+            is URLError -> app.getString(R.string.net_url_error)
+            is HostError -> app.getString(R.string.net_host_error)
+            is ConnectTimeoutError -> app.getString(R.string.net_connect_timeout_error)
+            is ReadException -> app.getString(R.string.net_read_exception)
+            is WriteException -> app.getString(R.string.net_write_exception)
+            is ConnectException -> app.getString(R.string.net_connect_exception)
+            is ReadTimeoutError -> app.getString(R.string.net_read_timeout_error)
+            is DownloadError -> app.getString(R.string.net_download_error)
+            is NoCacheError -> app.getString(R.string.net_no_cache_error)
+            is ParseError -> app.getString(R.string.net_parse_error)
+            is RequestParamsException -> app.getString(R.string.net_request_error)
+            is ServerResponseException -> app.getString(R.string.net_server_error)
+            is ExecutionException -> app.getString(R.string.net_image_error)
             is ResponseException -> msg
-            else -> {
-                app.getString(R.string.other_error)
-            }
+            else -> app.getString(R.string.net_other_error)
         }
 
         printStackTrace()
         when (this) {
-            is ParseError, is ResponseException -> {
-                app.toast(message)
-            }
+            is ParseError, is ResponseException -> app.toast(message)
         }
     }
 }

net/src/main/java/com/drake/net/convert/JsonConverter.kt → net/src/main/java/com/drake/net/convert/DefaultConvert.kt

@@ -19,14 +19,14 @@ import java.lang.reflect.Type
 
 
 /**
- * 默认的转换器实现, 如果不满足需求推荐参考后自定义实现
+ * 默认的转换器实现, 如果不满足需求建议将该文件复制到项目中修改
  *
  * @property success String 错误码表示请求成功的值
  * @property code String 错误码的Key名称
  * @property msg String 错误信息的Key名称
  */
 @Suppress("UNCHECKED_CAST")
-abstract class JsonConverter(
+abstract class DefaultConvert(
     val success: String = "0",
     val code: String = "code",
     val msg: String = "msg"
@@ -53,11 +53,8 @@ abstract class JsonConverter(
                 val responseCode = jsonObject.getString(this.code)
 
                 if (responseCode == success) {
-                    succeedData = if (succeed === String::class.java) {
-                        body as S
-                    } else {
-                        convert(succeed, body)
-                    }
+                    succeedData = if (succeed === String::class.java) body as S
+                    else convert(succeed, body)
                 } else {
                     failedData = jsonObject.getString(msg) as F
                     code = responseCode.toInt()

net/src/main/java/com/drake/net/error/ResponseException.kt

@@ -9,8 +9,9 @@ package com.drake.net.error
 
 /**
  *  对应网络请求后台定义的错误信息
- * @property msg String 网络请求错误信息
- * @property code Int 网络请求错误码
- * @constructor
+ * @param msg 网络请求错误信息
+ * @param code 网络请求错误码
+ * @param tag 应对错误码判断为错时但是后端又返回了需要使用的数据(建议后端修改). 一般在Convert中设置数据
  */
-class ResponseException(val msg: String, val code: Int) : Throwable("code: $code, msg: $msg")
+class ResponseException(val msg: String, val code: Int, val tag: Any? = null) :
+    Throwable("code: $code, msg: $msg")

net/src/main/java/com/drake/net/observable/FromObservable.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright (C) 2018, Umbrella CompanyLimited All rights reserved.
+ * Project：Net
+ * Author：Drake
+ * Date：12/7/19 1:08 PM
+ */
+
+package com.drake.net.observable;
+
+import java.util.concurrent.Callable;
+
+import io.reactivex.Observable;
+import io.reactivex.Observer;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.internal.observers.DeferredScalarDisposable;
+
+
+public final class FromObservable<T> extends Observable<T> implements Callable<T> {
+    final Callable<? extends T> callable;
+
+    public FromObservable(Callable<? extends T> callable) {
+        this.callable = callable;
+    }
+
+    @Override
+    public void subscribeActual(Observer<? super T> observer) {
+
+        DeferredScalarDisposable<T> d = new DeferredScalarDisposable<T>(observer);
+        observer.onSubscribe(d);
+        if (d.isDisposed()) {
+            return;
+        }
+        T value;
+        try {
+            value = ObjectHelper.requireNonNull(callable.call(), "Callable returned null");
+        } catch (Throwable e) {
+            Exceptions.throwIfFatal(e);
+            return;
+        }
+
+        d.complete(value);
+    }
+
+    @Override
+    public T call() throws Exception {
+        return ObjectHelper.requireNonNull(callable.call(), "The callable returned a null value");
+    }
+}

net/src/main/java/com/drake/net/Interval.kt → net/src/main/java/com/drake/net/observable/Interval.kt

@@ -2,9 +2,9 @@
  * Copyright (C) 2018, Umbrella CompanyLimited All rights reserved.
  * Project：Net
  * Author：Drake
- * Date：11/28/19 11:45 PM
+ * Date：12/7/19 1:08 PM
  */
-package com.drake.net
+package com.drake.net.observable
 
 import io.reactivex.Observable
 import io.reactivex.Observer
@@ -61,7 +61,8 @@ class Interval(
 
     public override fun subscribeActual(observer: Observer<in Long?>) {
 
-        val agentObserver = IntervalRangeObserver(observer)
+        val agentObserver =
+            IntervalRangeObserver(observer)
         observerList.add(agentObserver)
 
         observer.onSubscribe(agentObserver)

net/src/main/java/com/drake/net/observable/ShootObservable.java

@@ -0,0 +1,290 @@
+/*
+ * Copyright (C) 2018, Umbrella CompanyLimited All rights reserved.
+ * Project：Net
+ * Author：Drake
+ * Date：12/7/19 1:30 PM
+ */
+package com.drake.net.observable;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.reactivex.Observable;
+import io.reactivex.ObservableOnSubscribe;
+import io.reactivex.Observer;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Cancellable;
+import io.reactivex.internal.disposables.CancellableDisposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.fuseable.SimpleQueue;
+import io.reactivex.internal.queue.SpscLinkedArrayQueue;
+import io.reactivex.internal.util.AtomicThrowable;
+import io.reactivex.plugins.RxJavaPlugins;
+
+public final class ShootObservable<T> extends Observable<T> {
+    final ObservableOnSubscribe<T> source;
+
+    public ShootObservable(ObservableOnSubscribe<T> source) {
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(Observer<? super T> observer) {
+        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
+        observer.onSubscribe(parent);
+
+        try {
+            source.subscribe(parent);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+        }
+    }
+
+    static final class CreateEmitter<T>
+            extends AtomicReference<Disposable>
+            implements io.reactivex.ObservableEmitter<T>, Disposable {
+
+        private static final long serialVersionUID = -3434801548987643227L;
+
+        final Observer<? super T> observer;
+
+        CreateEmitter(Observer<? super T> observer) {
+            this.observer = observer;
+        }
+
+        @Override
+        public void onNext(T t) {
+            if (t == null) {
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+                return;
+            }
+            if (!isDisposed()) {
+                observer.onNext(t);
+            }
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            if (!tryOnError(t)) {
+                RxJavaPlugins.onError(t);
+            }
+        }
+
+        @Override
+        public boolean tryOnError(Throwable t) {
+            if (t == null) {
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
+            }
+            if (!isDisposed()) {
+                try {
+                    observer.onError(t);
+                } finally {
+                    dispose();
+                }
+                return true;
+            }
+            return false;
+        }
+
+        @Override
+        public void onComplete() {
+            if (!isDisposed()) {
+                try {
+                    observer.onComplete();
+                } finally {
+                    dispose();
+                }
+            }
+        }
+
+        @Override
+        public void setDisposable(Disposable d) {
+            DisposableHelper.set(this, d);
+        }
+
+        @Override
+        public void setCancellable(Cancellable c) {
+            setDisposable(new CancellableDisposable(c));
+        }
+
+        @Override
+        public io.reactivex.ObservableEmitter<T> serialize() {
+            return new SerializedEmitter<T>(this);
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+
+        @Override
+        public String toString() {
+            return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
+        }
+    }
+
+    /**
+     * Serializes calls to onNext, onError and onComplete.
+     *
+     * @param <T> the value type
+     */
+    static final class SerializedEmitter<T>
+            extends AtomicInteger
+            implements io.reactivex.ObservableEmitter<T> {
+
+        private static final long serialVersionUID = 4883307006032401862L;
+
+        final io.reactivex.ObservableEmitter<T> emitter;
+
+        final AtomicThrowable error;
+
+        final SpscLinkedArrayQueue<T> queue;
+
+        volatile boolean done;
+
+        SerializedEmitter(io.reactivex.ObservableEmitter<T> emitter) {
+            this.emitter = emitter;
+            this.error = new AtomicThrowable();
+            this.queue = new SpscLinkedArrayQueue<T>(16);
+        }
+
+        @Override
+        public void onNext(T t) {
+            if (emitter.isDisposed() || done) {
+                return;
+            }
+            if (t == null) {
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+                return;
+            }
+            if (get() == 0 && compareAndSet(0, 1)) {
+                emitter.onNext(t);
+                if (decrementAndGet() == 0) {
+                    return;
+                }
+            } else {
+                SimpleQueue<T> q = queue;
+                synchronized (q) {
+                    q.offer(t);
+                }
+                if (getAndIncrement() != 0) {
+                    return;
+                }
+            }
+            drainLoop();
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            if (!tryOnError(t)) {
+                RxJavaPlugins.onError(t);
+            }
+        }
+
+        @Override
+        public boolean tryOnError(Throwable t) {
+            if (emitter.isDisposed() || done) {
+                return false;
+            }
+            if (t == null) {
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
+            }
+            if (error.addThrowable(t)) {
+                done = true;
+                drain();
+                return true;
+            }
+            return false;
+        }
+
+        @Override
+        public void onComplete() {
+            if (emitter.isDisposed() || done) {
+                return;
+            }
+            done = true;
+            drain();
+        }
+
+        void drain() {
+            if (getAndIncrement() == 0) {
+                drainLoop();
+            }
+        }
+
+        void drainLoop() {
+            io.reactivex.ObservableEmitter<T> e = emitter;
+            SpscLinkedArrayQueue<T> q = queue;
+            AtomicThrowable error = this.error;
+            int missed = 1;
+            for (; ; ) {
+
+                for (; ; ) {
+                    if (e.isDisposed()) {
+                        q.clear();
+                        return;
+                    }
+
+                    if (error.get() != null) {
+                        q.clear();
+                        e.onError(error.terminate());
+                        return;
+                    }
+
+                    boolean d = done;
+                    T v = q.poll();
+
+                    boolean empty = v == null;
+
+                    if (d && empty) {
+                        e.onComplete();
+                        return;
+                    }
+
+                    if (empty) {
+                        break;
+                    }
+
+                    e.onNext(v);
+                }
+
+                missed = addAndGet(-missed);
+                if (missed == 0) {
+                    break;
+                }
+            }
+        }
+
+        @Override
+        public void setDisposable(Disposable d) {
+            emitter.setDisposable(d);
+        }
+
+        @Override
+        public void setCancellable(Cancellable c) {
+            emitter.setCancellable(c);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return emitter.isDisposed();
+        }
+
+        @Override
+        public io.reactivex.ObservableEmitter<T> serialize() {
+            return this;
+        }
+
+        @Override
+        public String toString() {
+            return emitter.toString();
+        }
+    }
+
+}

net/src/main/java/com/drake/net/observer/DialogObserver.kt

@@ -19,7 +19,7 @@ import androidx.lifecycle.OnLifecycleEvent
 import com.drake.net.NetConfig
 import com.drake.net.NetConfig.defaultDialog
 import com.drake.net.R
-import io.reactivex.observers.DisposableObserver
+import io.reactivex.disposables.Disposable
 
 /**
  * 自动加载对话框网络请求
@@ -37,7 +37,7 @@ abstract class DialogObserver<M>(
     val activity: FragmentActivity?,
     var dialog: Dialog? = null,
     val cancelable: Boolean = true
-) : DisposableObserver<M>(), LifecycleObserver {
+) : TryObserver<DialogObserver<M>, M>(), LifecycleObserver {
 
     private var error: (DialogObserver<M>.(e: Throwable) -> Unit)? = null
 
@@ -48,7 +48,7 @@ abstract class DialogObserver<M>(
     ) : this(fragment?.activity, dialog, cancelable)
 
 
-    override fun onStart() {
+    override fun trySubscribe(d: Disposable) {
         activity ?: return
         activity.lifecycle.addObserver(this)
 
@@ -57,7 +57,7 @@ abstract class DialogObserver<M>(
             defaultDialog != null -> defaultDialog?.invoke(this, activity)
             else -> {
                 val progress = ProgressDialog(activity)
-                progress.setMessage(activity.getString(R.string.dialog_msg))
+                progress.setMessage(activity.getString(R.string.net_dialog_msg))
                 progress
             }
         }
@@ -74,29 +74,22 @@ abstract class DialogObserver<M>(
         }
     }
 
-    abstract override fun onNext(it: M)
-
-    fun error(block: (DialogObserver<M>.(e: Throwable) -> Unit)?): DialogObserver<M> {
-        error = block
-        return this
-    }
-
 
     /**
      * 关闭进度对话框并提醒错误信息
      *
      * @param e 包括错误信息
      */
-    override fun onError(e: Throwable) {
+    override fun tryError(e: Throwable) {
+        super.tryError(e)
         dismiss()
-        error?.invoke(this, e) ?: handleError(e)
     }
 
-    fun handleError(e: Throwable) {
+    override fun handleError(e: Throwable) {
         NetConfig.onError(e)
     }
 
-    override fun onComplete() {
+    override fun tryComplete() {
         dismiss()
     }
 

net/src/main/java/com/drake/net/observer/NetObserver.kt

@@ -7,17 +7,17 @@
 
 package com.drake.net.observer
 
-import androidx.lifecycle.*
+import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.LifecycleEventObserver
+import androidx.lifecycle.LifecycleObserver
+import androidx.lifecycle.LifecycleOwner
 import com.drake.net.NetConfig
-import io.reactivex.observers.DisposableObserver
 
 
 /**
  * 自动显示错误信息
  */
-abstract class NetObserver<M>() : DisposableObserver<M>(), LifecycleObserver {
-
-    private var error: (NetObserver<M>.(e: Throwable) -> Unit)? = null
+abstract class NetObserver<M>() : TryObserver<NetObserver<M>, M>(), LifecycleObserver {
 
     /**
      * 跟随生命周期
@@ -29,39 +29,12 @@ abstract class NetObserver<M>() : DisposableObserver<M>(), LifecycleObserver {
     ) : this() {
         lifecycleOwner?.lifecycle?.addObserver(object : LifecycleEventObserver {
             override fun onStateChanged(source: LifecycleOwner, event: Lifecycle.Event) {
-                if (event == lifecycle) {
-                    cancel()
-                }
+                if (event == lifecycle) dispose()
             }
         })
     }
 
-    abstract override fun onNext(it: M)
-
-    /**
-     * 关闭进度对话框并提醒错误信息
-     *
-     * @param e 包括错误信息
-     */
-    override fun onError(e: Throwable) {
-        error?.invoke(this, e) ?: handleError(e)
-    }
-
-    fun handleError(e: Throwable) {
+    override fun handleError(e: Throwable) {
         NetConfig.onError(e)
     }
-
-    @OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
-    private fun cancel() {
-        dispose()
-    }
-
-    override fun onComplete() {
-    }
-
-    fun error(block: (NetObserver<M>.(e: Throwable) -> Unit)?): NetObserver<M> {
-        error = block
-        return this
-    }
-
 }

net/src/main/java/com/drake/net/observer/ObservableUtils.kt

@@ -12,10 +12,12 @@ import android.os.Looper
 import androidx.lifecycle.Lifecycle
 import androidx.lifecycle.LifecycleEventObserver
 import androidx.lifecycle.LifecycleOwner
+import com.drake.net.observable.FromObservable
+import com.drake.net.observable.ShootObservable
 import io.reactivex.Observable
+import io.reactivex.ObservableEmitter
 import io.reactivex.android.schedulers.AndroidSchedulers
 import io.reactivex.schedulers.Schedulers
-import org.reactivestreams.Subscriber
 
 // <editor-fold desc="被观察者线程">
 
@@ -94,11 +96,11 @@ fun <M> Observable<M>.async(): Observable<M> {
  * @param M 返回值定义被观察者发射事件类型
  */
 fun <M> from(block: () -> M): Observable<M> {
-    return Observable.fromCallable(block)
+    return FromObservable(block)
 }
 
-fun <M> publish(block: (emitter: Subscriber<in M>) -> Unit): Observable<M> {
-    return Observable.fromPublisher(block)
+fun <M> shoot(block: (ObservableEmitter<in M>) -> Unit): Observable<M> {
+    return ShootObservable(block)
 }
 
 // </editor-fold>

net/src/main/java/com/drake/net/observer/ObserverUtils.kt

@@ -28,7 +28,7 @@ fun <M> Observable<M>.net(
     block: (NetObserver<M>.(M) -> Unit) = {}
 ): NetObserver<M> {
     val observer = object : NetObserver<M>(lifecycleOwner) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
@@ -49,7 +49,7 @@ fun <M> Observable<M>.state(
     block: StateObserver<M>.(M) -> Unit = {}
 ): StateObserver<M> {
     val observer = object : StateObserver<M>(stateLayout) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
@@ -62,7 +62,7 @@ fun <M> Observable<M>.state(
     block: StateObserver<M>.(M) -> Unit = {}
 ): StateObserver<M> {
     val observer = object : StateObserver<M>(view) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
@@ -75,7 +75,7 @@ fun <M> Observable<M>.state(
     block: StateObserver<M>.(M) -> Unit
 ): StateObserver<M> {
     val observer = object : StateObserver<M>(activity) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
@@ -86,7 +86,7 @@ fun <M> Observable<M>.state(
 fun <M> Observable<M>.state(fragment: Fragment, block: StateObserver<M>.(M) -> Unit = {}) {
 
     subscribe(object : StateObserver<M>(fragment) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     })
@@ -111,7 +111,7 @@ fun <M> Observable<M>.dialog(
     block: (DialogObserver<M>.(M) -> Unit) = {}
 ): DialogObserver<M> {
     val observer = object : DialogObserver<M>(activity, dialog, cancelable) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
@@ -127,7 +127,7 @@ fun <M> Observable<M>.dialog(
     block: (DialogObserver<M>.(M) -> Unit) = {}
 ): DialogObserver<M> {
     val observer = object : DialogObserver<M>(fragment, dialog, cancelable) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
@@ -150,14 +150,16 @@ fun <M> Observable<M>.refresh(
     block: RefreshObserver<M>.(M) -> Unit = {}
 ): RefreshObserver<M> {
     val observer = object : RefreshObserver<M>(refreshLayout, loadMore) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
     subscribe(observer)
     return observer
 }
+/*
 
+*/
 /**
  * 自动处理分页加载更多和下拉加载
  *
@@ -165,15 +167,16 @@ fun <M> Observable<M>.refresh(
  * @param pageRefreshLayout PageRefreshLayout
  * @param block (M) -> UnitUtils
  */
+
 fun <M> Observable<M>.page(
     pageRefreshLayout: PageRefreshLayout,
     block: PageObserver<M>.(M) -> Unit = {}
 ): PageObserver<M> {
     val observer = object : PageObserver<M>(pageRefreshLayout) {
-        override fun onNext(it: M) {
+        override fun tryNext(it: M) {
             block(it)
         }
     }
     subscribe(observer)
     return observer
-}
+}

net/src/main/java/com/drake/net/observer/PageObserver.kt

@@ -13,20 +13,18 @@ import com.drake.brv.BindingAdapter
 import com.drake.brv.PageRefreshLayout
 import com.drake.net.NetConfig.onStateError
 import com.scwang.smart.refresh.layout.constant.RefreshState
-import io.reactivex.observers.DisposableObserver
 
 /**
  * 自动结束下拉刷新和上拉加载状态
  * 自动展示缺省页
  * 自动分页加载
  */
-abstract class PageObserver<M>(val page: PageRefreshLayout) : DisposableObserver<M>() {
+@Suppress("unused", "MemberVisibilityCanBePrivate")
+abstract class PageObserver<M>(val page: PageRefreshLayout) :
+    TryObserver<PageObserver<M>, M>() {
 
 
-    val index
-        get() = page.index
-
-    private var error: (PageObserver<M>.(e: Throwable) -> Unit)? = null
+    val index get() = page.index
 
     init {
         page.addOnAttachStateChangeListener(object : OnAttachStateChangeListener {
@@ -40,27 +38,20 @@ abstract class PageObserver<M>(val page: PageRefreshLayout) : DisposableObserver
         })
     }
 
-    abstract override fun onNext(it: M)
-
     /**
      * 关闭进度对话框并提醒错误信息
      */
-    override fun onError(e: Throwable) {
+    override fun tryError(e: Throwable) {
+        super.tryError(e)
         if (page.state == RefreshState.Refreshing) {
             page.showError()
         } else page.finish(false)
-        error?.invoke(this, e) ?: handleError(e)
     }
 
-    fun handleError(e: Throwable) {
+    override fun handleError(e: Throwable) {
         onStateError.invoke(e, page)
     }
 
-    fun error(block: (PageObserver<M>.(e: Throwable) -> Unit)?): PageObserver<M> {
-        error = block
-        return this
-    }
-
     /**
      * 自动判断是添加数据还是覆盖数据, 以及数据为空或者NULL时[showEmpty]
      * @param hasMore 如果不穿数据, 默认已加载完全部(建议此时可以关闭[PageRefreshLayout]的加载更多功能)
@@ -80,7 +71,7 @@ abstract class PageObserver<M>(val page: PageRefreshLayout) : DisposableObserver
         dispose()
     }
 
-    override fun onComplete() {
+    override fun tryComplete() {
         if (page.stateEnabled) page.showContent() else page.finish()
     }
 

net/src/main/java/com/drake/net/observer/RefreshObserver.kt

@@ -11,7 +11,6 @@ import android.view.View
 import android.view.View.OnAttachStateChangeListener
 import com.drake.net.NetConfig
 import com.scwang.smart.refresh.layout.SmartRefreshLayout
-import io.reactivex.observers.DisposableObserver
 
 /**
  * 自动结束下拉刷新
@@ -19,9 +18,8 @@ import io.reactivex.observers.DisposableObserver
 abstract class RefreshObserver<M>(
     val refresh: SmartRefreshLayout,
     val loadMore: Boolean = false
-) : DisposableObserver<M>() {
+) : TryObserver<RefreshObserver<M>, M>() {
 
-    private var error: (RefreshObserver<M>.(e: Throwable) -> Unit)? = null
 
     init {
         refresh.setEnableLoadMore(loadMore)
@@ -36,23 +34,16 @@ abstract class RefreshObserver<M>(
         })
     }
 
-    abstract override fun onNext(it: M)
-
-    override fun onError(e: Throwable) {
+    override fun tryError(e: Throwable) {
+        super.tryError(e)
         refresh.finishRefresh(false)
-        error?.invoke(this, e) ?: handleError(e)
     }
 
-    fun handleError(e: Throwable) {
+    override fun handleError(e: Throwable) {
         NetConfig.onError(e)
     }
 
-    override fun onComplete() {
+    override fun tryComplete() {
         refresh.finishRefresh(true)
     }
-
-    fun error(block: (RefreshObserver<M>.(e: Throwable) -> Unit)?): RefreshObserver<M> {
-        error = block
-        return this
-    }
 }

net/src/main/java/com/drake/net/observer/StateObserver.kt

@@ -14,12 +14,12 @@ import androidx.fragment.app.Fragment
 import com.drake.net.NetConfig
 import com.drake.statelayout.StateLayout
 import com.drake.statelayout.state
-import io.reactivex.observers.DisposableObserver
+import io.reactivex.disposables.Disposable
 
 /**
  * 自动显示多状态布局
  */
-abstract class StateObserver<M> : DisposableObserver<M> {
+abstract class StateObserver<M> : TryObserver<StateObserver<M>, M> {
 
     val state: StateLayout
 
@@ -41,7 +41,7 @@ abstract class StateObserver<M> : DisposableObserver<M> {
         this.state = stateLayout
     }
 
-    override fun onStart() {
+    override fun trySubscribe(d: Disposable) {
         state.addOnAttachStateChangeListener(object : OnAttachStateChangeListener {
             override fun onViewAttachedToWindow(v: View?) {
             }
@@ -52,23 +52,19 @@ abstract class StateObserver<M> : DisposableObserver<M> {
         })
     }
 
-    abstract override fun onNext(it: M)
 
-    override fun onError(e: Throwable) {
+    override fun tryError(e: Throwable) {
+        super.tryError(e)
         state.showError()
         error?.invoke(this, e) ?: handleError(e)
     }
 
-    fun handleError(e: Throwable) {
+    override fun handleError(e: Throwable) {
         NetConfig.onStateError(e, state)
     }
 
-    fun error(block: (StateObserver<M>.(e: Throwable) -> Unit)?): StateObserver<M> {
-        error = block
-        return this
-    }
 
-    override fun onComplete() {
+    override fun tryComplete() {
         state.showContent()
     }
 

net/src/main/java/com/drake/net/observer/TryObserver.kt

@@ -0,0 +1,114 @@
+/*
+ * Copyright (C) 2018, Umbrella CompanyLimited All rights reserved.
+ * Project：Net
+ * Author：Drake
+ * Date：12/6/19 5:38 PM
+ */
+package com.drake.net.observer
+
+import io.reactivex.Observer
+import io.reactivex.disposables.Disposable
+import io.reactivex.exceptions.Exceptions
+import io.reactivex.internal.disposables.DisposableHelper
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * 捕捉所有异常
+ */
+
+abstract class TryObserver<T : Observer<M>, M> : AtomicReference<Disposable>(), Observer<M>,
+    Disposable {
+
+    private var error: (T.(e: Throwable) -> Unit)? = null
+
+    /**
+     * 错误处理回调
+     *
+     * 只在观察者和被观察者事件发送时存在不同步的情况下生效(一般用于异步处理)
+     */
+    fun error(block: (T.(e: Throwable) -> Unit)?): T {
+        error = block
+        @Suppress("UNCHECKED_CAST")
+        return this as T
+    }
+
+    // <editor-fold desc="异常捕捉">
+
+
+    protected open fun trySubscribe(d: Disposable) {}
+
+    @Suppress("UNCHECKED_CAST")
+    protected open fun tryError(e: Throwable) {
+        error?.invoke(this as T, e) ?: handleError(e)
+    }
+
+    protected abstract fun tryNext(it: M)
+
+    protected open fun tryComplete() {}
+
+    abstract fun handleError(e: Throwable)
+
+    // </editor-fold>
+
+    override fun onSubscribe(d: Disposable) {
+        if (DisposableHelper.setOnce(this, d)) {
+            try {
+                trySubscribe(d)
+            } catch (ex: Throwable) {
+                Exceptions.throwIfFatal(ex)
+                d.dispose()
+                onError(ex)
+            }
+        }
+    }
+
+    override fun onNext(t: M) {
+        if (!isDisposed) {
+            try {
+                tryNext(t)
+            } catch (e: Throwable) {
+                Exceptions.throwIfFatal(e)
+                get().dispose()
+                onError(e)
+            }
+        }
+    }
+
+
+    override fun onError(t: Throwable) {
+        if (!isDisposed) {
+            lazySet(DisposableHelper.DISPOSED)
+            try {
+                tryError(t)
+            } catch (e: Throwable) {
+                e.printStackTrace()
+                Exceptions.throwIfFatal(e)
+            }
+        }
+    }
+
+    override fun onComplete() {
+        if (!isDisposed) {
+            lazySet(DisposableHelper.DISPOSED)
+            try {
+                tryComplete()
+            } catch (e: Throwable) {
+                e.printStackTrace()
+                Exceptions.throwIfFatal(e)
+            }
+        }
+    }
+
+    override fun dispose() {
+        DisposableHelper.dispose(this)
+    }
+
+    override fun isDisposed(): Boolean {
+        return get() === DisposableHelper.DISPOSED
+    }
+
+    companion object {
+        private const val serialVersionUID = -7251123623727029452L
+    }
+
+}

net/src/main/res/values/strings.xml

@@ -3,23 +3,24 @@
 
 
     <!--网络请求异常-->
-    <string name="network_error">当前网络不可用</string>
-    <string name="url_error">请求资源地址错误</string>
-    <string name="host_error">无法找到指定服务器主机</string>
-    <string name="connect_timeout_error">连接服务器超时，请重试</string>
-    <string name="connect_exception">请检查网络连接</string>
-    <string name="read_exception">读取数据错误</string>
-    <string name="write_exception">发送数据错误</string>
-    <string name="read_timeout_error">读取服务器数据超时，请检查网络</string>
-    <string name="download_error">下载过程发生错误</string>
-    <string name="no_cache_error">读取缓存错误</string>
-    <string name="parse_error">解析数据时发生异常</string>
-    <string name="request_error">请求参数错误</string>
-    <string name="server_error">服务响应错误</string>
-    <string name="image_error">图片下载错误</string>
-    <string name="other_error">服务器未响应</string>
+    <string name="net_network_error">当前网络不可用</string>
+    <string name="net_url_error">请求资源地址错误</string>
+    <string name="net_host_error">无法找到指定服务器主机</string>
+    <string name="net_connect_timeout_error">连接服务器超时，请重试</string>
+    <string name="net_connect_exception">请检查网络连接</string>
+    <string name="net_read_exception">读取数据错误</string>
+    <string name="net_write_exception">发送数据错误</string>
+    <string name="net_read_timeout_error">读取服务器数据超时，请检查网络</string>
+    <string name="net_download_error">下载过程发生错误</string>
+    <string name="net_no_cache_error">读取缓存错误</string>
+    <string name="net_parse_error">解析数据时发生异常</string>
+    <string name="net_request_error">请求参数错误</string>
+    <string name="net_server_error">服务响应错误</string>
+    <string name="net_image_error">图片下载错误</string>
+    <string name="net_null_error">数据为空</string>
+    <string name="net_other_error">服务器未响应</string>
 
     <!--对话框-->
-    <string name="dialog_msg">加载中</string>
+    <string name="net_dialog_msg">加载中</string>
 
 </resources>

sample/src/main/java/com/drake/net/sample/MainActivity.kt

@@ -2,7 +2,6 @@ package com.drake.net.sample
 
 import android.os.Bundle
 import androidx.appcompat.app.AppCompatActivity
-import com.drake.net.download
 
 class MainActivity : AppCompatActivity() {
 
@@ -11,10 +10,6 @@ class MainActivity : AppCompatActivity() {
         setContentView(R.layout.activity_main)
 
 
-        val download = download(
-            "https://cdn.sspai.com/article/ebe361e4-c891-3afd-8680-e4bad609723e.jpg?imageMogr2/quality/95/thumbnail/!2880x620r/gravity/Center/crop/2880x620/interlace/1"
-        )
     }
 
-
 }