Ability to cancel task in running state. #208

redisson/src/main/java/org/redisson/RedissonExecutorService.java

@@ -30,7 +30,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -106,7 +106,7 @@ public RedissonExecutorService(Codec codec, CommandExecutor commandExecutor, Red
         remoteService.setTasksCounterName(tasksCounter.getName());
         remoteService.setStatusName(status.getName());
 
-        asyncService = remoteService.get(RemoteExecutorServiceAsync.class, RemoteInvocationOptions.defaults().noAck());
+        asyncService = remoteService.get(RemoteExecutorServiceAsync.class, RemoteInvocationOptions.defaults().noAck().expectResultWithin(Integer.MAX_VALUE * 2));
         asyncServiceWithoutResult = remoteService.get(RemoteExecutorServiceAsync.class, RemoteInvocationOptions.defaults().noAck().noResult());
     }
 
@@ -116,7 +116,7 @@ public void registerWorkers(int executors) {
     }
 
     @Override
-    public void registerWorkers(int executors, Executor executor) {
+    public void registerWorkers(int executors, ExecutorService executor) {
         RemoteExecutorServiceImpl service = new RemoteExecutorServiceImpl(commandExecutor, redisson, codec, requestQueueName);
         service.setStatusName(status.getName());
         service.setTasksCounterName(tasksCounter.getName());

redisson/src/main/java/org/redisson/RedissonQueueSemaphore.java

@@ -20,13 +20,17 @@
 import java.util.Collection;
 import java.util.List;
 
-import org.redisson.client.codec.LongCodec;
 import org.redisson.client.protocol.RedisCommands;
 import org.redisson.command.CommandExecutor;
 import org.redisson.pubsub.SemaphorePubSub;
 
 import io.netty.util.concurrent.Future;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class RedissonQueueSemaphore extends RedissonSemaphore {
 
     private String queueName;

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

@@ -51,6 +51,6 @@ public interface RExecutorService extends ExecutorService {
      * 
      * @param workers - workers amount
      */
-    void registerWorkers(int workers, Executor executor);
+    void registerWorkers(int workers, ExecutorService executor);
 
 }

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

@@ -16,6 +16,7 @@
 package org.redisson.api;
 
 import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -83,7 +84,7 @@ public interface RRemoteService {
      * @param object
      * @param workersAmount
      */
-    <T> void register(Class<T> remoteInterface, T object, int workersAmount, Executor executor);
+    <T> void register(Class<T> remoteInterface, T object, int workersAmount, ExecutorService executor);
 
     /**
      * Get remote service object for remote invocations.

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

@@ -82,7 +82,7 @@ public RemoteInvocationOptions(RemoteInvocationOptions copy) {
     public static RemoteInvocationOptions defaults() {
         return new RemoteInvocationOptions()
                 .expectAckWithin(1, TimeUnit.SECONDS)
-                .expectResultWithin(20, TimeUnit.SECONDS);
+                .expectResultWithin(30, TimeUnit.SECONDS);
     }
 
     public Long getAckTimeoutInMillis() {

redisson/src/main/java/org/redisson/client/RedisConnection.java

@@ -62,10 +62,6 @@ public static <C extends RedisConnection> C getFrom(Channel channel) {
         return (C) channel.attr(RedisConnection.CONNECTION).get();
     }
 
-    public void removeCurrentCommand() {
-        channel.attr(CommandsQueue.CURRENT_COMMAND).remove();
-    }
-
     public CommandData getCurrentCommand() {
         QueueCommand command = channel.attr(CommandsQueue.CURRENT_COMMAND).get();
         if (command instanceof CommandData) {

redisson/src/main/java/org/redisson/client/RedisException.java

@@ -15,6 +15,11 @@
  */
 package org.redisson.client;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class RedisException extends RuntimeException {
 
     private static final long serialVersionUID = 3389820652701696154L;

redisson/src/main/java/org/redisson/client/handler/ConnectionWatchdog.java

@@ -160,7 +160,8 @@ private void refresh(RedisConnection connection, Channel channel) {
 
     private void reattachBlockingQueue(RedisConnection connection, final CommandData<?, ?> commandData) {
         if (commandData == null 
-                || !commandData.isBlockingCommand()) {
+                || !commandData.isBlockingCommand()
+                    || commandData.getPromise().isDone()) {
             return;
         }
 

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

@@ -89,16 +89,26 @@ public void operationComplete(Future<V> future) throws Exception {
                 l.countDown();
             }
         });
-        try {
-            l.await();
-        } catch (InterruptedException e) {
+
+        boolean interrupted = false;
+        while (!future.isDone()) {
+            try {
+                l.await();
+            } catch (InterruptedException e) {
+                interrupted = true;
+            }
+        }
+
+        if (interrupted) {
             Thread.currentThread().interrupt();
         }
+
         // commented out due to blocking issues up to 200 ms per minute for each thread
         // future.awaitUninterruptibly();
         if (future.isSuccess()) {
             return future.getNow();
         }
+
         throw convertException(future);
     }
 
@@ -509,9 +519,9 @@ private <V, R> void checkWriteFuture(final AsyncDetails<V, R> details, final Red
 
         details.getTimeout().cancel();
 
-        int timeoutTime = connectionManager.getConfig().getTimeout();
+        long timeoutTime = connectionManager.getConfig().getTimeout();
         if (QueueCommand.TIMEOUTLESS_COMMANDS.contains(details.getCommand().getName())) {
-            Integer popTimeout = Integer.valueOf(details.getParams()[details.getParams().length - 1].toString());
+            Long popTimeout = Long.valueOf(details.getParams()[details.getParams().length - 1].toString());
             handleBlockingOperations(details, connection, popTimeout);
             if (popTimeout == 0) {
                 return;
@@ -521,7 +531,7 @@ private <V, R> void checkWriteFuture(final AsyncDetails<V, R> details, final Red
             timeoutTime += 1000;
         }
 
-        final int timeoutAmount = timeoutTime;
+        final long timeoutAmount = timeoutTime;
         TimerTask timeoutTask = new TimerTask() {
             @Override
             public void run(Timeout timeout) throws Exception {
@@ -535,7 +545,7 @@ public void run(Timeout timeout) throws Exception {
         details.setTimeout(timeout);
     }
 
-    private <R, V> void handleBlockingOperations(final AsyncDetails<V, R> details, final RedisConnection connection, Integer popTimeout) {
+    private <R, V> void handleBlockingOperations(final AsyncDetails<V, R> details, final RedisConnection connection, Long popTimeout) {
         final FutureListener<Boolean> listener = new FutureListener<Boolean>() {
             @Override
             public void operationComplete(Future<Boolean> future) throws Exception {
@@ -551,7 +561,7 @@ public void operationComplete(Future<Boolean> future) throws Exception {
             scheduledFuture = connectionManager.getGroup().schedule(new Runnable() {
                 @Override
                 public void run() {
-                    // there is no re-connection was made
+                    // re-connection wasn't made
                     // and connection is still active
                     if (orignalChannel == connection.getChannel() 
                             && connection.isActive()) {
@@ -590,17 +600,6 @@ public void operationComplete(Future<R> future) throws Exception {
             }
         });
 
-        details.getAttemptPromise().addListener(new FutureListener<R>() {
-            @Override
-            public void operationComplete(Future<R> future) throws Exception {
-                if (future.isCancelled()) {
-                    // command should be removed due to 
-                    // ConnectionWatchdog blockingQueue reconnection logic
-                    connection.removeCurrentCommand();
-                }
-            }
-        });
-
         synchronized (listener) {
             if (!details.getMainPromise().isDone()) {
                 connectionManager.getShutdownPromise().addListener(listener);
@@ -717,7 +716,7 @@ private <R, V> void checkAttemptFuture(final NodeSource source, final AsyncDetai
                 }
                 ((RedisClientResult)res).setRedisClient(addr);
             }
-            details.getMainPromise().setSuccess(res);
+            details.getMainPromise().trySuccess(res);
         } else {
             details.getMainPromise().tryFailure(future.cause());
         }

redisson/src/main/java/org/redisson/executor/RemotePromise.java

@@ -40,5 +40,9 @@ public void setAddFuture(Future<Boolean> addFuture) {
     public Future<Boolean> getAddFuture() {
         return addFuture;
     }
+
+    public void doCancel() {
+        super.cancel(true);
+    }
 
 }

redisson/src/main/java/org/redisson/remote/RRemoteServiceResponse.java

@@ -17,6 +17,11 @@
 
 import java.io.Serializable;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public interface RRemoteServiceResponse extends Serializable {
 
 }

redisson/src/main/java/org/redisson/remote/RemoteServiceAck.java

@@ -25,4 +25,6 @@
  */
 public class RemoteServiceAck implements RRemoteServiceResponse, Serializable {
 
+    private static final long serialVersionUID = -6332680404562746984L;
+
 }

redisson/src/main/java/org/redisson/remote/RemoteServiceCancelRequest.java

@@ -0,0 +1,43 @@
+/**
+ * Copyright 2016 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.remote;
+
+import java.io.Serializable;
+
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
+public class RemoteServiceCancelRequest implements Serializable {
+
+    private static final long serialVersionUID = -4800574267648904260L;
+
+    private boolean mayInterruptIfRunning;
+
+    public RemoteServiceCancelRequest() {
+    }
+
+    public RemoteServiceCancelRequest(boolean mayInterruptIfRunning) {
+        super();
+        this.mayInterruptIfRunning = mayInterruptIfRunning;
+    }
+
+    public boolean isMayInterruptIfRunning() {
+        return mayInterruptIfRunning;
+    }
+
+}

redisson/src/main/java/org/redisson/remote/RemoteServiceCancelResponse.java

@@ -0,0 +1,29 @@
+/**
+ * Copyright 2016 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.remote;
+
+import java.io.Serializable;
+
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
+public class RemoteServiceCancelResponse implements RRemoteServiceResponse, Serializable {
+
+    private static final long serialVersionUID = -4356901222132702182L;
+
+}

redisson/src/main/java/org/redisson/remote/RemoteServiceKey.java

@@ -15,6 +15,11 @@
  */
 package org.redisson.remote;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class RemoteServiceKey {
 
     private final Class<?> serviceInterface;

redisson/src/main/java/org/redisson/remote/RemoteServiceMethod.java

@@ -17,6 +17,11 @@
 
 import java.lang.reflect.Method;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class RemoteServiceMethod {
 
     private final Object bean;

redisson/src/main/java/org/redisson/remote/RemoteServiceRequest.java

@@ -20,8 +20,15 @@
 
 import org.redisson.api.RemoteInvocationOptions;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class RemoteServiceRequest implements Serializable {
 
+    private static final long serialVersionUID = -1711385312384040075L;
+
     private String requestId;
     private String methodName;
     private Object[] args;

redisson/src/main/java/org/redisson/remote/RemoteServiceResponse.java

@@ -17,8 +17,15 @@
 
 import java.io.Serializable;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class RemoteServiceResponse implements RRemoteServiceResponse, Serializable {
 
+    private static final long serialVersionUID = -1958922748139674253L;
+
     private Object result;
     private Throwable error;