实现对缓存策略的支持

Author: alafighting

app/src/main/java/com/im4j/kakacache/rxjava/demo/MainActivity.java

@@ -7,6 +7,7 @@
 import android.widget.Button;
 
 import com.im4j.kakacache.rxjava.netcache.RxRemoteCache;
+import com.im4j.kakacache.rxjava.netcache.strategy.FirstCacheStrategy;
 
 import java.util.concurrent.TimeUnit;
 
@@ -26,7 +27,7 @@ protected void onCreate(@Nullable Bundle savedInstanceState) {
 
         btnTest = (Button) findViewById(R.id.btn_test_cache);
         btnTest.setOnClickListener(view -> {
-            RxRemoteCache.cache(remote(), KEY_CACHE).subscribe(data -> {
+            remote().compose(RxRemoteCache.transformer(KEY_CACHE, new FirstCacheStrategy())).subscribe(data -> {
                 Log.e("main", "next  data="+data);
             }, error -> {
                 Log.e("main", "error", error);

library/src/main/java/com/im4j/kakacache/rxjava/core/CacheTarget.java

@@ -9,14 +9,14 @@ public enum CacheTarget {
     NONE,
     Memory,
     Disk,
-    MemoryDisk;
+    MemoryAndDisk;
 
     public boolean supportMemory() {
-        return this==Memory || this==MemoryDisk;
+        return this==Memory || this== MemoryAndDisk;
     }
 
     public boolean supportDisk() {
-        return this==Disk || this==MemoryDisk;
+        return this==Disk || this== MemoryAndDisk;
     }
 
 }

library/src/main/java/com/im4j/kakacache/rxjava/manager/RxCacheManager.java

@@ -19,13 +19,14 @@ public final void call(Subscriber<? super T> subscriber) {
             try {
                 execute(subscriber);
             } catch (Throwable e) {
-                subscriber.onError(e);
+                onError(subscriber, e);
                 return;
             }
 
             subscriber.onCompleted();
         }
         abstract void execute(Subscriber<? super T> subscriber);
+        abstract void onError(Subscriber<? super T> subscriber, Throwable e);
     }
 
 
@@ -42,11 +43,12 @@ public <T> rx.Observable<T> load(final String key) {
             @Override
             void execute(Subscriber<? super T> subscriber) {
                 T value = _load(key);
-//                if (value == null) {
-//                    subscriber.onError(new NotFoundException("找不到缓存’"+key+"'"));
-//                }
                 subscriber.onNext(value);
             }
+            @Override
+            void onError(Subscriber<? super T> subscriber, Throwable e) {
+                subscriber.onNext(null);
+            }
         }).subscribeOn(Schedulers.computation());
     }
 
@@ -61,6 +63,10 @@ void execute(Subscriber<? super Boolean> subscriber) {
                 _save(key, value, expires, target);
                 subscriber.onNext(true);
             }
+            @Override
+            void onError(Subscriber<? super Boolean> subscriber, Throwable e) {
+                subscriber.onNext(false);
+            }
         });
     }
 
@@ -76,6 +82,10 @@ void execute(Subscriber<? super Boolean> subscriber) {
                 boolean value = _containsKey(key);
                 subscriber.onNext(value);
             }
+            @Override
+            void onError(Subscriber<? super Boolean> subscriber, Throwable e) {
+                subscriber.onNext(false);
+            }
         });
     }
 
@@ -90,6 +100,10 @@ void execute(Subscriber<? super Boolean> subscriber) {
                 _remove(key);
                 subscriber.onNext(true);
             }
+            @Override
+            void onError(Subscriber<? super Boolean> subscriber, Throwable e) {
+                subscriber.onNext(false);
+            }
         });
     }
 
@@ -103,6 +117,10 @@ void execute(Subscriber<? super Boolean> subscriber) {
                 _clear();
                 subscriber.onNext(true);
             }
+            @Override
+            void onError(Subscriber<? super Boolean> subscriber, Throwable e) {
+                subscriber.onNext(false);
+            }
         });
     }
 

library/src/main/java/com/im4j/kakacache/rxjava/netcache/CacheTransformer.java

@@ -0,0 +1,52 @@
+package com.im4j.kakacache.rxjava.netcache;
+
+/**
+ * 数据
+ * @version alafighting 2016-06
+ */
+public class ResultData<T> {
+
+    public ResultFrom from;
+    public String key;
+    public T data;
+
+    public ResultData(ResultFrom from, String key, T data) {
+        this.from = from;
+        this.key = key;
+        this.data = data;
+    }
+
+    public ResultFrom getFrom() {
+        return from;
+    }
+
+    public void setFrom(ResultFrom from) {
+        this.from = from;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public T getData() {
+        return data;
+    }
+
+    public void setData(T data) {
+        this.data = data;
+    }
+
+    @Override
+    public String toString() {
+        return "ResultData{" +
+                "from=" + from +
+                ", key='" + key + '\'' +
+                ", data=" + data +
+                '}';
+    }
+
+}

library/src/main/java/com/im4j/kakacache/rxjava/netcache/ResultData.java

@@ -0,0 +1,9 @@
+package com.im4j.kakacache.rxjava.netcache;
+
+/**
+ * 数据来源
+ * @version alafighting 2016-06
+ */
+public enum ResultFrom {
+    Remote, Cache
+}

library/src/main/java/com/im4j/kakacache/rxjava/netcache/ResultFrom.java

@@ -12,6 +12,7 @@
 import com.im4j.kakacache.rxjava.core.memory.journal.LRUMemoryJournal;
 import com.im4j.kakacache.rxjava.core.memory.storage.SimpleMemoryStorage;
 import com.im4j.kakacache.rxjava.manager.RxCacheManager;
+import com.im4j.kakacache.rxjava.netcache.strategy.CacheStrategy;
 
 import java.io.File;
 
@@ -59,13 +60,30 @@ public static <T> rx.Observable<Boolean> save(String key, T value, CacheTarget t
     }
 
     public static <T> rx.Observable<Boolean> save(String key, T value) {
-        return save(key, value, CacheTarget.MemoryDisk);
+        return save(key, value, CacheTarget.MemoryAndDisk);
     }
 
 
 
-    public static <T> Observable<CacheTransformer.CacheData<T>> cache(Observable<T> observable, String key) {
-        return observable.compose(new CacheTransformer<>(key));
+    public static <T> Observable.Transformer<T, ResultData<T>> transformer(String key, CacheStrategy strategy) {
+        return new CacheTransformer<T>(key, strategy);
+    }
+
+
+
+    private static class CacheTransformer<T> implements Observable.Transformer<T, ResultData<T>> {
+        private String key;
+        private CacheStrategy<T> strategy;
+
+        public CacheTransformer(String key, CacheStrategy<T> strategy) {
+            this.key = key;
+            this.strategy = strategy;
+        }
+
+        @Override
+        public Observable<ResultData<T>> call(Observable<T> source) {
+            return strategy.execute(key, source);
+        }
     }
 
 }

library/src/main/java/com/im4j/kakacache/rxjava/netcache/RxRemoteCache.java

@@ -0,0 +1,29 @@
+package com.im4j.kakacache.rxjava.netcache.strategy;
+
+import android.util.Log;
+
+import com.im4j.kakacache.rxjava.netcache.ResultData;
+import com.im4j.kakacache.rxjava.netcache.ResultFrom;
+import com.im4j.kakacache.rxjava.netcache.RxRemoteCache;
+
+import rx.Observable;
+
+/**
+ * 缓存策略基类
+ * @version alafighting 2016-06
+ */
+abstract class BasicCacheStrategy<T> implements CacheStrategy {
+
+    Observable<ResultData<T>> loadCache(String key) {
+        return RxRemoteCache.load(key).map(it -> new ResultData<>(ResultFrom.Cache, key, (T) it));
+    }
+
+    Observable<ResultData<T>> loadRemote(String key, Observable<T> source) {
+        return source.map(it -> {
+            RxRemoteCache.save(key, it).subscribe(status -> Log.e("save status", "status="+status) );
+            Log.e("transformer", "save  key="+key+", value="+it);
+            return new ResultData<T>(ResultFrom.Remote, key, it);
+        });
+    }
+
+}

library/src/main/java/com/im4j/kakacache/rxjava/netcache/strategy/BasicCacheStrategy.java

@@ -0,0 +1,15 @@
+package com.im4j.kakacache.rxjava.netcache.strategy;
+
+import com.im4j.kakacache.rxjava.netcache.ResultData;
+
+import rx.Observable;
+
+/**
+ * 缓存策略
+ * @version alafighting 2016-06
+ */
+public interface CacheStrategy<T> {
+
+    public Observable<ResultData<T>> execute(String key, Observable<T> source);
+
+}

library/src/main/java/com/im4j/kakacache/rxjava/netcache/strategy/CacheStrategy.java

@@ -0,0 +1,22 @@
+package com.im4j.kakacache.rxjava.netcache.strategy;
+
+import com.im4j.kakacache.rxjava.netcache.ResultData;
+
+import rx.Observable;
+import rx.schedulers.Schedulers;
+
+/**
+ * 优先缓存
+ * @version alafighting 2016-06
+ */
+public class FirstCacheStrategy<T> extends BasicCacheStrategy<T> {
+
+    @Override
+    public Observable<ResultData<T>> execute(String key, Observable source) {
+        return Observable.concat(
+                loadCache(key),
+                loadRemote(key, source)
+        ).first(it -> ((ResultData<T>)it).data != null).subscribeOn(Schedulers.io());
+    }
+
+}