Feature - Rx and Reactive interfaces for RLocalCachedMap object. #3181

redisson/src/main/java/org/redisson/RedissonLocalCachedMap.java

@@ -67,16 +67,16 @@ public class RedissonLocalCachedMap<K, V> extends RedissonMap<K, V> implements R
     public RedissonLocalCachedMap(CommandAsyncExecutor commandExecutor, String name, LocalCachedMapOptions<K, V> options, 
             EvictionScheduler evictionScheduler, RedissonClient redisson, WriteBehindService writeBehindService) {
         super(commandExecutor, name, redisson, options, writeBehindService);
-        init(options, redisson, evictionScheduler);
+        init(options, evictionScheduler);
     }
 
     public RedissonLocalCachedMap(Codec codec, CommandAsyncExecutor connectionManager, String name, LocalCachedMapOptions<K, V> options, 
             EvictionScheduler evictionScheduler, RedissonClient redisson, WriteBehindService writeBehindService) {
         super(codec, connectionManager, name, redisson, options, writeBehindService);
-        init(options, redisson, evictionScheduler);
+        init(options, evictionScheduler);
     }
 
-    private void init(LocalCachedMapOptions<K, V> options, RedissonClient redisson, EvictionScheduler evictionScheduler) {
+    private void init(LocalCachedMapOptions<K, V> options, EvictionScheduler evictionScheduler) {
         syncStrategy = options.getSyncStrategy();
         storeMode = options.getStoreMode();
         storeCacheMiss = options.isStoreCacheMiss();

redisson/src/main/java/org/redisson/RedissonReactive.java

@@ -624,6 +624,22 @@ public <K, V> RMapReactive<K, V> getMap(String name, Codec codec, MapOptions<K,
                 new RedissonMapReactive<>(map, commandExecutor), RMapReactive.class);
     }
 
+    @Override
+    public <K, V> RLocalCachedMapReactive<K, V> getLocalCachedMap(String name, LocalCachedMapOptions<K, V> options) {
+        RMap<K, V> map = new RedissonLocalCachedMap<>(commandExecutor, name,
+                                                        options, evictionScheduler, null, writeBehindService);
+        return ReactiveProxyBuilder.create(commandExecutor, map,
+                new RedissonMapReactive<>(map, commandExecutor), RLocalCachedMapReactive.class);
+    }
+
+    @Override
+    public <K, V> RLocalCachedMapReactive<K, V> getLocalCachedMap(String name, Codec codec, LocalCachedMapOptions<K, V> options) {
+        RMap<K, V> map = new RedissonLocalCachedMap<>(codec, commandExecutor, name,
+                                                        options, evictionScheduler, null, writeBehindService);
+        return ReactiveProxyBuilder.create(commandExecutor, map,
+                new RedissonMapReactive<>(map, commandExecutor), RLocalCachedMapReactive.class);
+    }
+
     @Override
     public RTransactionReactive createTransaction(TransactionOptions options) {
         return new RedissonTransactionReactive(commandExecutor, options);

redisson/src/main/java/org/redisson/RedissonRx.java

@@ -604,6 +604,20 @@ public <K, V> RMapRx<K, V> getMap(String name, Codec codec, MapOptions<K, V> opt
                 new RedissonMapRx<K, V>(map, commandExecutor), RMapRx.class);
     }
 
+    @Override
+    public <K, V> RLocalCachedMapRx<K, V> getLocalCachedMap(String name, LocalCachedMapOptions<K, V> options) {
+        RMap<K, V> map = new RedissonLocalCachedMap<>(commandExecutor, name, options, evictionScheduler, null, writeBehindService);
+        return RxProxyBuilder.create(commandExecutor, map,
+                new RedissonMapRx<>(map, commandExecutor), RLocalCachedMapRx.class);
+    }
+
+    @Override
+    public <K, V> RLocalCachedMapRx<K, V> getLocalCachedMap(String name, Codec codec, LocalCachedMapOptions<K, V> options) {
+        RMap<K, V> map = new RedissonLocalCachedMap<>(codec, commandExecutor, name, options, evictionScheduler, null, writeBehindService);
+        return RxProxyBuilder.create(commandExecutor, map,
+                new RedissonMapRx<>(map, commandExecutor), RLocalCachedMapRx.class);
+    }
+
     @Override
     public RTransactionRx createTransaction(TransactionOptions options) {
         return new RedissonTransactionRx(commandExecutor, options);

redisson/src/main/java/org/redisson/api/RLocalCachedMapReactive.java

@@ -0,0 +1,57 @@
+package org.redisson.api;
+
+import reactor.core.publisher.Mono;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Map object with local entry cache support.
+ * <p>
+ * Each instance maintains local cache to achieve fast read operations.
+ * Suitable for maps which used mostly for read operations and network roundtrip delays are undesirable.
+ *
+ * @author Nikita Koksharov
+ *
+ * @param <K> map key
+ * @param <V> map value
+ */
+public interface RLocalCachedMapReactive<K, V> extends RMapReactive<K, V> {
+
+    /**
+     * Clears local cache across all instances
+     *
+     * @return void
+     */
+    Mono<Void> clearLocalCache();
+
+    /**
+     * Returns all keys stored in local cache
+     *
+     * @return keys
+     */
+    Set<K> cachedKeySet();
+
+    /**
+     * Returns all values stored in local cache
+     *
+     * @return values
+     */
+    Collection<V> cachedValues();
+
+    /**
+     * Returns all map entries stored in local cache
+     *
+     * @return entries
+     */
+    Set<Map.Entry<K, V>> cachedEntrySet();
+
+    /**
+     * Returns state of local cache
+     *
+     * @return map
+     */
+    Map<K, V> getCachedMap();
+
+}

redisson/src/main/java/org/redisson/api/RLocalCachedMapRx.java

@@ -0,0 +1,57 @@
+package org.redisson.api;
+
+import io.reactivex.rxjava3.core.Completable;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Map object with local entry cache support.
+ * <p>
+ * Each instance maintains local cache to achieve fast read operations.
+ * Suitable for maps which used mostly for read operations and network roundtrip delays are undesirable.
+ *
+ * @author Nikita Koksharov
+ *
+ * @param <K> map key
+ * @param <V> map value
+ */
+public interface RLocalCachedMapRx<K, V> extends RMapRx<K, V> {
+
+    /**
+     * Clears local cache across all instances
+     *
+     * @return void
+     */
+    Completable clearLocalCache();
+
+    /**
+     * Returns all keys stored in local cache
+     *
+     * @return keys
+     */
+    Set<K> cachedKeySet();
+
+    /**
+     * Returns all values stored in local cache
+     *
+     * @return values
+     */
+    Collection<V> cachedValues();
+
+    /**
+     * Returns all map entries stored in local cache
+     *
+     * @return entries
+     */
+    Set<Map.Entry<K, V>> cachedEntrySet();
+
+    /**
+     * Returns state of local cache
+     *
+     * @return map
+     */
+    Map<K, V> getCachedMap();
+
+}

redisson/src/main/java/org/redisson/api/RedissonReactiveClient.java

@@ -546,6 +546,31 @@ public interface RedissonReactiveClient {
      */
     <K, V> RMapReactive<K, V> getMap(String name, Codec codec, MapOptions<K, V> options);
 
+    /**
+     * Returns local cached map instance by name.
+     * Configured by parameters of options-object.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param options - local map options
+     * @return LocalCachedMap object
+     */
+    <K, V> RLocalCachedMapReactive<K, V> getLocalCachedMap(String name, LocalCachedMapOptions<K, V> options);
+
+    /**
+     * Returns local cached map instance by name
+     * using provided codec. Configured by parameters of options-object.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param codec - codec for keys and values
+     * @param options - local map options
+     * @return LocalCachedMap object
+     */
+    <K, V> RLocalCachedMapReactive<K, V> getLocalCachedMap(String name, Codec codec, LocalCachedMapOptions<K, V> options);
+
     /**
      * Returns set instance by name.
      *

redisson/src/main/java/org/redisson/api/RedissonRxClient.java

@@ -536,6 +536,31 @@ public interface RedissonRxClient {
      */
     <K, V> RMapRx<K, V> getMap(String name, Codec codec, MapOptions<K, V> options);
 
+    /**
+     * Returns local cached map instance by name.
+     * Configured by parameters of options-object.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param options - local map options
+     * @return LocalCachedMap object
+     */
+    <K, V> RLocalCachedMapRx<K, V> getLocalCachedMap(String name, LocalCachedMapOptions<K, V> options);
+
+    /**
+     * Returns local cached map instance by name
+     * using provided codec. Configured by parameters of options-object.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param codec - codec for keys and values
+     * @param options - local map options
+     * @return LocalCachedMap object
+     */
+    <K, V> RLocalCachedMapRx<K, V> getLocalCachedMap(String name, Codec codec, LocalCachedMapOptions<K, V> options);
+
     /**
      * Returns set instance by name.
      *

redisson/src/test/java/org/redisson/RedissonLocalCachedMapReactiveTest.java

@@ -0,0 +1,25 @@
+package org.redisson;
+
+import org.junit.jupiter.api.Test;
+import org.redisson.api.LocalCachedMapOptions;
+import org.redisson.api.RLocalCachedMapReactive;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class RedissonLocalCachedMapReactiveTest extends BaseReactiveTest {
+
+    @Test
+    public void test1() {
+        RLocalCachedMapReactive<String, String> m1 = redisson.getLocalCachedMap("test", LocalCachedMapOptions.defaults());
+        RLocalCachedMapReactive<String, String> m2 = redisson.getLocalCachedMap("test", LocalCachedMapOptions.defaults());
+
+        m1.put("1", "4").block();
+        m1.put("2", "5").block();
+        m1.put("3", "6").block();
+
+        assertThat(m1.getCachedMap()).containsKeys("1", "2", "3");
+
+        assertThat(m1.get("1").block()).isEqualTo(m2.get("1").block()).isEqualTo("4");
+    }
+
+}