tt/FLYSN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cache): 重构缓存架构并新增多级缓存及统计接口

Browse Source
This commit is contained in:
tongtongstudio
2026-07-27 11:42:58 +08:00
parent a7c6e1a00a
commit 6e995e0795
8 changed files with 418 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/build.gradle
View File

@@ -502,6 +502,12 @@ dependencies {
implementation 'androidx.recyclerview:recyclerview:1.2.1'
implementation 'androidx.legacy:legacy-support-v4:1.0.0'
implementation 'androidx.lifecycle:lifecycle-viewmodel:2.6.2'
implementation 'androidx.lifecycle:lifecycle-livedata:2.6.2'
implementation 'androidx.lifecycle:lifecycle-runtime:2.6.2'
// 可选：RxJava3 Observable → LiveData 的桥接
implementation 'androidx.lifecycle:lifecycle-reactivestreams:2.6.2'
testImplementation 'junit:junit:4.13.2'
androidTestImplementation 'androidx.test.ext:junit:1.1.3'
androidTestImplementation 'androidx.test.espresso:espresso-core:3.4.0'
@@ -533,6 +539,8 @@ dependencies {
annotationProcessor 'com.github.bumptech.glide:compiler:4.13.2'
//磁盘缓存
implementation 'com.jakewharton:disklrucache:2.0.2'
//图表
implementation 'com.github.PhilJay:MPAndroidChart:v3.1.0'
//Aria
implementation 'com.arialyy.aria:core:3.8.15'
annotationProcessor 'com.arialyy.aria:compiler:3.8.15'

96
app/src/main/java/com/fuying/sn/disklrucache/CacheHelper.java
View File

@@ -6,9 +6,7 @@ import android.graphics.drawable.Drawable;
import android.os.Environment;
import android.util.Log;
import com.fuying.sn.config.CommonConfig;
import com.jakewharton.disklrucache.DiskLruCache;
import com.tencent.mmkv.MMKV;
import org.json.JSONArray;
import org.json.JSONException;
@@ -19,20 +17,21 @@ import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;
/**
* 磁盘缓存帮助类
* https://www.cnblogs.com/aademeng/articles/6817058.html
*/
public class CacheHelper {
private static final String TAG = "DiskLruCacheHelper";
private MMKV mMMKV = MMKV.mmkvWithID(CommonConfig.MMKV_ID, MMKV.MULTI_PROCESS_MODE);
private static final String DIR_NAME = "diskCache";
private static final int MAX_COUNT = 1024 * 1024 * 1024;
private static final int DEFAULT_APP_VERSION = 1;
@@ -72,9 +71,8 @@ public class CacheHelper {
int appVersion = context == null ? DEFAULT_APP_VERSION : Utils.getAppVersion(context);
DiskLruCache diskLruCache = null;
try {
diskLruCache = DiskLruCache.open(
return DiskLruCache.open(
dir,
appVersion,
DEFAULT_APP_VERSION,
@@ -82,13 +80,12 @@ public class CacheHelper {
} catch (IOException e) {
e.printStackTrace();
}
return diskLruCache;
return null;
}
private DiskLruCache generateCache(Context context, String dirName, int maxCount) {
DiskLruCache diskLruCache = null;
try {
diskLruCache = DiskLruCache.open(
return DiskLruCache.open(
getDiskCacheDir(context, dirName),
Utils.getAppVersion(context),
DEFAULT_APP_VERSION,
@@ -96,7 +93,7 @@ public class CacheHelper {
} catch (IOException e) {
e.printStackTrace();
}
return diskLruCache;
return null;
}
// =======================================
@@ -104,9 +101,7 @@ public class CacheHelper {
// =======================================
public void put(String key, String value) {
// Log.e(TAG, "put: key = " + key + " value = " + value);
mMMKV.encode(key + "_time", System.currentTimeMillis());
mMMKV.encode(key + "_mmkv", value);
Log.e(TAG, "put: key = " + key + " value = " + value);
DiskLruCache.Editor edit = null;
BufferedWriter bw = null;
@@ -116,15 +111,16 @@ public class CacheHelper {
return;
}
OutputStream os = edit.newOutputStream(0);
bw = new BufferedWriter(new OutputStreamWriter(os));
bw = new BufferedWriter(new OutputStreamWriter(os, StandardCharsets.UTF_8));
bw.write(value);
edit.commit();//write CLEAN
} catch (IOException e) {
e.printStackTrace();
Log.e(TAG, "put: " + e.getMessage());
try {
//s
edit.abort();//write REMOVE
if (edit != null) {
edit.abort();//write REMOVE
}
} catch (IOException e1) {
e1.printStackTrace();
Log.e(TAG, "put: " + e1.getMessage());
@@ -142,30 +138,30 @@ public class CacheHelper {
}
public String getAsString(String key) {
// Log.e(TAG, "getAsString: " + key);
Log.e(TAG, "getAsString: " + key);
InputStream inputStream = null;
try {
//write READ
inputStream = get(key);
if (inputStream == null) {
return mMMKV.decodeString(key + "_mmkv", null);
return null;
}
StringBuilder sb = new StringBuilder();
int len = 0;
byte[] buf = new byte[128];
while ((len = inputStream.read(buf)) != -1) {
sb.append(new String(buf, 0, len));
InputStreamReader isr = new InputStreamReader(inputStream, StandardCharsets.UTF_8);
char[] buf = new char[1024];
int len;
while ((len = isr.read(buf)) != -1) {
sb.append(buf, 0, len);
}
return sb.toString();
} catch (IOException e) {
e.printStackTrace();
Log.e(TAG, "getAsString: " + e.getMessage());
} finally {
if (inputStream != null) {
try {
inputStream.close();
} catch (IOException e1) {
e1.printStackTrace();
Log.e(TAG, "getAsString: " + e1.getMessage());
} catch (IOException e) {
e.printStackTrace();
}
}
}
@@ -199,9 +195,9 @@ public class CacheHelper {
public JSONArray getAsJSONArray(String key) {
String JSONString = getAsString(key);
if (JSONString == null) return null;
try {
JSONArray obj = new JSONArray(JSONString);
return obj;
return new JSONArray(JSONString);
} catch (Exception e) {
e.printStackTrace();
return null;
@@ -233,7 +229,9 @@ public class CacheHelper {
} catch (Exception e) {
e.printStackTrace();
try {
editor.abort();//write REMOVE
if (editor != null) {
editor.abort();//write REMOVE
}
} catch (IOException e1) {
e1.printStackTrace();
}
@@ -256,8 +254,8 @@ public class CacheHelper {
if (is == null) {
return null;
}
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] buf = new byte[256];
int len = 0;
while ((len = is.read(buf)) != -1) {
@@ -266,6 +264,12 @@ public class CacheHelper {
res = baos.toByteArray();
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
is.close();
} catch (IOException e) {
e.printStackTrace();
}
}
return res;
}
@@ -289,7 +293,9 @@ public class CacheHelper {
} catch (IOException e) {
e.printStackTrace();
try {
editor.abort();
if (editor != null) {
editor.abort();
}
} catch (IOException e1) {
e1.printStackTrace();
}
@@ -365,8 +371,8 @@ public class CacheHelper {
// =======================================
public boolean remove(String key) {
try {
key = Utils.hashKeyForDisk(key);
return mDiskLruCache.remove(key);
String hashKey = Utils.hashKeyForDisk(key);
return mDiskLruCache.remove(hashKey);
} catch (IOException e) {
e.printStackTrace();
}
@@ -379,7 +385,6 @@ public class CacheHelper {
public void delete() throws IOException {
mDiskLruCache.delete();
mMMKV.clearAll();
}
public void flush() throws IOException {
@@ -413,12 +418,12 @@ public class CacheHelper {
//basic editor
public DiskLruCache.Editor editor(String key) {
try {
key = Utils.hashKeyForDisk(key);
String hashKey = Utils.hashKeyForDisk(key);
//wirte DIRTY
DiskLruCache.Editor edit = mDiskLruCache.edit(key);
DiskLruCache.Editor edit = mDiskLruCache.edit(hashKey);
//edit maybe null :the entry is editing
if (edit == null) {
Log.w(TAG, "the entry spcified key:" + key + " is editing by other . ");
Log.w(TAG, "the entry spcified key:" + hashKey + " is editing by other . ");
}
return edit;
} catch (IOException e) {
@@ -436,7 +441,7 @@ public class CacheHelper {
DiskLruCache.Snapshot snapshot = mDiskLruCache.get(Utils.hashKeyForDisk(key));
if (snapshot == null) //not find entry , or entry.readable = false
{
// Log.e(TAG, "not find entry , or entry.readable = false");
Log.e(TAG, "not find entry , or entry.readable = false");
return null;
}
//write READ
@@ -455,20 +460,17 @@ public class CacheHelper {
// =======================================
private File getDiskCacheDir(Context context, String uniqueName) {
String cachePath;
File cacheDir;
if (Environment.MEDIA_MOUNTED.equals(Environment.getExternalStorageState())
|| !Environment.isExternalStorageRemovable()) {
if (context.getExternalCacheDir() != null) {
cachePath = context.getExternalCacheDir().getPath();
} else if (context.getExternalFilesDir("cache") != null) {
cachePath = context.getExternalFilesDir("cache").getPath();
} else {
cachePath = context.getCacheDir().getPath();
cacheDir = context.getExternalCacheDir();
if (cacheDir == null) {
cacheDir = context.getCacheDir();
}
} else {
cachePath = context.getCacheDir().getPath();
cacheDir = context.getCacheDir();
}
return new File(cachePath + File.separator + uniqueName);
return new File(cacheDir, uniqueName);
}
}

178
app/src/main/java/com/fuying/sn/network/NetInterfaceManager.java
View File

@@ -67,6 +67,8 @@ import com.fuying.sn.network.api.SnControlApi;
import com.fuying.sn.network.api.SnInfoApi;
import com.fuying.sn.network.api.SnLogApi;
import com.fuying.sn.network.api.newly.ControlApi;
import com.fuying.sn.network.api.newly.StatisticsApi;
import com.fuying.sn.network.cache.RxCacheManager;
import com.fuying.sn.network.interceptor.RepeatRequestInterceptor;
import com.fuying.sn.service.LogcatService;
import com.fuying.sn.utils.ApkUtils;
@@ -125,19 +127,39 @@ import retrofit2.converter.gson.GsonConverterFactory;
public class NetInterfaceManager {
private static final String TAG = "NetInterfaceManager";
private static final RxCacheManager CACHE_MANAGER = RxCacheManager.getInstance();
public static void clearAllCache() {
Log.e(TAG, "clearAllCache: ");
CACHE_MANAGER.clearAllCache();
}
public static void removeCache(String key) {
Log.e(TAG, "removeCache: key = " + key);
CACHE_MANAGER.removeCache(key);
}
//缓存过期时间1秒
private static final long CACHE_EXPIRE_ONE_SECOND = 1000;
//默认 缓存过期时间10秒
private static final long DEFAULT_CACHE_EXPIRE = 10 * CACHE_EXPIRE_ONE_SECOND;
//缓存过期时间1分钟
private static final long CACHE_EXPIRE_ONE_MINUTE = 60 * CACHE_EXPIRE_ONE_SECOND;
//超时时间
private static final int TIME_OUT = 30;
// 缓存文件最大限制大小20M
private static final long CACHE_SIZE = 1024 * 1024 * 64;
private MMKV mMMKV = MMKV.mmkvWithID(CommonConfig.MMKV_ID, MMKV.MULTI_PROCESS_MODE);
@SuppressLint("StaticFieldLeak")
private static NetInterfaceManager INSTANCE;
private Context mContext;
private ContentResolver crv;
private MMKV mMMKV = MMKV.mmkvWithID(CommonConfig.MMKV_ID, MMKV.MULTI_PROCESS_MODE);
private Retrofit mRetrofit;
private OkHttpClient okHttpClient;
private CacheHelper mCacheHelper;
//超时时间
private static final int timeOut = 15;
// 缓存文件最大限制大小20M
private static final long cacheSize = 1024 * 1024 * 64;
private Retrofit mRetrofit;
private OkHttpClient mOkHttpClient;
private NetInterfaceManager(Context context) {
if (context == null) {
@@ -148,23 +170,23 @@ public class NetInterfaceManager {
this.mCacheHelper = new CacheHelper(mContext);
if (null == mRetrofit) {
if (okHttpClient == null) {
if (mOkHttpClient == null) {
//如果无法生存缓存文件目录，检测权限使用已经加上，检测手机是否把文件读写权限禁止了
OkHttpClient.Builder builder = new OkHttpClient.Builder();
builder.connectTimeout(timeOut, TimeUnit.SECONDS); // 设置连接超时时间
builder.writeTimeout(timeOut, TimeUnit.SECONDS);// 设置写入超时时间
builder.readTimeout(timeOut, TimeUnit.SECONDS);// 设置读取数据超时时间
builder.connectTimeout(TIME_OUT, TimeUnit.SECONDS); // 设置连接超时时间
builder.writeTimeout(TIME_OUT, TimeUnit.SECONDS);// 设置写入超时时间
builder.readTimeout(TIME_OUT, TimeUnit.SECONDS);// 设置读取数据超时时间
builder.retryOnConnectionFailure(true);// 设置进行连接失败重试
builder.addInterceptor(new RepeatRequestInterceptor());
// 设置缓存文件路径
String cacheDirectory = getCacheDir() + "/OkHttpCache";
Cache cache = new Cache(new File(cacheDirectory), cacheSize);
Cache cache = new Cache(new File(cacheDirectory), CACHE_SIZE);
builder.cache(cache);// 设置缓存
okHttpClient = builder.build();
mOkHttpClient = builder.build();
}
mRetrofit = new Retrofit.Builder()
.client(okHttpClient)
.client(mOkHttpClient)
.baseUrl(BuildConfig.ROOT_URL)
.addConverterFactory(GsonConverterFactory.create())
.addCallAdapterFactory(RxJava3CallAdapterFactory.create())
@@ -206,10 +228,132 @@ public class NetInterfaceManager {
}
public OkHttpClient getOkHttpClient() {
return okHttpClient;
return mOkHttpClient;
}
/**
* 极简调用：仅需 网络请求Observable + 缓存过期时间
*
* @param networkObservable 原始网络请求
* @param expireTimeMillis 缓存有效期（毫秒）
* @return 带自动缓存的Observable
*/
public static <T> Observable<T> withCache(Observable<T> networkObservable, String cacheKey, long expireTimeMillis) {
return Observable.defer(() -> {
// Log.e(TAG, "withCache: cacheKey = " + cacheKey);
// 1. 读取缓存
T cacheData = CACHE_MANAGER.getCache(cacheKey);
if (cacheData != null) {
Log.e(TAG, "withCache: getCache " + cacheKey);
return Observable.just(cacheData);
}
Log.e(TAG, "withCache: request " + cacheKey);
// 2. 无缓存则发起网络请求，成功后缓存
return networkObservable
.subscribeOn(Schedulers.io())
.doOnNext(data -> CACHE_MANAGER.saveCache(cacheKey, data, expireTimeMillis));
})
.subscribeOn(Schedulers.io()) // 缓存IO切到子线程
.observeOn(AndroidSchedulers.mainThread());
}
/**
* 通用缓存策略：优先从内存缓存获取，其次从磁盘缓存获取，最后请求网络
* 统一使用 cacheKey 同时作为内存和磁盘缓存的键，且支持 Stale-While-Revalidate 策略
*
* @param networkObservable 网络请求Observable
* @param cacheKey 缓存键（内存与磁盘通用）
* @param type 数据类型（通过 TypeToken 获取，支持泛型）
* @param expireTimeMillis 内存缓存过期时间（毫秒）
* @return 带三级缓存策略的Observable，可能会发射两次（缓存 + 最新网络数据）
*/
public static <T> Observable<BaseResponse<T>> withMultiLevelCache(
Observable<BaseResponse<T>> networkObservable,
String cacheKey,
Type type,
long expireTimeMillis) {
return Observable.<BaseResponse<T>>create(emitter -> {
Log.e(TAG, "withMultiLevelCache: cacheKey = " + cacheKey);
// 1. 优先从内存缓存获取
BaseResponse<T> memoryCache = CACHE_MANAGER.getCache(cacheKey);
if (memoryCache != null) {
Log.e(TAG, "withMultiLevelCache: hit memory cache");
emitter.onNext(memoryCache);
} else {
// 2. 内存未命中，从磁盘缓存获取
try {
String diskCacheJson = INSTANCE.mCacheHelper.getAsString(cacheKey);
if (!TextUtils.isEmpty(diskCacheJson)) {
Log.e(TAG, "withMultiLevelCache: hit disk cache");
BaseResponse<T> diskCache = GsonUtils.toJavaObject(diskCacheJson, type);
if (diskCache != null) {
emitter.onNext(diskCache);
}
}
} catch (Exception e) {
Log.e(TAG, "withMultiLevelCache: read disk cache error", e);
}
}
// 3. 始终请求网络数据以同步更新
Log.e(TAG, "withMultiLevelCache: request from network");
networkObservable
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<BaseResponse<T>>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
emitter.setDisposable(d);
}
@Override
public void onNext(BaseResponse<T> response) {
Log.e(TAG, "withMultiLevelCache: network success code = " + (response != null ? response.code : "null"));
if (response != null && response.code == 200) {
// 保存到内存和磁盘缓存
CACHE_MANAGER.saveCache(cacheKey, response, expireTimeMillis);
try {
String json = GsonUtils.toJSONString(response);
INSTANCE.mCacheHelper.put(cacheKey, json);
} catch (Exception e) {
Log.e(TAG, "withMultiLevelCache: save disk cache error", e);
}
} else {
// 状态码不为 200，清除缓存
Log.e(TAG, "withMultiLevelCache: code != 200, clear cache");
CACHE_MANAGER.removeCache(cacheKey);
INSTANCE.mCacheHelper.remove(cacheKey);
}
emitter.onNext(response);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "withMultiLevelCache: network error", e);
// 网络错误，清除缓存
CACHE_MANAGER.removeCache(cacheKey);
INSTANCE.mCacheHelper.remove(cacheKey);
if (!emitter.isDisposed()) {
emitter.onError(e);
}
}
@Override
public void onComplete() {
if (!emitter.isDisposed()) {
emitter.onComplete();
}
}
});
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
}
/*
*
* API
@@ -257,6 +401,10 @@ public class NetInterfaceManager {
return mRetrofit.create(ControlApi.class);
}
public StatisticsApi getStatisticsApi() {
return mRetrofit.create(StatisticsApi.class);
}
/*
*

1
app/src/main/java/com/fuying/sn/network/UrlAddress.java
View File

@@ -91,6 +91,7 @@ public class UrlAddress {
public final static String DEVICE_INFO = "device/info";
/*检查设备守护更新*/
public final static String CHECK_UPDATE = "device/check-update";
/*使用统计*/
/*获取整机控制配置*/
public final static String CONTROL_STATUS = "statistics/control-status";

30
app/src/main/java/com/fuying/sn/network/cache/CacheEntry.java vendored Normal file
View File

@@ -0,0 +1,30 @@
package com.fuying.sn.network.cache;
/**
* 缓存实体：存储请求结果、创建时间、有效期
*
* @param <T> 接口返回数据类型
*/
public class CacheEntry<T> {
// 缓存的成功数据
private final T data;
// 缓存创建时间（毫秒）
private final long createTime;
// 缓存有效期（毫秒）
private final long expireTimeMillis;
public CacheEntry(T data, long createTime, long expireTimeMillis) {
this.data = data;
this.createTime = createTime;
this.expireTimeMillis = expireTimeMillis;
}
// 判断缓存是否过期
public boolean isExpired() {
return System.currentTimeMillis() - createTime > expireTimeMillis;
}
public T getData() {
return data;
}
}

63
app/src/main/java/com/fuying/sn/network/cache/RxCacheManager.java vendored Normal file
View File

@@ -0,0 +1,63 @@
package com.fuying.sn.network.cache;
import java.util.concurrent.ConcurrentHashMap;
/**
* 全局缓存管理器：线程安全，存储所有接口缓存
*/
public class RxCacheManager {
// 单例
private static volatile RxCacheManager INSTANCE;
// 缓存容器：key=请求唯一标识，value=缓存实体
private final ConcurrentHashMap<String, CacheEntry<?>> cacheMap;
private RxCacheManager() {
cacheMap = new ConcurrentHashMap<>();
}
public static RxCacheManager getInstance() {
if (INSTANCE == null) {
synchronized (RxCacheManager.class) {
if (INSTANCE == null) {
INSTANCE = new RxCacheManager();
}
}
}
return INSTANCE;
}
/**
* 保存缓存
*/
public <T> void saveCache(String key, T data, long expireTimeMillis) {
if (key == null || data == null) return;
cacheMap.put(key, new CacheEntry<>(data, System.currentTimeMillis(), expireTimeMillis));
}
/**
* 获取缓存（自动判断过期）
*/
public <T> T getCache(String key) {
CacheEntry<?> entry = cacheMap.get(key);
if (entry == null || entry.isExpired()) {
// 缓存不存在/已过期，移除并返回null
cacheMap.remove(key);
return null;
}
return (T) entry.getData();
}
/**
* 清空所有缓存
*/
public void clearAllCache() {
cacheMap.clear();
}
/**
* 删除指定缓存
*/
public void removeCache(String key) {
cacheMap.remove(key);
}
}

38
app/src/main/java/com/fuying/sn/network/cache/RxCacheRequest.java vendored Normal file
View File

@@ -0,0 +1,38 @@
package com.fuying.sn.network.cache;
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;
/**
* 优化后：无需手动传CacheKey，自动从请求获取
*/
public class RxCacheRequest {
private static final RxCacheManager CACHE_MANAGER = RxCacheManager.getInstance();
/**
* 极简调用：仅需 网络请求Observable + 缓存过期时间
*
* @param networkObservable 原始网络请求
* @param expireTimeMillis 缓存有效期（毫秒）
* @return 带自动缓存的Observable
*/
public static <T> Observable<T> withCache(Observable<T> networkObservable, String cacheKey, long expireTimeMillis) {
return Observable.defer(() -> {
// 2. 读取缓存
T cacheData = CACHE_MANAGER.getCache(cacheKey);
if (cacheData != null) {
return Observable.just(cacheData);
}
// 3. 无缓存则发起网络请求，成功后缓存
return networkObservable
.subscribeOn(Schedulers.io())
.doOnNext(data -> CACHE_MANAGER.saveCache(cacheKey, data, expireTimeMillis));
})
.subscribeOn(Schedulers.io()) // 缓存IO切到子线程
.observeOn(AndroidSchedulers.mainThread());
}
}

66
app/src/main/java/com/fuying/sn/network/cache/RxTimeCacheManager.java vendored Normal file
View File

@@ -0,0 +1,66 @@
package com.fuying.sn.network.cache;
import java.util.concurrent.ConcurrentHashMap;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.ObservableTransformer;
public class RxTimeCacheManager {
// Thread-safe memory cache to hold our data and its timestamp
private static final ConcurrentHashMap<String, CacheEntry<?>> cacheMap = new ConcurrentHashMap<>();
/**
* Industry Standard RxJava Cache Transformer.
* * @param cacheKey Unique key for this specific request (e.g., URL or Method Name)
*
* @param validTimeMillis How long the data is valid before a new network request is allowed
*/
public static <T> ObservableTransformer<T, T> applyCache(final String cacheKey, final long validTimeMillis) {
return new ObservableTransformer<T, T>() {
@Override
public ObservableSource<T> apply(Observable<T> upstream) {
// defer() ensures this logic runs EVERY time someone subscribes, not at initialization
return Observable.defer(() -> {
CacheEntry<?> entry = cacheMap.get(cacheKey);
long currentTime = System.currentTimeMillis();
// 1. Check if cache exists and is within the time limit
if (entry != null && (currentTime - entry.timestamp) < validTimeMillis) {
// Cache HIT -> Return cached data immediately, do not trigger network
@SuppressWarnings("unchecked")
T cachedData = (T) entry.data;
return Observable.just(cachedData);
}
// 2. Cache MISS or EXPIRED -> Trigger upstream (network) and save result
return upstream.doOnNext(data -> {
// Save the fresh data and current timestamp back to the cache
cacheMap.put(cacheKey, new CacheEntry<>(data, System.currentTimeMillis()));
});
});
}
};
}
/**
* Clears a specific cache key if you need to force a refresh manually.
*/
public static void clearCache(String cacheKey) {
cacheMap.remove(cacheKey);
}
/**
* Internal wrapper to hold the payload and the time it was fetched.
*/
private static class CacheEntry<T> {
final T data;
final long timestamp;
CacheEntry(T data, long timestamp) {
this.data = data;
this.timestamp = timestamp;
}
}
}