Add numMessagesInQueue and getAvailablePermits for PartitionedConsume…

…rImpl
apache · Oct 18, 2016 · 8ec9f88 · 8ec9f88
1 parent b2805b8
commit 8ec9f88
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 0 deletions.
diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerBase.java
@@ -291,6 +291,10 @@ protected SubType getSubType() {
 
     abstract public boolean isConnected();
 
+    abstract public int getAvailablePermits();
+
+    abstract public int numMessagesInQueue();
+
     public CompletableFuture<Consumer> subscribeFuture() {
         return subscribeFuture;
     }

diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java
@@ -858,10 +858,12 @@ int getPartitionIndex() {
         return partitionIndex;
     }
 
+    @Override
     public int getAvailablePermits() {
         return availablePermits.get();
     }
 
+    @Override
     public int numMessagesInQueue() {
         return incomingMessages.size();
     }

diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedConsumerImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedConsumerImpl.java
@@ -398,6 +398,16 @@ List<ConsumerImpl> getConsumers() {
         return consumers;
     }
 
+    @Override
+    public int getAvailablePermits() {
+        return consumers.stream().mapToInt(ConsumerImpl::getAvailablePermits).sum();
+    }
+
+    @Override
+    public int numMessagesInQueue() {
+        return consumers.stream().mapToInt(ConsumerImpl::numMessagesInQueue).sum();
+    }
+
     @Override
     public synchronized ConsumerStats getStats() {
         if (stats == null) {