Rxcache should only throw RxCacheException.

core/src/main/java/io/rx_cache2/RxCacheException.java

@@ -5,10 +5,18 @@
  */
 public final class RxCacheException extends RuntimeException {
 
+  public RxCacheException() {
+    super();
+  }
+
   public RxCacheException(String message) {
     super(message);
   }
 
+  public RxCacheException(Throwable throwable) {
+    super(throwable);
+  }
+
   public RxCacheException(String message, Throwable exception) {
     super(message, exception);
   }

core/src/main/java/io/rx_cache2/internal/Disk.java

@@ -27,6 +27,7 @@
 import java.util.Map;
 import javax.inject.Inject;
 
+import io.rx_cache2.RxCacheException;
 import io.rx_cache2.internal.encrypt.FileEncryptor;
 import io.victoralbertos.jolyglot.JolyglotGenerics;
 
@@ -128,7 +129,7 @@ public void save(String key, Object data, boolean isEncrypted, String encryptKey
         fileEncryptor.encrypt(encryptKey, new File(cacheDirectory, key));
       }
     } catch (Exception e) {
-      throw new RuntimeException(e);
+      throw new RxCacheException(e);
     } finally {
       try {
         if (fileWriter != null) {

core/src/main/java/io/rx_cache2/internal/ProcessorProvidersBehaviour.java

@@ -16,6 +16,10 @@
 
 package io.rx_cache2.internal;
 
+import java.util.concurrent.Callable;
+
+import javax.inject.Inject;
+
 import io.reactivex.Completable;
 import io.reactivex.Observable;
 import io.reactivex.ObservableSource;
@@ -27,8 +31,6 @@
 import io.rx_cache2.Reply;
 import io.rx_cache2.Source;
 import io.rx_cache2.internal.cache.GetDeepCopy;
-import java.util.concurrent.Callable;
-import javax.inject.Inject;
 
 public final class ProcessorProvidersBehaviour implements ProcessorProviders {
   private final io.rx_cache2.internal.cache.TwoLayersCache twoLayersCache;
@@ -108,44 +110,46 @@ <T> Observable<T> getData(final io.rx_cache2.ConfigProvider configProvider) {
 
   private Observable<Reply> getDataFromLoader(final io.rx_cache2.ConfigProvider configProvider,
       final Record record) {
-    return configProvider.getLoaderObservable().map(new Function<Object, Reply>() {
-      @Override public Reply apply(Object data) throws Exception {
+    return configProvider.getLoaderObservable().flatMap(new Function<Object, ObservableSource<Reply>>() {
+      @Override
+      public ObservableSource<Reply> apply(Object data) throws Exception {
         boolean useExpiredData = configProvider.useExpiredDataIfNotLoaderAvailable() != null ?
             configProvider.useExpiredDataIfNotLoaderAvailable()
             : useExpiredDataIfLoaderNotAvailable;
 
         if (data == null && useExpiredData && record != null) {
-          return new Reply(record.getData(), record.getSource(), configProvider.isEncrypted());
+          return Observable.just(new Reply(record.getData(), record.getSource(), configProvider.isEncrypted()));
         }
 
         clearKeyIfNeeded(configProvider);
 
         if (data == null) {
-          throw new io.rx_cache2.RxCacheException(io.rx_cache2.internal.Locale.NOT_DATA_RETURN_WHEN_CALLING_OBSERVABLE_LOADER
+          return Observable.error(new io.rx_cache2.RxCacheException(io.rx_cache2.internal.Locale.NOT_DATA_RETURN_WHEN_CALLING_OBSERVABLE_LOADER
               + " "
-              + configProvider.getProviderKey());
+              + configProvider.getProviderKey()));
         }
 
         twoLayersCache.save(configProvider.getProviderKey(), configProvider.getDynamicKey(),
             configProvider.getDynamicKeyGroup(), data, configProvider.getLifeTimeMillis(),
             configProvider.isExpirable(), configProvider.isEncrypted());
-        return new Reply(data, Source.CLOUD, configProvider.isEncrypted());
+        return Observable.just(new Reply(data, Source.CLOUD, configProvider.isEncrypted()));
       }
-    }).onErrorReturn(new Function<Object, Object>() {
-      @Override public Object apply(Object o) throws Exception {
+    }).onErrorResumeNext(new Function<Throwable, ObservableSource>() {
+      @Override
+      public ObservableSource apply(Throwable throwable) throws Exception {
         clearKeyIfNeeded(configProvider);
 
         boolean useExpiredData = configProvider.useExpiredDataIfNotLoaderAvailable() != null ?
             configProvider.useExpiredDataIfNotLoaderAvailable()
             : useExpiredDataIfLoaderNotAvailable;
 
         if (useExpiredData && record != null) {
-          return new Reply(record.getData(), record.getSource(), configProvider.isEncrypted());
+          return Observable.just(new Reply(record.getData(), record.getSource(), configProvider.isEncrypted()));
         }
 
-        throw new io.rx_cache2.RxCacheException(io.rx_cache2.internal.Locale.NOT_DATA_RETURN_WHEN_CALLING_OBSERVABLE_LOADER
+        return Observable.error(new io.rx_cache2.RxCacheException(io.rx_cache2.internal.Locale.NOT_DATA_RETURN_WHEN_CALLING_OBSERVABLE_LOADER
             + " "
-            + configProvider.getProviderKey(), (Throwable) o);
+            + configProvider.getProviderKey(), throwable));
       }
     });
   }

runtime/src/main/java/io/rx_cache2/internal/ProxyProviders.java

@@ -16,6 +16,13 @@
 
 package io.rx_cache2.internal;
 
+import java.lang.annotation.Annotation;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+
 import io.reactivex.BackpressureStrategy;
 import io.reactivex.Maybe;
 import io.reactivex.Observable;
@@ -24,13 +31,8 @@
 import io.rx_cache2.EncryptKey;
 import io.rx_cache2.Migration;
 import io.rx_cache2.MigrationCache;
+import io.rx_cache2.RxCacheException;
 import io.rx_cache2.SchemeMigration;
-import java.lang.annotation.Annotation;
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Callable;
 
 public final class ProxyProviders implements InvocationHandler {
   private final io.rx_cache2.internal.ProcessorProviders processorProviders;
@@ -90,7 +92,7 @@ public List<MigrationCache> getMigrations(Class<?> providersClass) {
         }
 
         String errorMessage = method.getName() + io.rx_cache2.internal.Locale.INVALID_RETURN_TYPE;
-        throw new RuntimeException(errorMessage);
+        return Observable.error(new RxCacheException(errorMessage));
       }
     }).blockingFirst();
   }