Fixed - RMapCacheRx and RMapCacheReactive interfaces miss addListener() method. #3439

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

@@ -16,6 +16,7 @@
 package org.redisson.api;
 
 import org.redisson.api.map.MapLoader;
+import org.redisson.api.map.event.MapEntryListener;
 import reactor.core.publisher.Mono;
 
 import java.util.concurrent.TimeUnit;
@@ -291,4 +292,17 @@ public interface RMapCacheReactive<K, V> extends RMapReactive<K, V>, RDestroyabl
      */
     Mono<Long> remainTimeToLive(K key);
 
+    /**
+     * Adds map entry listener
+     *
+     * @see org.redisson.api.map.event.EntryCreatedListener
+     * @see org.redisson.api.map.event.EntryUpdatedListener
+     * @see org.redisson.api.map.event.EntryRemovedListener
+     * @see org.redisson.api.map.event.EntryExpiredListener
+     *
+     * @param listener - entry listener
+     * @return listener id
+     */
+    Mono<Integer> addListener(MapEntryListener listener);
+
 }

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

@@ -19,6 +19,7 @@ import io.reactivex.rxjava3.core.Completable;
 import io.reactivex.rxjava3.core.Maybe;
 import io.reactivex.rxjava3.core.Single;
 import org.redisson.api.map.MapLoader;
+import org.redisson.api.map.event.MapEntryListener;
 
 import java.util.concurrent.TimeUnit;
 
@@ -293,4 +294,17 @@ public interface RMapCacheRx<K, V> extends RMapRx<K, V>, RDestroyable {
      */
     Single<Long> remainTimeToLive(K key);
 
+    /**
+     * Adds map entry listener
+     *
+     * @see org.redisson.api.map.event.EntryCreatedListener
+     * @see org.redisson.api.map.event.EntryUpdatedListener
+     * @see org.redisson.api.map.event.EntryRemovedListener
+     * @see org.redisson.api.map.event.EntryExpiredListener
+     *
+     * @param listener - entry listener
+     * @return listener id
+     */
+    Single<Integer> addListener(MapEntryListener listener);
+
 }

redisson/src/test/java/org/redisson/rx/RedissonMapCacheRxTest.java

@@ -11,12 +11,17 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.redisson.api.ExpiredObjectListener;
 import org.redisson.api.RMapCacheRx;
 import org.redisson.api.RMapReactive;
 import org.redisson.api.RMapRx;
+import org.redisson.api.map.event.EntryEvent;
+import org.redisson.api.map.event.EntryExpiredListener;
 import org.redisson.codec.MsgPackJacksonCodec;
 
 public class RedissonMapCacheRxTest extends BaseRxTest {
@@ -174,6 +179,27 @@ public class RedissonMapCacheRxTest extends BaseRxTest {
         assertThat(toIterator(cache.keyIterator())).toIterable().containsOnly("0", "2", "3");
     }
 
+    @Test
+    public void testExpireListener() throws InterruptedException {
+        RMapCacheRx<String, String> cache = redisson.getMapCache("simple");
+        sync(cache.put("0", "8", 1, TimeUnit.SECONDS));
+
+        AtomicBoolean received = new AtomicBoolean();
+        cache.addListener(new EntryExpiredListener<String, String>() {
+            @Override
+            public void onExpired(EntryEvent<String, String> event) {
+                assertThat(event.getKey()).isEqualTo("0");
+                assertThat(event.getValue()).isEqualTo("8");
+                received.set(true);
+            }
+        }).blockingGet();
+
+        Thread.sleep(5100);
+
+        assertThat(received).isTrue();
+        assertThat(cache.size().blockingGet()).isZero();
+    }
+
     @Test
     public void testExpire() throws InterruptedException {
         RMapCacheRx<String, String> cache = redisson.getMapCache("simple");