Redisson Reference support for batch, reactive and reactive batch

redisson/src/main/java/org/redisson/Redisson.java

@@ -159,7 +159,11 @@ public static RedissonReactiveClient createReactive() {
      * @return Redisson instance
      */
     public static RedissonReactiveClient createReactive(Config config) {
-        return new RedissonReactive(config);
+        RedissonReactive react = new RedissonReactive(config);
+        if (config.isRedissonReferenceEnabled()) {
+            react.enableRedissonReferenceSupport();
+        }
+        return react;
     }
 
     @Override
@@ -489,7 +493,11 @@ public RKeys getKeys() {
 
     @Override
     public RBatch createBatch() {
-        return new RedissonBatch(evictionScheduler, connectionManager);
+        RedissonBatch batch = new RedissonBatch(evictionScheduler, connectionManager);
+        if (config.isRedissonReferenceEnabled()) {
+            batch.enableRedissonReferenceSupport(this);
+        }
+        return batch;
     }
 
     @Override

redisson/src/main/java/org/redisson/RedissonBatch.java

@@ -56,7 +56,7 @@ public class RedissonBatch implements RBatch {
     private final EvictionScheduler evictionScheduler;
     private final CommandBatchService executorService;
 
-    public RedissonBatch(EvictionScheduler evictionScheduler, ConnectionManager connectionManager) {
+    protected RedissonBatch(EvictionScheduler evictionScheduler, ConnectionManager connectionManager) {
         this.executorService = new CommandBatchService(connectionManager);
         this.evictionScheduler = evictionScheduler;
     }
@@ -281,5 +281,8 @@ public <K, V> RMultimapCacheAsync<K, V> getListMultimapCache(String name, Codec
         return new RedissonListMultimapCache<K, V>(evictionScheduler, codec, executorService, name);
     }
 
+    protected void enableRedissonReferenceSupport(Redisson redisson) {
+        this.executorService.enableRedissonReferenceSupport(redisson);
+    }
 
 }

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

@@ -45,6 +45,7 @@
 import org.redisson.api.RedissonReactiveClient;
 import org.redisson.client.codec.Codec;
 import org.redisson.client.protocol.RedisCommands;
+import org.redisson.codec.CodecProvider;
 import org.redisson.command.CommandReactiveService;
 import org.redisson.config.Config;
 import org.redisson.config.ConfigSupport;
@@ -82,14 +83,16 @@ public class RedissonReactive implements RedissonReactiveClient {
     protected final CommandReactiveService commandExecutor;
     protected final ConnectionManager connectionManager;
     protected final Config config;
-
+    protected final CodecProvider codecProvider;
+
     protected RedissonReactive(Config config) {
         this.config = config;
         Config configCopy = new Config(config);
 
         connectionManager = ConfigSupport.createConnectionManager(configCopy);
         commandExecutor = new CommandReactiveService(connectionManager);
         evictionScheduler = new EvictionScheduler(commandExecutor);
+        codecProvider = config.getCodecProvider();
     }
 
 
@@ -259,7 +262,11 @@ public RScriptReactive getScript() {
 
     @Override
     public RBatchReactive createBatch() {
-        return new RedissonBatchReactive(evictionScheduler, connectionManager);
+        RedissonBatchReactive batch = new RedissonBatchReactive(evictionScheduler, connectionManager);
+        if (config.isRedissonReferenceEnabled()) {
+            batch.enableRedissonReferenceSupport(this);
+        }
+        return batch;
     }
 
     @Override
@@ -272,6 +279,11 @@ public Config getConfig() {
         return config;
     }
 
+    @Override
+    public CodecProvider getCodecProvider() {
+        return codecProvider;
+    }
+
     @Override
     public NodesGroup<Node> getNodesGroup() {
         return new RedisNodes<Node>(connectionManager);
@@ -300,5 +312,8 @@ public boolean isShuttingDown() {
         return connectionManager.isShuttingDown();
     }
 
+    protected void enableRedissonReferenceSupport() {
+        this.commandExecutor.enableRedissonReferenceSupport(this);
+    }
 }
 

redisson/src/main/java/org/redisson/RedissonReference.java

@@ -17,6 +17,7 @@
 
 import org.redisson.client.codec.Codec;
 import org.redisson.api.RObject;
+import org.redisson.api.RObjectReactive;
 import org.redisson.api.annotation.REntity;
 
 /**
@@ -37,12 +38,14 @@ public RedissonReference(Class type, String keyName) {
     }
 
     public RedissonReference(Class type, String keyName, Codec codec) {
-        if (!type.isAnnotationPresent(REntity.class) && !RObject.class.isAssignableFrom(type)) {
-            throw new IllegalArgumentException("Class reference has to be a type of either RObject or RLiveObject");
+        if (!type.isAnnotationPresent(REntity.class) && !RObject.class.isAssignableFrom(type) && !RObjectReactive.class.isAssignableFrom(type)) {
+            throw new IllegalArgumentException("Class reference has to be a type of either RObject or RLiveObject or RObjectReactive");
         }
-        this.type = type.getName();
+        this.type = RObjectReactive.class.isAssignableFrom(type)
+                ? type.getName().substring(0, type.getName().length() - "Reactive".length()).replaceFirst(".reactive", "")
+                : type.getName();
         this.keyName = keyName;
-        this.codec = codec != null ? codec.getClass().getCanonicalName() : null;
+        this.codec = codec != null ? codec.getClass().getName() : null;
     }
 
     public boolean isDefaultCodec() {
@@ -60,21 +63,39 @@ public Class<?> getType() throws Exception {
         return Class.forName(type);
     }
 
+    /**
+     * @return the type
+     * @throws java.lang.Exception - which could be:
+     *     LinkageError - if the linkage fails
+     *     ExceptionInInitializerError - if the initialization provoked by this method fails
+     *     ClassNotFoundException - if the class cannot be located
+     */
+    public Class<?> getReactiveType() throws Exception {
+        return Class.forName(type.replaceFirst("org.redisson", "org.redisson.reactive") + "Reactive");//live object is not supported in reactive client
+    }
+
     /**
      * @return type name in string
      */
     public String getTypeName() {
         return type;
     }
 
+    /**
+     * @return type name in string
+     */
+    public String getReactiveTypeName() {
+        return type + "Reactive";
+    }
+
     /**
      * @param type the type to set
      */
     public void setType(Class<?> type) {
-        if (!type.isAnnotationPresent(REntity.class) && !RObject.class.isAssignableFrom(type)) {
-            throw new IllegalArgumentException("Class reference has to be a type of either RObject or RLiveObject");
+        if (!type.isAnnotationPresent(REntity.class) && (!RObject.class.isAssignableFrom(type) || !RObjectReactive.class.isAssignableFrom(type))) {
+            throw new IllegalArgumentException("Class reference has to be a type of either RObject or RLiveObject or RObjectReactive");
         }
-        this.type = type.getCanonicalName();
+        this.type = type.getName();
     }
 
     /**
@@ -115,7 +136,7 @@ public String getCodecName() {
      * @param codec the codec to set
      */
     public void setCodecType(Class<? extends Codec> codec) {
-        this.codec = codec.getCanonicalName();
+        this.codec = codec.getName();
     }
 
 }

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

@@ -16,6 +16,7 @@
 package org.redisson.api;
 
 import org.reactivestreams.Publisher;
+import org.redisson.client.codec.Codec;
 
 /**
  * Base interface for all Redisson objects
@@ -26,7 +27,8 @@
 public interface RObjectReactive {
 
     String getName();
-
+
+    Codec getCodec();
     /**
      * Transfer a object from a source Redis instance to a destination Redis instance
      * in  mode

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

@@ -18,6 +18,7 @@
 import java.util.List;
 
 import org.redisson.client.codec.Codec;
+import org.redisson.codec.CodecProvider;
 import org.redisson.config.Config;
 
 /**
@@ -356,7 +357,14 @@ public interface RedissonReactiveClient {
      * @return Config object
      */
     Config getConfig();
-
+
+    /**
+     * Returns the CodecProvider instance
+     * 
+     * @return CodecProvider
+     */
+    CodecProvider getCodecProvider();
+
     /**
      * Get Redis nodes group for server operations
      *

redisson/src/main/java/org/redisson/command/CommandAsyncExecutor.java

@@ -28,8 +28,8 @@
 import org.redisson.connection.ConnectionManager;
 import org.redisson.connection.MasterSlaveEntry;
 
-import io.netty.util.concurrent.Future;
-import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.api.RedissonReactiveClient;
 
 /**
  *
@@ -40,8 +40,12 @@ public interface CommandAsyncExecutor {
 
     ConnectionManager getConnectionManager();
 
-    CommandAsyncExecutor enableRedissonReferenceSupport(Redisson redisson);
-
+    CommandAsyncExecutor enableRedissonReferenceSupport(RedissonClient redisson);
+
+    CommandAsyncExecutor enableRedissonReferenceSupport(RedissonReactiveClient redissonReactive);
+
+    boolean isRedissonReferenceSupportEnabled();
+
     <V> RedisException convertException(RFuture<V> RFuture);
 
     boolean await(RFuture<?> RFuture, long timeout, TimeUnit timeoutUnit) throws InterruptedException;

redisson/src/main/java/org/redisson/command/CommandAsyncService.java

@@ -61,8 +61,11 @@
 import io.netty.util.concurrent.FutureListener;
 import io.netty.util.concurrent.Promise;
 import org.redisson.Redisson;
+import org.redisson.RedissonReactive;
 import org.redisson.RedissonReference;
 import org.redisson.api.RObject;
+import org.redisson.api.RedissonClient;
+import org.redisson.api.RedissonReactiveClient;
 import org.redisson.liveobject.misc.RedissonObjectFactory;
 
 /**
@@ -75,7 +78,8 @@ public class CommandAsyncService implements CommandAsyncExecutor {
     private static final Logger log = LoggerFactory.getLogger(CommandAsyncService.class);
 
     final ConnectionManager connectionManager;
-    private Redisson redisson;
+    protected RedissonClient redisson;
+    protected RedissonReactiveClient redissonReactive;
 
     public CommandAsyncService(ConnectionManager connectionManager) {
         this.connectionManager = connectionManager;
@@ -87,13 +91,28 @@ public ConnectionManager getConnectionManager() {
     }
 
     @Override
-    public CommandAsyncExecutor enableRedissonReferenceSupport(Redisson redisson) {
+    public CommandAsyncExecutor enableRedissonReferenceSupport(RedissonClient redisson) {
         if (redisson != null) {
             this.redisson = redisson;
+            this.redissonReactive = null;
+        }
+        return this;
+    }
+
+    @Override
+    public CommandAsyncExecutor enableRedissonReferenceSupport(RedissonReactiveClient redissonReactive) {
+        if (redissonReactive != null) {
+            this.redissonReactive = redissonReactive;
+            this.redisson = null;
         }
         return this;
     }
 
+    @Override
+    public boolean isRedissonReferenceSupportEnabled() {
+        return redisson != null || redissonReactive != null;
+    }
+
     @Override
     public <V> V get(RFuture<V> future) {
         final CountDownLatch l = new CountDownLatch(1);
@@ -450,9 +469,11 @@ protected <V, R> void async(final boolean readOnlyMode, final NodeSource source,
         }
 
         final AsyncDetails<V, R> details = AsyncDetails.acquire();
-        if (redisson != null) {
+        if (isRedissonReferenceSupportEnabled()) {
             for (int i = 0; i < params.length; i++) {
-                RedissonReference reference = RedissonObjectFactory.toReference(redisson, params[i]);
+                RedissonReference reference = redisson != null
+                        ? RedissonObjectFactory.toReference(redisson, params[i])
+                        : RedissonObjectFactory.toReference(redissonReactive, params[i]);
                 params[i] = reference == null ? params[i] : reference;
             }
         }
@@ -730,9 +751,25 @@ private <R, V> void checkAttemptFuture(final NodeSource source, final AsyncDetai
                 }
                 ((RedisClientResult)res).setRedisClient(addr);
             }
-            if (redisson != null && res instanceof RedissonReference) {
+
+            if (isRedissonReferenceSupportEnabled() && res instanceof List) {
+                List r = (List) res;
+                for (int i = 0; i < r.size(); i++) {
+                    if (r.get(i) instanceof RedissonReference) {
+                        try {
+                            r.set(i ,(redisson != null
+                                    ? RedissonObjectFactory.<R>fromReference(redisson, (RedissonReference) r.get(i))
+                                    : RedissonObjectFactory.<R>fromReference(redissonReactive, (RedissonReference) r.get(i))));
+                        } catch (Exception exception) {//skip and carry on to next one.
+                        }
+                    }
+                }
+                details.getMainPromise().trySuccess(res);
+            } else if (isRedissonReferenceSupportEnabled() && res instanceof RedissonReference) {
                 try {
-                    details.getMainPromise().trySuccess(RedissonObjectFactory.<R>fromReference(redisson, (RedissonReference) res));
+                    details.getMainPromise().trySuccess(redisson != null
+                            ? RedissonObjectFactory.<R>fromReference(redisson, (RedissonReference) res)
+                            : RedissonObjectFactory.<R>fromReference(redissonReactive, (RedissonReference) res));
                 } catch (Exception exception) {
                     details.getMainPromise().trySuccess(res);//fallback
                 }

redisson/src/main/java/org/redisson/command/CommandBatchService.java

@@ -50,8 +50,9 @@
 import io.netty.util.TimerTask;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
-import io.netty.util.concurrent.Promise;
 import io.netty.util.internal.PlatformDependent;
+import org.redisson.RedissonReference;
+import org.redisson.liveobject.misc.RedissonObjectFactory;
 
 public class CommandBatchService extends CommandReactiveService {
 
@@ -109,7 +110,14 @@ protected <V, R> void async(boolean readOnlyMode, NodeSource nodeSource,
         if (!readOnlyMode) {
             entry.setReadOnlyMode(false);
         }
-
+        if (isRedissonReferenceSupportEnabled()) {
+            for (int i = 0; i < params.length; i++) {
+                RedissonReference reference = redisson != null
+                        ? RedissonObjectFactory.toReference(redisson, params[i])
+                        : RedissonObjectFactory.toReference(redissonReactive, params[i]);
+                params[i] = reference == null ? params[i] : reference;
+            }
+        }
         BatchCommandData<V, R> commandData = new BatchCommandData<V, R>(mainPromise, codec, command, params, index.incrementAndGet());
         entry.getCommands().add(commandData);
     }
@@ -171,7 +179,14 @@ public void operationComplete(Future<Void> future) throws Exception {
                 Collections.sort(entries);
                 List<Object> result = new ArrayList<Object>(entries.size());
                 for (BatchCommandData<?, ?> commandEntry : entries) {
-                    result.add(commandEntry.getPromise().getNow());
+                    Object entryResult = commandEntry.getPromise().getNow();
+                    if (isRedissonReferenceSupportEnabled() && entryResult instanceof RedissonReference) {
+                        result.add(redisson != null
+                                ? RedissonObjectFactory.<Object>fromReference(redisson, (RedissonReference) entryResult)
+                                : RedissonObjectFactory.<Object>fromReference(redissonReactive, (RedissonReference) entryResult));
+                    } else {
+                        result.add(commandEntry.getPromise().getNow());
+                    }
                 }
                 promise.setSuccess(result);
                 commands = null;