Feature - getPendingInvocations method added to RRemoteService object. #2171

redisson/src/main/java/org/redisson/RedissonRemoteService.java

@@ -134,6 +134,13 @@ public class RedissonRemoteService extends BaseRemoteService implements RRemoteS
         }
     }
     
+    @Override
+    public int getPendingInvocations(Class<?> remoteInterface) {
+        String requestQueueName = getRequestQueueName(remoteInterface);
+        RBlockingQueue<String> requestQueue = getBlockingQueue(requestQueueName, StringCodec.INSTANCE);
+        return requestQueue.size();
+    }
+    
     @Override
     public int getFreeWorkers(Class<?> remoteInterface) {
         Entry entry = remoteMap.get(remoteInterface);

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

@@ -58,13 +58,21 @@ import java.util.concurrent.TimeUnit;
 public interface RRemoteService {
 
     /**
-     * Returns free workers amount available for tasks 
+     * Returns free workers amount available for invocations 
      * 
      * @param remoteInterface - remote service interface
      * @return workers amount
      */
     int getFreeWorkers(Class<?> remoteInterface);
     
+    /**
+     * Returns pending invocations amount for handling in free workers.
+     * 
+     * @param remoteInterface - remote service interface
+     * @return invocations amount
+     */
+    int getPendingInvocations(Class<?> remoteInterface);
+    
     /**
      * Register remote service with single worker
      *

redisson/src/test/java/org/redisson/RedissonRemoteServiceTest.java

@@ -293,6 +293,28 @@ public class RedissonRemoteServiceTest extends BaseTest {
         remoteService.deregister(RemoteInterface.class);
     }
 
+    @Test
+    public void testPendingInvocations() throws InterruptedException, ExecutionException {
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        RRemoteService rs = redisson.getRemoteService();
+        rs.register(RemoteInterface.class, new RemoteImpl(), 1, executor);
+        
+        assertThat(rs.getPendingInvocations(RemoteInterface.class)).isEqualTo(0);
+        
+        RemoteInterfaceAsync ri = redisson.getRemoteService().get(RemoteInterfaceAsync.class);
+        
+        for (int i = 0; i < 5; i++) {
+            ri.timeoutMethod();
+        }
+        Thread.sleep(1000);
+        assertThat(rs.getPendingInvocations(RemoteInterface.class)).isEqualTo(4);
+        Thread.sleep(9000);
+        assertThat(rs.getPendingInvocations(RemoteInterface.class)).isEqualTo(0);
+
+        executor.shutdown();
+        executor.awaitTermination(1, TimeUnit.MINUTES);
+    }
+    
     @Test
     public void testFreeWorkers() throws InterruptedException, ExecutionException {
         RedissonClient r1 = createInstance();