Merge pull request #3140 from mlkammer/master

Implemented feature request #3139: Added cache multimap variants for the Rx API

CHANGELOG.md

@@ -3,7 +3,13 @@ Redisson Releases History
 
 Сonsider __[Redisson PRO](https://redisson.pro)__ version for advanced features and support by SLA.
 
-### 13-Sep-2020 - 3.13.6 released
+### Unreleased
+
+Feature - `RListMultimapCacheRx` and `RSetMultimapCacheRx` interfaces added, usable from `RedissonRx` and `RBatchRx` APIs
+
+Fixed - `RedissonSetMultimapRx` could throw a class cast exception on its `get()` method because it actually contained a list based multimap instance
+
+### 13-Oct-2020 - 3.13.6 released
 
 Improvement - set pingConnectionInterval = 30000 by default
 

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

@@ -190,26 +190,58 @@ public class RedissonRx implements RedissonRxClient {
 
     @Override
     public <K, V> RListMultimapRx<K, V> getListMultimap(String name) {
-        return RxProxyBuilder.create(commandExecutor, new RedissonListMultimap<K, V>(commandExecutor, name), 
-                new RedissonListMultimapRx<K, V>(commandExecutor, name), RListMultimapRx.class);
+        RedissonListMultimap<K, V> listMultimap = new RedissonListMultimap<>(commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, listMultimap,
+                new RedissonListMultimapRx<K, V>(listMultimap, commandExecutor), RListMultimapRx.class);
     }
 
     @Override
     public <K, V> RListMultimapRx<K, V> getListMultimap(String name, Codec codec) {
-        return RxProxyBuilder.create(commandExecutor, new RedissonListMultimap<K, V>(codec, commandExecutor, name), 
-                new RedissonListMultimapRx<K, V>(codec, commandExecutor, name), RListMultimapRx.class);
+        RedissonListMultimap<K, V> listMultimap = new RedissonListMultimap<>(codec, commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, listMultimap,
+                new RedissonListMultimapRx<K, V>(listMultimap, commandExecutor), RListMultimapRx.class);
+    }
+
+    @Override
+    public <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name) {
+        RedissonListMultimapCache<K, V> listMultimap = new RedissonListMultimapCache<>(evictionScheduler, commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, listMultimap,
+                new RedissonListMultimapCacheRx<K, V>(listMultimap, commandExecutor), RListMultimapCacheRx.class);
+    }
+
+    @Override
+    public <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name, Codec codec) {
+        RedissonListMultimapCache<K, V> listMultimap = new RedissonListMultimapCache<>(evictionScheduler, codec, commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, listMultimap,
+                new RedissonListMultimapCacheRx<K, V>(listMultimap, commandExecutor), RListMultimapCacheRx.class);
     }
 
     @Override
     public <K, V> RSetMultimapRx<K, V> getSetMultimap(String name) {
-        return RxProxyBuilder.create(commandExecutor, new RedissonSetMultimap<K, V>(commandExecutor, name), 
-                new RedissonSetMultimapRx<K, V>(commandExecutor, name, this), RSetMultimapRx.class);
+        RedissonSetMultimap<K, V> setMultimap = new RedissonSetMultimap<>(commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, setMultimap,
+                new RedissonSetMultimapRx<K, V>(setMultimap, commandExecutor, this), RSetMultimapRx.class);
     }
 
     @Override
     public <K, V> RSetMultimapRx<K, V> getSetMultimap(String name, Codec codec) {
-        return RxProxyBuilder.create(commandExecutor, new RedissonSetMultimap<K, V>(codec, commandExecutor, name), 
-                new RedissonSetMultimapRx<K, V>(codec, commandExecutor, name, this), RSetMultimapRx.class);
+        RedissonSetMultimap<K, V> setMultimap = new RedissonSetMultimap<>(codec, commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, setMultimap,
+                new RedissonSetMultimapRx<K, V>(setMultimap, commandExecutor, this), RSetMultimapRx.class);
+    }
+
+    @Override
+    public <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name) {
+        RedissonSetMultimapCache<K, V> setMultimap = new RedissonSetMultimapCache<>(evictionScheduler, commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, setMultimap,
+                new RedissonSetMultimapCacheRx<K, V>(setMultimap, commandExecutor, this), RSetMultimapCacheRx.class);
+    }
+
+    @Override
+    public <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name, Codec codec) {
+        RedissonSetMultimapCache<K, V> setMultimap = new RedissonSetMultimapCache<>(evictionScheduler, codec, commandExecutor, name);
+        return RxProxyBuilder.create(commandExecutor, setMultimap,
+                new RedissonSetMultimapCacheRx<K, V>(setMultimap, commandExecutor, this), RSetMultimapCacheRx.class);
     }
 
     @Override

redisson/src/main/java/org/redisson/RedissonSetMultimapCache.java

@@ -39,7 +39,7 @@ public class RedissonSetMultimapCache<K, V> extends RedissonSetMultimap<K, V> im
 
     private final RedissonMultimapCache<K> baseCache;
     
-    RedissonSetMultimapCache(EvictionScheduler evictionScheduler, CommandAsyncExecutor connectionManager, String name) {
+    public RedissonSetMultimapCache(EvictionScheduler evictionScheduler, CommandAsyncExecutor connectionManager, String name) {
         super(connectionManager, name);
         if (evictionScheduler != null) {
             evictionScheduler.scheduleCleanMultimap(name, getTimeoutSetName());
@@ -47,7 +47,7 @@ public class RedissonSetMultimapCache<K, V> extends RedissonSetMultimap<K, V> im
         baseCache = new RedissonMultimapCache<K>(connectionManager, this, getTimeoutSetName(), prefix);
     }
 
-    RedissonSetMultimapCache(EvictionScheduler evictionScheduler, Codec codec, CommandAsyncExecutor connectionManager, String name) {
+    public RedissonSetMultimapCache(EvictionScheduler evictionScheduler, Codec codec, CommandAsyncExecutor connectionManager, String name) {
         super(codec, connectionManager, name);
         if (evictionScheduler != null) {
             evictionScheduler.scheduleCleanMultimap(name, getTimeoutSetName());

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

@@ -98,7 +98,32 @@ public interface RBatchRx {
      * @return SetMultimap object
      */
     <K, V> RSetMultimapRx<K, V> getSetMultimap(String name, Codec codec);
-    
+
+    /**
+     * Returns Set based Multimap cache instance by name.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular set multimap {@link #getSetMultimap(String)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @return RSetMultimapCacheRx object
+     */
+    <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name);
+
+    /**
+     * Returns Set based Multimap cache instance by name using provided codec for both map keys and values.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular set multimap {@link #getSetMultimap(String, Codec)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param codec - codec for keys and values
+     * @return RSetMultimapCacheRx object
+     */
+    <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name, Codec codec);
+
     /**
      * Returns set-based cache instance by <code>name</code>.
      * Uses map (value_hash, value) under the hood for minimal memory consumption.
@@ -209,7 +234,33 @@ public interface RBatchRx {
      * @return ListMultimap object
      */
     <K, V> RListMultimapRx<K, V> getListMultimap(String name, Codec codec);
-    
+
+
+    /**
+     * Returns List based Multimap cache instance by name.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular list multimap {@link #getListMultimap(String)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @return RListMultimapCacheRx object
+     */
+    <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name);
+
+    /**
+     * Returns List based Multimap cache instance by name using provided codec for both map keys and values.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular list multimap {@link #getListMultimap(String, Codec)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param codec - codec for keys and values
+     * @return RListMultimapCacheRx object
+     */
+    <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name, Codec codec);
+
     /**
      * Returns map instance by name.
      *

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

+/**
+ * Copyright (c) 2013-2020 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.api;
+
+/**
+ * Rx-ified version of {@link RListMultimapCache}.
+ *
+ * @author Marnix Kammer
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public interface RListMultimapCacheRx<K, V> extends RListMultimapRx<K, V>, RMultimapCacheRx<K, V> {
+
+}

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

+/**
+ * Copyright (c) 2013-2020 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.api;
+
+import io.reactivex.Single;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Rx-ified version of {@link RMultimapCache}.
+ *
+ * @author Marnix Kammer
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public interface RMultimapCacheRx<K, V> {
+
+    /**
+     * Set a timeout for key. After the timeout has expired, the key and its values will automatically be deleted.
+     *
+     * @param key - map key
+     * @param timeToLive - timeout before key will be deleted
+     * @param timeUnit - timeout time unit
+     * @return A Single that will emit <code>true</code> if key exists and the timeout was set and <code>false</code>
+     * if key not exists
+     */
+    Single<Boolean> expireKey(K key, long timeToLive, TimeUnit timeUnit);
+}

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

+/**
+ * Copyright (c) 2013-2020 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.api;
+
+/**
+ * Rx-ified version of {@link RSetMultimapCache}.
+ *
+ * @author Marnix Kammer
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public interface RSetMultimapCacheRx<K, V> extends RSetMultimapRx<K, V>, RMultimapCacheRx<K, V> {
+
+}

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

@@ -367,6 +367,31 @@ public interface RedissonRxClient {
      */
     <K, V> RListMultimapRx<K, V> getListMultimap(String name, Codec codec);
 
+    /**
+     * Returns List based Multimap cache instance by name.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular list multimap {@link #getListMultimap(String)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @return RListMultimapCacheRx object
+     */
+    <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name);
+
+    /**
+     * Returns List based Multimap cache instance by name using provided codec for both map keys and values.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular list multimap {@link #getListMultimap(String, Codec)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param codec - codec for keys and values
+     * @return RListMultimapCacheRx object
+     */
+    <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name, Codec codec);
+
     /**
      * Returns Set based Multimap instance by name.
      * 
@@ -388,7 +413,32 @@ public interface RedissonRxClient {
      * @return SetMultimap object
      */
     <K, V> RSetMultimapRx<K, V> getSetMultimap(String name, Codec codec);
-    
+
+    /**
+     * Returns Set based Multimap cache instance by name.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular set multimap {@link #getSetMultimap(String)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @return RSetMultimapCacheRx object
+     */
+    <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name);
+
+    /**
+     * Returns Set based Multimap cache instance by name using provided codec for both map keys and values.
+     * Supports key eviction by specifying a time to live.
+     * If eviction is not required then it's better to use regular set multimap {@link #getSetMultimap(String, Codec)}.
+     *
+     * @param <K> type of key
+     * @param <V> type of value
+     * @param name - name of object
+     * @param codec - codec for keys and values
+     * @return RSetMultimapCacheRx object
+     */
+    <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name, Codec codec);
+
     /**
      * Returns map instance by name.
      *

redisson/src/main/java/org/redisson/rx/RedissonBatchRx.java

@@ -263,26 +263,58 @@ public class RedissonBatchRx implements RBatchRx {
 
     @Override
     public <K, V> RSetMultimapRx<K, V> getSetMultimap(String name) {
-        return RxProxyBuilder.create(executorService, new RedissonSetMultimap<K, V>(executorService, name), 
-                new RedissonSetMultimapRx<K, V>(executorService, name, null), RSetMultimapRx.class);
+        RedissonSetMultimap<K, V> setMultimap = new RedissonSetMultimap<>(executorService, name);
+        return RxProxyBuilder.create(executorService, setMultimap,
+                new RedissonSetMultimapRx<K, V>(setMultimap, executorService, null), RSetMultimapRx.class);
     }
 
     @Override
     public <K, V> RSetMultimapRx<K, V> getSetMultimap(String name, Codec codec) {
-        return RxProxyBuilder.create(executorService, new RedissonSetMultimap<K, V>(codec, executorService, name), 
-                new RedissonSetMultimapRx<K, V>(codec, executorService, name, null), RSetMultimapRx.class);
+        RedissonSetMultimap<K, V> setMultimap = new RedissonSetMultimap<>(codec, executorService, name);
+        return RxProxyBuilder.create(executorService, setMultimap,
+                new RedissonSetMultimapRx<K, V>(setMultimap, executorService, null), RSetMultimapRx.class);
+    }
+
+    @Override
+    public <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name) {
+        RedissonSetMultimapCache<K, V> setMultimap = new RedissonSetMultimapCache<>(evictionScheduler, executorService, name);
+        return RxProxyBuilder.create(executorService, setMultimap,
+                new RedissonSetMultimapCacheRx<K, V>(setMultimap, executorService, null), RSetMultimapCacheRx.class);
+    }
+
+    @Override
+    public <K, V> RSetMultimapCacheRx<K, V> getSetMultimapCache(String name, Codec codec) {
+        RedissonSetMultimapCache<K, V> setMultimap = new RedissonSetMultimapCache<>(evictionScheduler, codec, executorService, name);
+        return RxProxyBuilder.create(executorService, setMultimap,
+                new RedissonSetMultimapCacheRx<K, V>(setMultimap, executorService, null), RSetMultimapCacheRx.class);
     }
 
     @Override
     public <K, V> RListMultimapRx<K, V> getListMultimap(String name) {
-        return RxProxyBuilder.create(executorService, new RedissonListMultimap<K, V>(executorService, name), 
-                new RedissonListMultimapRx<K, V>(executorService, name), RListMultimapRx.class);
+        RedissonListMultimap<K, V> listMultimap = new RedissonListMultimap<>(executorService, name);
+        return RxProxyBuilder.create(executorService, listMultimap,
+                new RedissonListMultimapRx<K, V>(listMultimap, executorService), RListMultimapRx.class);
     }
 
     @Override
     public <K, V> RListMultimapRx<K, V> getListMultimap(String name, Codec codec) {
-        return RxProxyBuilder.create(executorService, new RedissonListMultimap<K, V>(codec, executorService, name), 
-                new RedissonListMultimapRx<K, V>(codec, executorService, name), RListMultimapRx.class);
+        RedissonListMultimap<K, V> listMultimap = new RedissonListMultimap<>(codec, executorService, name);
+        return RxProxyBuilder.create(executorService, listMultimap,
+                new RedissonListMultimapRx<K, V>(listMultimap, executorService), RListMultimapRx.class);
+    }
+
+    @Override
+    public <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name) {
+        RedissonListMultimapCache<K, V> listMultimap = new RedissonListMultimapCache<>(evictionScheduler, executorService, name);
+        return RxProxyBuilder.create(executorService, listMultimap,
+                new RedissonListMultimapCacheRx<K, V>(listMultimap, executorService), RListMultimapCacheRx.class);
+    }
+
+    @Override
+    public <K, V> RListMultimapCacheRx<K, V> getListMultimapCache(String name, Codec codec) {
+        RedissonListMultimapCache<K, V> listMultimap = new RedissonListMultimapCache<>(evictionScheduler, codec, executorService, name);
+        return RxProxyBuilder.create(executorService, listMultimap,
+                new RedissonListMultimapCacheRx<K, V>(listMultimap, executorService), RListMultimapCacheRx.class);
     }
 
     @Override

redisson/src/main/java/org/redisson/rx/RedissonListMultimapCacheRx.java

+/**
+ * Copyright (c) 2013-2020 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.rx;
+
+import org.redisson.RedissonList;
+import org.redisson.RedissonListMultimapCache;
+import org.redisson.api.RListRx;
+
+/**
+ *
+ * @author Marnix Kammer
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public class RedissonListMultimapCacheRx<K, V> {
+
+    private final RedissonListMultimapCache<K, V> instance;
+    private final CommandRxExecutor commandExecutor;
+
+    public RedissonListMultimapCacheRx(RedissonListMultimapCache<K, V> instance, CommandRxExecutor commandExecutor) {
+        this.instance = instance;
+        this.commandExecutor = commandExecutor;
+    }
+
+    public RListRx<V> get(K key) {
+        RedissonList<V> list = (RedissonList<V>) instance.get(key);
+        return RxProxyBuilder.create(commandExecutor, list, new RedissonListRx<>(list), RListRx.class);
+    }
+}

redisson/src/main/java/org/redisson/rx/RedissonListMultimapRx.java

@@ -17,9 +17,7 @@ package org.redisson.rx;
 
 import org.redisson.RedissonList;
 import org.redisson.RedissonListMultimap;
-import org.redisson.api.RListMultimap;
 import org.redisson.api.RListRx;
-import org.redisson.client.codec.Codec;
 
 /**
  * 
@@ -33,18 +31,13 @@ public class RedissonListMultimapRx<K, V> {
     private final CommandRxExecutor commandExecutor;
     private final RedissonListMultimap<K, V> instance;
     
-    public RedissonListMultimapRx(CommandRxExecutor commandExecutor, String name) {
-        this.instance = new RedissonListMultimap<K, V>(commandExecutor, name);
-        this.commandExecutor = commandExecutor;
-    }
-
-    public RedissonListMultimapRx(Codec codec, CommandRxExecutor commandExecutor, String name) {
-        this.instance = new RedissonListMultimap<K, V>(codec, commandExecutor, name);
+    public RedissonListMultimapRx(RedissonListMultimap<K, V> instance, CommandRxExecutor commandExecutor) {
+        this.instance = instance;
         this.commandExecutor = commandExecutor;
     }
 
     public RListRx<V> get(K key) {
-        RedissonList<V> list = (RedissonList<V>) ((RListMultimap<K, V>) instance).get(key);
+        RedissonList<V> list = (RedissonList<V>) instance.get(key);
         return RxProxyBuilder.create(commandExecutor, instance, 
                 new RedissonListRx<V>(list), RListRx.class);
     }

redisson/src/main/java/org/redisson/rx/RedissonSetMultimapCacheRx.java

+/**
+ * Copyright (c) 2013-2020 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.rx;
+
+import org.redisson.RedissonSet;
+import org.redisson.RedissonSetMultimapCache;
+import org.redisson.api.RSetRx;
+import org.redisson.api.RedissonRxClient;
+
+/**
+ *
+ * @author Marnix Kammer
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public class RedissonSetMultimapCacheRx<K, V> {
+
+    private final RedissonSetMultimapCache<K, V> instance;
+    private final CommandRxExecutor commandExecutor;
+    private final RedissonRxClient redisson;
+
+    public RedissonSetMultimapCacheRx(RedissonSetMultimapCache<K, V> instance, CommandRxExecutor commandExecutor,
+                                      RedissonRxClient redisson) {
+        this.instance = instance;
+        this.redisson = redisson;
+        this.commandExecutor = commandExecutor;
+    }
+
+    public RSetRx<V> get(K key) {
+        RedissonSet<V> set = (RedissonSet<V>) instance.get(key);
+        return RxProxyBuilder.create(commandExecutor, set, new RedissonSetRx<>(set, redisson), RSetRx.class);
+    }
+}

redisson/src/main/java/org/redisson/rx/RedissonSetMultimapRx.java

@@ -15,12 +15,10 @@
  */
 package org.redisson.rx;
 
-import org.redisson.RedissonListMultimap;
 import org.redisson.RedissonSet;
-import org.redisson.api.RSetMultimap;
+import org.redisson.RedissonSetMultimap;
 import org.redisson.api.RSetRx;
 import org.redisson.api.RedissonRxClient;
-import org.redisson.client.codec.Codec;
 
 /**
  * 
@@ -33,22 +31,16 @@ public class RedissonSetMultimapRx<K, V> {
 
     private final RedissonRxClient redisson;
     private final CommandRxExecutor commandExecutor;
-    private final RedissonListMultimap<K, V> instance;
+    private final RedissonSetMultimap<K, V> instance;
     
-    public RedissonSetMultimapRx(CommandRxExecutor commandExecutor, String name, RedissonRxClient redisson) {
-        this.instance = new RedissonListMultimap<K, V>(commandExecutor, name);
-        this.redisson = redisson;
-        this.commandExecutor = commandExecutor;
-    }
-
-    public RedissonSetMultimapRx(Codec codec, CommandRxExecutor commandExecutor, String name, RedissonRxClient redisson) {
-        this.instance = new RedissonListMultimap<K, V>(codec, commandExecutor, name);
+    public RedissonSetMultimapRx(RedissonSetMultimap<K, V> instance, CommandRxExecutor commandExecutor, RedissonRxClient redisson) {
+        this.instance = instance;
         this.redisson = redisson;
         this.commandExecutor = commandExecutor;
     }
 
     public RSetRx<V> get(K key) {
-        RedissonSet<V> set = (RedissonSet<V>) ((RSetMultimap<K, V>) instance).get(key);
+        RedissonSet<V> set = (RedissonSet<V>) instance.get(key);
         return RxProxyBuilder.create(commandExecutor, set, 
                 new RedissonSetRx<V>(set, redisson), RSetRx.class);
     }