refactoring

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

@@ -173,7 +173,6 @@ public class CommandBatchService extends CommandAsyncService {
         return executeAsync(BatchOptions.defaults());
     }
     
-    @SuppressWarnings("MethodLength")
     public <R> RFuture<R> executeAsync(BatchOptions options) {
         if (executed.get()) {
             throw new IllegalStateException("Batch already executed!");
@@ -190,6 +189,108 @@ public class CommandBatchService extends CommandAsyncService {
         }
         
         if (isRedisBasedQueue()) {
+            return executeRedisBasedQueue();
+        }
+
+        if (this.options.getExecutionMode() != ExecutionMode.IN_MEMORY) {
+            for (Entry entry : commands.values()) {
+                BatchCommandData<?, ?> multiCommand = new BatchCommandData(RedisCommands.MULTI, new Object[] {}, index.incrementAndGet());
+                entry.getCommands().addFirst(multiCommand);
+                BatchCommandData<?, ?> execCommand = new BatchCommandData(RedisCommands.EXEC, new Object[] {}, index.incrementAndGet());
+                entry.getCommands().add(execCommand);
+            }
+        }
+        
+        if (this.options.isSkipResult()) {
+            for (Entry entry : commands.values()) {
+                BatchCommandData<?, ?> offCommand = new BatchCommandData(RedisCommands.CLIENT_REPLY, new Object[] { "OFF" }, index.incrementAndGet());
+                entry.getCommands().addFirst(offCommand);
+                BatchCommandData<?, ?> onCommand = new BatchCommandData(RedisCommands.CLIENT_REPLY, new Object[] { "ON" }, index.incrementAndGet());
+                entry.getCommands().add(onCommand);
+            }
+        }
+        
+        if (this.options.getSyncSlaves() > 0) {
+            for (Entry entry : commands.values()) {
+                BatchCommandData<?, ?> waitCommand = new BatchCommandData(RedisCommands.WAIT, 
+                                    new Object[] { this.options.getSyncSlaves(), this.options.getSyncTimeout() }, index.incrementAndGet());
+                entry.getCommands().add(waitCommand);
+            }
+        }
+        
+        RPromise<R> resultPromise;
+        RPromise<Void> voidPromise = new RedissonPromise<Void>();
+        if (this.options.isSkipResult()) {
+            voidPromise.onComplete((res, e) -> {
+//              commands = null;
+                executed.set(true);
+                nestedServices.clear();
+            });
+            resultPromise = (RPromise<R>) voidPromise;
+        } else {
+            RPromise<Object> promise = new RedissonPromise<Object>();
+            voidPromise.onComplete((res, ex) -> {
+                executed.set(true);
+                if (ex != null) {
+                    promise.tryFailure(ex);
+                    commands = null;
+                    nestedServices.clear();
+                    return;
+                }
+                
+                List<BatchCommandData> entries = new ArrayList<BatchCommandData>();
+                for (Entry e : commands.values()) {
+                    entries.addAll(e.getCommands());
+                }
+                Collections.sort(entries);
+                List<Object> responses = new ArrayList<Object>(entries.size());
+                int syncedSlaves = 0;
+                for (BatchCommandData<?, ?> commandEntry : entries) {
+                    if (isWaitCommand(commandEntry)) {
+                        syncedSlaves = (Integer) commandEntry.getPromise().getNow();
+                    } else if (!commandEntry.getCommand().getName().equals(RedisCommands.MULTI.getName())
+                            && !commandEntry.getCommand().getName().equals(RedisCommands.EXEC.getName())) {
+                        Object entryResult = commandEntry.getPromise().getNow();
+                        try {
+                            entryResult = RedisExecutor.tryHandleReference(objectBuilder, entryResult);
+                        } catch (ReflectiveOperationException exc) {
+                            log.error("Unable to handle reference from " + entryResult, exc);
+                        }
+                        responses.add(entryResult);
+                    }
+                }
+                
+                BatchResult<Object> result = new BatchResult<Object>(responses, syncedSlaves);
+                promise.trySuccess(result);
+                
+                commands = null;
+                nestedServices.clear();
+            });
+            resultPromise = (RPromise<R>) promise;
+        }
+
+        AtomicInteger slots = new AtomicInteger(commands.size());
+        
+        for (Map.Entry<RFuture<?>, List<CommandBatchService>> entry : nestedServices.entrySet()) {
+            slots.incrementAndGet();
+            for (CommandBatchService service : entry.getValue()) {
+                service.executeAsync();
+            }
+            
+            entry.getKey().onComplete((res, e) -> {
+                handle(voidPromise, slots, entry.getKey());
+            });
+        }
+        
+        for (Map.Entry<MasterSlaveEntry, Entry> e : commands.entrySet()) {
+            RedisCommonBatchExecutor executor = new RedisCommonBatchExecutor(new NodeSource(e.getKey()), voidPromise, 
+                                                    connectionManager, this.options, e.getValue(), slots);
+            executor.execute();
+        }
+        return resultPromise;
+    }
+
+    private <R> RFuture<R> executeRedisBasedQueue() {
         int permits = 0;
         for (Entry entry : commands.values()) {
             permits += entry.getCommands().size();
@@ -299,110 +400,9 @@ public class CommandBatchService extends CommandAsyncService {
         return resultPromise;
     }
 
-        if (this.options.getExecutionMode() != ExecutionMode.IN_MEMORY) {
-            for (Entry entry : commands.values()) {
-                BatchCommandData<?, ?> multiCommand = new BatchCommandData(RedisCommands.MULTI, new Object[] {}, index.incrementAndGet());
-                entry.getCommands().addFirst(multiCommand);
-                BatchCommandData<?, ?> execCommand = new BatchCommandData(RedisCommands.EXEC, new Object[] {}, index.incrementAndGet());
-                entry.getCommands().add(execCommand);
-            }
-        }
-        
-        if (this.options.isSkipResult()) {
-            for (Entry entry : commands.values()) {
-                BatchCommandData<?, ?> offCommand = new BatchCommandData(RedisCommands.CLIENT_REPLY, new Object[] { "OFF" }, index.incrementAndGet());
-                entry.getCommands().addFirst(offCommand);
-                BatchCommandData<?, ?> onCommand = new BatchCommandData(RedisCommands.CLIENT_REPLY, new Object[] { "ON" }, index.incrementAndGet());
-                entry.getCommands().add(onCommand);
-            }
-        }
-        
-        if (this.options.getSyncSlaves() > 0) {
-            for (Entry entry : commands.values()) {
-                BatchCommandData<?, ?> waitCommand = new BatchCommandData(RedisCommands.WAIT, 
-                                    new Object[] { this.options.getSyncSlaves(), this.options.getSyncTimeout() }, index.incrementAndGet());
-                entry.getCommands().add(waitCommand);
-            }
-        }
-        
-        RPromise<R> resultPromise;
-        RPromise<Void> voidPromise = new RedissonPromise<Void>();
-        if (this.options.isSkipResult()) {
-            voidPromise.onComplete((res, e) -> {
-//              commands = null;
-                executed.set(true);
-                nestedServices.clear();
-            });
-            resultPromise = (RPromise<R>) voidPromise;
-        } else {
-            RPromise<Object> promise = new RedissonPromise<Object>();
-            voidPromise.onComplete((res, ex) -> {
-                executed.set(true);
-                if (ex != null) {
-                    promise.tryFailure(ex);
-                    commands = null;
-                    nestedServices.clear();
-                    return;
-                }
-                
-                List<BatchCommandData> entries = new ArrayList<BatchCommandData>();
-                for (Entry e : commands.values()) {
-                    entries.addAll(e.getCommands());
-                }
-                Collections.sort(entries);
-                List<Object> responses = new ArrayList<Object>(entries.size());
-                int syncedSlaves = 0;
-                for (BatchCommandData<?, ?> commandEntry : entries) {
-                    if (isWaitCommand(commandEntry)) {
-                        syncedSlaves = (Integer) commandEntry.getPromise().getNow();
-                    } else if (!commandEntry.getCommand().getName().equals(RedisCommands.MULTI.getName())
-                            && !commandEntry.getCommand().getName().equals(RedisCommands.EXEC.getName())) {
-                        Object entryResult = commandEntry.getPromise().getNow();
-                        try {
-                            entryResult = RedisExecutor.tryHandleReference(objectBuilder, entryResult);
-                        } catch (ReflectiveOperationException exc) {
-                            log.error("Unable to handle reference from " + entryResult, exc);
-                        }
-                        responses.add(entryResult);
-                    }
-                }
-                
-                BatchResult<Object> result = new BatchResult<Object>(responses, syncedSlaves);
-                promise.trySuccess(result);
-                
-                commands = null;
-                nestedServices.clear();
-            });
-            resultPromise = (RPromise<R>) promise;
-        }
-
-        AtomicInteger slots = new AtomicInteger(commands.size());
-        
-        for (java.util.Map.Entry<RFuture<?>, List<CommandBatchService>> entry : nestedServices.entrySet()) {
-            slots.incrementAndGet();
-            for (CommandBatchService service : entry.getValue()) {
-                service.executeAsync();
-            }
-            
-            entry.getKey().onComplete((res, e) -> {
-                handle(voidPromise, slots, entry.getKey());
-            });
-        }
-        
-        for (java.util.Map.Entry<MasterSlaveEntry, Entry> e : commands.entrySet()) {
-            execute(e.getValue(), new NodeSource(e.getKey()), voidPromise, slots, this.options);
-        }
-        return resultPromise;
-    }
-
     protected boolean isRedisBasedQueue() {
-        return options != null && (this.options.getExecutionMode() == ExecutionMode.REDIS_READ_ATOMIC || this.options.getExecutionMode() == ExecutionMode.REDIS_WRITE_ATOMIC);
-    }
-
-    private void execute(Entry entry, NodeSource source, RPromise<Void> mainPromise, AtomicInteger slots, 
-            BatchOptions options) {
-        RedisCommonBatchExecutor executor = new RedisCommonBatchExecutor(source, mainPromise, connectionManager, options, entry, slots);
-        executor.execute();
+        return options != null && (options.getExecutionMode() == ExecutionMode.REDIS_READ_ATOMIC 
+                                    || options.getExecutionMode() == ExecutionMode.REDIS_WRITE_ATOMIC);
     }
 
     protected boolean isWaitCommand(CommandData<?, ?> c) {

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

@@ -17,12 +17,11 @@ package org.redisson.command;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.redisson.api.BatchOptions;
-import org.redisson.api.RFuture;
 import org.redisson.api.BatchOptions.ExecutionMode;
+import org.redisson.api.RFuture;
 import org.redisson.client.RedisConnection;
 import org.redisson.client.codec.StringCodec;
 import org.redisson.client.protocol.CommandData;

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

@@ -73,7 +73,7 @@ import io.netty.util.concurrent.FutureListener;
  * @param <V> type of value
  * @param <R> type of returned value
  */
-@SuppressWarnings({"NestedIfDepth", "MethodLength"})
+@SuppressWarnings({"NestedIfDepth"})
 public class RedisExecutor<V, R> {
     
     static final Logger log = LoggerFactory.getLogger(RedisExecutor.class);
@@ -152,6 +152,48 @@ public class RedisExecutor<V, R> {
             });
         }
 
+        scheduleRetryTimeout(connectionFuture, attemptPromise);
+
+        connectionFuture.onComplete((connection, e) -> {
+            if (connectionFuture.isCancelled()) {
+                connectionManager.getShutdownLatch().release();
+                return;
+            }
+
+            if (!connectionFuture.isSuccess()) {
+                connectionManager.getShutdownLatch().release();
+                exception = convertException(connectionFuture);
+                return;
+            }
+
+            if (attemptPromise.isDone() || mainPromise.isDone()) {
+                releaseConnection(attemptPromise, connectionFuture);
+                return;
+            }
+
+            sendCommand(attemptPromise, connection);
+
+            writeFuture.addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture future) throws Exception {
+                    checkWriteFuture(writeFuture, attemptPromise, connection);
+                }
+            });
+
+            releaseConnection(attemptPromise, connectionFuture);
+        });
+
+        attemptPromise.onComplete((r, e) -> {
+            checkAttemptPromise(attemptPromise, connectionFuture);
+        });
+    }
+
+    private void scheduleRetryTimeout(RFuture<RedisConnection> connectionFuture, RPromise<R> attemptPromise) {
+        if (retryInterval == 0 || attempts == 0) {
+            this.timeout = MasterSlaveConnectionManager.DUMMY_TIMEOUT;
+            return;
+        }
+        
         TimerTask retryTimerTask = new TimerTask() {
 
             @Override
@@ -233,44 +275,7 @@ public class RedisExecutor<V, R> {
 
         };
 
-        if (retryInterval > 0 && attempts > 0) {
-            this.timeout = connectionManager.newTimeout(retryTimerTask, retryInterval, TimeUnit.MILLISECONDS);
-        } else {
-            this.timeout = MasterSlaveConnectionManager.DUMMY_TIMEOUT;            
-        }
-
-        connectionFuture.onComplete((connection, e) -> {
-            if (connectionFuture.isCancelled()) {
-                connectionManager.getShutdownLatch().release();
-                return;
-            }
-
-            if (!connectionFuture.isSuccess()) {
-                connectionManager.getShutdownLatch().release();
-                exception = convertException(connectionFuture);
-                return;
-            }
-
-            if (attemptPromise.isDone() || mainPromise.isDone()) {
-                releaseConnection(attemptPromise, connectionFuture);
-                return;
-            }
-
-            sendCommand(attemptPromise, connection);
-
-            writeFuture.addListener(new ChannelFutureListener() {
-                @Override
-                public void operationComplete(ChannelFuture future) throws Exception {
-                    checkWriteFuture(writeFuture, attemptPromise, connection);
-                }
-            });
-
-            releaseConnection(attemptPromise, connectionFuture);
-        });
-
-        attemptPromise.onComplete((r, e) -> {
-            checkAttemptPromise(attemptPromise, connectionFuture);
-        });
+        timeout = connectionManager.newTimeout(retryTimerTask, retryInterval, TimeUnit.MILLISECONDS);
     }
     
     protected void free() {