optimize cache and net

lib_cache/src/main/java/com/android/sdk/cache/DiskLruStorageImpl.java

@@ -156,7 +156,7 @@ public class DiskLruStorageImpl implements Storage {
     }
 
     @Override
-    public <T> Flowable<Optional<T>> optionalFlowable(String key, Type type) {
+    public <T> Flowable<Optional<T>> flowableOptional(String key, Type type) {
         return CommonImpl.flowableOptionalEntity(key, type, this);
     }
 

lib_cache/src/main/java/com/android/sdk/cache/MMKVStorageImpl.java

@@ -165,7 +165,7 @@ public class MMKVStorageImpl implements Storage {
     }
 
     @Override
-    public <T> Flowable<Optional<T>> optionalFlowable(String key, Type type) {
+    public <T> Flowable<Optional<T>> flowableOptional(String key, Type type) {
         return CommonImpl.flowableOptionalEntity(key, type, this);
     }
 

lib_cache/src/main/java/com/android/sdk/cache/Storage.java

@@ -29,9 +29,25 @@ public interface Storage {
     @Nullable
     <T> T getEntity(String key, Type type);
 
+    /**
+     * 如果没有获取到缓存，那么 Flowable 将不会发送任何数据，默认在调用线程加载缓存。
+     *
+     * @param key  缓存的 key
+     * @param type 缓存实体类型，如果是泛型类型，请使用 {@link TypeFlag}标识
+     * @param <T>  缓存实体类型
+     * @return 缓存
+     */
     <T> Flowable<T> flowable(String key, Type type);
 
-    <T> Flowable<Optional<T>> optionalFlowable(String key, Type type);
+    /**
+     * 默认在调用线程加载缓存。
+     *
+     * @param key  缓存的 key
+     * @param type 缓存实体类型，如果是泛型类型，请使用 {@link TypeFlag}标识
+     * @param <T>  缓存实体类型
+     * @return 缓存
+     */
+    <T> Flowable<Optional<T>> flowableOptional(String key, Type type);
 
     void putString(String key, String value);
 

lib_cache/src/main/java/com/android/sdk/cache/StorageEx.kt

@@ -3,7 +3,7 @@ package com.android.sdk.cache
 import com.github.dmstocking.optional.java.util.Optional
 import io.reactivex.Flowable
 
-inline fun <reified T> Storage.entity(key: String): T? {
+inline fun <reified T> Storage.getEntity(key: String): T? {
     return this.getEntity(key, object : TypeFlag<T>() {}.type)
 }
 
@@ -11,6 +11,6 @@ inline fun <reified T> Storage.flowable(key: String): Flowable<T> {
     return this.flowable(key, object : TypeFlag<T>() {}.type)
 }
 
-inline fun <reified T> Storage.optionalFlowable(key: String): Flowable<Optional<T>> {
-    return this.optionalFlowable(key, object : TypeFlag<T>() {}.type)
+inline fun <reified T> Storage.flowableOptional(key: String): Flowable<Optional<T>> {
+    return this.flowableOptional(key, object : TypeFlag<T>() {}.type)
 }

lib_network/src/main/java/com/android/sdk/net/NetProvider.java

@@ -81,7 +81,7 @@ class NetProviderImpl implements NetProvider {
 
     void checkRequired() {
         if (mErrorMessage == null || mErrorDataAdapter == null || mNetworkChecker == null || mHttpConfig == null) {
-            throw new NullPointerException("缺少必要的参数，必须提供：ErrorMessage、mErrorDataAdapter、mNetworkChecker、HttpConfig。");
+            throw new NullPointerException("You must provide following object：ErrorMessage, mErrorDataAdapter, mNetworkChecker, HttpConfig.");
         }
     }
 

lib_network/src/main/java/com/android/sdk/net/core/HttpResultTransformer.java

@@ -94,7 +94,8 @@ public class HttpResultTransformer<Upstream, Downstream, T extends Result<Upstre
         }
 
         if (mRequireNonNullData) {
-            if (rResult.getData() == null) {//如果约定必须返回的数据却没有返回数据，则认为是服务器错误
+            //如果约定必须返回的数据却没有返回数据，则认为是服务器错误
+            if (rResult.getData() == null) {
                 throwAs(new ServerErrorException(ServerErrorException.UNKNOW_ERROR));
             }
         }

lib_network/src/main/java/com/android/sdk/net/kit/CombinedResult.kt

@@ -0,0 +1,16 @@
+package com.android.sdk.net.kit
+
+/**
+ *@author Ztiany
+ *      Email: ztiany3@gmail.com
+ *      Date : 2019-09-10 11:11
+ */
+data class CombinedResult<T>(
+        val dataType: DataType,
+        val data: T?,
+        val error: Throwable?
+)
+
+enum class DataType {
+    Remote, Disk
+}

lib_network/src/main/java/com/android/sdk/net/kit/ResultHandlers.java

@@ -73,8 +73,8 @@ public class ResultHandlers {
     }
 
     /**
-     * 与{@link #resultExtractor()}的行为类型，但是最后把 HttpResult<T> 中的数据 T 用 {@link Optional} 包装后再转发到下游。
-     * 适用于 HttpResult.getData() 可以为 Null 的情况
+     * 与{@link #resultExtractor()}的行为类似，但是最后把 HttpResult<T> 中的数据 T 用 {@link Optional} 包装后再转发到下游。
+     * 适用于 HttpResult.getData() 可能为 null 的情况
      */
     @SuppressWarnings("unchecked")
     private static <Upstream, T extends Result<Upstream>> HttpResultTransformer<Upstream, Optional<Upstream>, T> _optionalExtractor() {

lib_network/src/main/java/com/android/sdk/net/kit/RxResultKit.java

@@ -1,7 +1,6 @@
 package com.android.sdk.net.kit;
 
 import android.support.annotation.NonNull;
-import android.support.annotation.Nullable;
 
 import com.android.sdk.net.NetContext;
 import com.android.sdk.net.exception.ApiErrorException;
@@ -26,23 +25,23 @@ public class RxResultKit {
 
     /**
      * <pre>
-     * 1. 如果网络不可用，直接返回缓存，如果没有缓存，报错没有网络连接
-     * 2. 如果存在网络
-     *      2.1 如果没有缓存，则从网络获取
-     *      2.1 如果有缓存，则先返回缓存，然后从网络获取
-     *      2.1 对比缓存与网络数据，如果没有更新，则忽略
-     *      2.1 如果有更新，则更新缓存，并返回网络数据
+     * 1. 如果网络不可用，直接返回缓存，如果没有缓存，报错没有网络连接。
+     * 2. 如果存在网络。
+     *      2.1 如果没有缓存，则从网络获取，此时网络加载发生错误将会被忽略
+     *      2.1 如果有缓存，则先返回缓存，然后从网络获取。
+     *      2.1 对比缓存与网络数据，如果没有更新，则忽略。
+     *      2.1 如果有更新，则更新缓存，并返回网络数据。
      * </pre>
      *
-     * @param remote    网络数据源
-     * @param local     本地数据源
-     * @param onNewData 当有更新时，返回新的数据，可以在这里存储
-     * @param <T>       数据类型
-     * @param selector  比较器，返回当 true 表示两者相等，参数顺序为 (local, remote)
+     * @param remote    网络数据源。
+     * @param local     本地数据源。
+     * @param onNewData 当有更新时，返回新的数据，可以在这里进行存储操作。
+     * @param <T>       数据类型。
+     * @param selector  比较器，返回当 true 表示两者相等，如果相等，则 remote 数据将会被忽略。
      * @return 组合后的Observable
      * </T>
      */
-    public static <T> Flowable<Optional<T>> composeMultiSource(
+    public static <T> Flowable<Optional<T>> concatMultiSource(
             Flowable<Optional<T>> remote,
             Flowable<Optional<T>> local,
             Selector<T> selector,
@@ -50,13 +49,15 @@ public class RxResultKit {
 
         //没有网络
         if (!NetContext.get().connected()) {
-            return local.flatMap((Function<Optional<T>, Publisher<Optional<T>>>) tOptional -> {
-                if (tOptional.isPresent()) {
-                    return Flowable.just(tOptional);
+
+            return local.flatMap((Function<Optional<T>, Publisher<Optional<T>>>) optional -> {
+                if (optional.isPresent()) {
+                    return Flowable.just(optional);
                 } else {
                     return Flowable.error(new NetworkErrorException());
                 }
             });
+
         }
 
         //有网络
@@ -70,7 +71,7 @@ public class RxResultKit {
                     if (!localData.isPresent()) {
                         return remote.doOnNext(tOptional -> onNewData.accept(tOptional.orElse(null)));
                     }
-                    /*有缓存时网络错误，不触发错误，只有在过期时返回新的数据*/
+                    /*有缓存时网络错误，不触发错误，只有在缓存过期时返回新的数据*/
                     return remote
                             .onErrorResumeNext(onErrorResumeFunction(onNewData))
                             .filter(remoteData -> selector.test(localData.get(), remoteData.orElse(null)))
@@ -80,22 +81,61 @@ public class RxResultKit {
         return Flowable.concat(sharedLocal.filter(Optional::isPresent), complexRemote);
     }
 
-    public static <T> Flowable<Optional<T>> selectLocalOrRemote(Flowable<Optional<T>> remote, @Nullable T local, Selector<T> selector, Consumer<T> onNewData) {
-        //没有网络没有缓存
-        if (!NetContext.get().connected() && local == null) {
-            return Flowable.error(new NetworkErrorException());
-        }
-        //有缓存
-        if (local != null) {
-            return Flowable.concat(
-                    Flowable.just(Optional.of(local)),
-                    /*有缓存时网络错误，不触发错误，只有在过期时返回新的数据*/
-                    remote.onErrorResumeNext(onErrorResumeFunction(onNewData))
-                            .filter(tOptional -> selector.test(local, tOptional.orElse(null)))
-                            .doOnNext(tOptional -> onNewData.accept(tOptional.orElse(null))));
-        } else {
-            return remote;
-        }
+    /**
+     * 该方式，始终能得到错误通知。
+     *
+     * <pre>
+     *   1. 如果没有缓存，则从网络获取。
+     *   2. 如果有缓存，则先返回缓存，然后从网络获取，如果网络有错误，会以数据形式通知到下游。
+     *   3. 对比缓存与网络数据，如果没有更新，则忽略。
+     *   4. 如果有更新，则更新缓存，并返回网络数据。
+     * </pre>
+     *
+     * @param remote    网络数据源。
+     * @param local     本地数据源。
+     * @param onNewData 当有更新时，返回新的数据，可以在这里进行存储操作。
+     * @param <T>       数据类型。
+     * @param selector  比较器，返回当 true 表示两者相等，如果相等，则 remote 数据将会被忽略。
+     * @return 组合后的Observable
+     * </T>
+     */
+    public static <T> Flowable<CombinedResult<T>> combineMultiSource(
+            Flowable<Optional<T>> remote,
+            Flowable<Optional<T>> local,
+            Selector<T> selector,
+            Consumer<T> onNewData) {
+
+        ConnectableFlowable<Optional<T>> sharedLocal = local.replay();
+
+        sharedLocal.connect();
+
+        //组合数据
+        Flowable<CombinedResult<T>> complexRemote = sharedLocal
+                .flatMap((Function<Optional<T>, Publisher<CombinedResult<T>>>) localData -> {
+                    //没有缓存
+                    if (!localData.isPresent()) {
+                        return remote
+                                .doOnNext(tOptional -> onNewData.accept(tOptional.orElse(null)))
+                                .map(optional -> new CombinedResult<>(DataType.Remote, optional.orElse(null), null));
+                    }
+                    /*有缓存*/
+                    return remote
+                            .map(optional -> new CombinedResult<>(DataType.Remote, optional.orElse(null), null))
+                            .filter(remoteData -> selector.test(localData.get(), remoteData.getData()))
+                            .doOnNext(newData -> onNewData.accept(newData.getData()))
+                            .onErrorReturn(throwable -> {
+                                if (throwable instanceof ApiErrorException) {
+                                    onNewData.accept(null);
+                                }
+                                return new CombinedResult<>(DataType.Remote, null, throwable);
+                            });
+                });
+
+        Flowable<CombinedResult<T>> mappedLocal = sharedLocal
+                .filter(Optional::isPresent)
+                .map(optional -> new CombinedResult<>(DataType.Disk, optional.get(), null));
+
+        return Flowable.concat(mappedLocal, complexRemote);
     }
 
     @NonNull
@@ -113,9 +153,7 @@ public class RxResultKit {
     }
 
     private static boolean isNetworkError(Throwable exception) {
-        return exception instanceof IOException
-                || exception instanceof HttpException
-                || exception instanceof NetworkErrorException;
+        return exception instanceof IOException || exception instanceof HttpException || exception instanceof NetworkErrorException;
     }
 
 }

lib_network/src/main/java/com/android/sdk/net/kit/RxResultKit.kt

@@ -43,29 +43,40 @@ fun <E, T : Result<E>> Single<T>.resultChecker(): Single<Result<E>> {
     return (this.compose(ResultHandlers.resultChecker<E>()))
 }
 
-/**组合远程数据与本地数据，参考 [RxResultKit.composeMultiSource]*/
-fun <T> composeMultiSource(
+/**组合远程数据与本地数据，参考 [RxResultKit.concatMultiSource]*/
+fun <T> concatMultiSource(
         remote: Flowable<Optional<T>>,
         local: Flowable<Optional<T>>,
-        selector: (local: T, remote: T?) -> Boolean,
+        selector: Selector<T>,
         onNewData: (T?) -> Unit
 ): Flowable<Optional<T>> {
-    return RxResultKit.composeMultiSource(remote, local, Selector(selector), Consumer { onNewData(it) })
+    return RxResultKit.concatMultiSource(remote, local, selector, Consumer { onNewData(it) })
 }
 
-/**组合远程数据与本地数据，参考 [RxResultKit.composeMultiSource]*/
-fun <T> composeMultiSource(
+/**组合远程数据与本地数据，参考 [RxResultKit.concatMultiSource]*/
+fun <T> concatMultiSource(
         remote: Flowable<Optional<T>>,
         local: Flowable<Optional<T>>,
         onNewData: (T?) -> Unit
 ): Flowable<Optional<T>> {
-    return RxResultKit.composeMultiSource(remote, local, { t1, t2 -> t1 != t2 }, { onNewData(it) })
+    return RxResultKit.concatMultiSource(remote, local, { t1, t2 -> t1 != t2 }, { onNewData(it) })
 }
 
-fun <T> selectLocalOrRemote(remote: Flowable<Optional<T>>, local: T?, selector: (local: T, remote: T?) -> Boolean, onNewData: (T?) -> Unit): Flowable<Optional<T>> {
-    return RxResultKit.selectLocalOrRemote(remote, local, selector, onNewData)
+/**组合远程数据与本地数据，参考 [RxResultKit.combineMultiSource]*/
+fun <T> combineMultiSource(
+        remote: Flowable<Optional<T>>,
+        local: Flowable<Optional<T>>,
+        selector: Selector<T>,
+        onNewData: (T?) -> Unit
+): Flowable<CombinedResult<T>> {
+    return RxResultKit.combineMultiSource(remote, local, selector, Consumer { onNewData(it) })
 }
 
-fun <T> selectLocalOrRemote(remote: Flowable<Optional<T>>, local: T?, onNewData: (T?) -> Unit): Flowable<Optional<T>> {
-    return RxResultKit.selectLocalOrRemote(remote, local, { t1, t2 -> t1 != t2 }, { onNewData(it) })
-}
+/**组合远程数据与本地数据，参考 [RxResultKit.combineMultiSource]*/
+fun <T> combineMultiSource(
+        remote: Flowable<Optional<T>>,
+        local: Flowable<Optional<T>>,
+        onNewData: (T?) -> Unit
+): Flowable<CombinedResult<T>> {
+    return RxResultKit.combineMultiSource(remote, local, { t1, t2 -> t1 != t2 }, { onNewData(it) })
+}