[GitHub] jai1 commented on a change in pull request #727: Admin Tool: Adding support to peek batch Messages

[gi...@git.apache.org]

jai1 commented on a change in pull request #727: Admin Tool: Adding support to peek batch Messages
URL: https://github.com/apache/incubator-pulsar/pull/727#discussion_r138760003
 
 

 ##########
 File path: pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PersistentTopicsImpl.java
 ##########
 @@ -557,39 +563,37 @@ public void failed(Throwable throwable) {
     @Override
     public CompletableFuture<List<Message>> peekMessagesAsync(String destination, String subName, int numMessages) {
         checkArgument(numMessages > 0);
-        List<Message> messages = Lists.newArrayList();
-        CompletableFuture<List<Message>> futures = new CompletableFuture<List<Message>>();
+        CompletableFuture<List<Message>> future = new CompletableFuture<List<Message>>();
+        peekMessagesAsync(destination, subName, numMessages, Lists.newArrayList(), future, 1);
+        return future;
+    }
+    
+    
+    private void peekMessagesAsync(String destination, String subName, int numMessages,
+            List<Message> messages, CompletableFuture<List<Message>> future, int nthMessage) {
+        if (numMessages <= 0) {
+            future.complete(messages);
+            return;
+        }
 
         // if peeking first message succeeds, we know that the topic and subscription exists
-        peekNthMessage(destination, subName, 1).handle((r, ex) -> {
+        peekNthMessage(destination, subName, nthMessage).handle((r, ex) -> {
             if (ex != null) {
-                futures.completeExceptionally(ex);
-            } else {
-                messages.add(r);
-                List<CompletableFuture<Message>> futureMessages = Lists.newArrayList();
-                for (int i = 2; i <= numMessages; i++) {
-                    futureMessages.add(peekNthMessage(destination, subName, i));
-                }
-
-                try {
-                    for (CompletableFuture<Message> futureMessage : futureMessages) {
-                        messages.add(futureMessage.get());
-                    }
-                } catch (Exception e) {
-                    // if we get a not found exception, it means that the position for the message we are trying to get
-                    // does not exist. At this point, we can return the already found messages.
-                if (!(e.getCause() instanceof NotFoundException)) {
-                    futures.completeExceptionally(e.getCause());
-                    return null;
+                // if we get a not found exception, it means that the position for the message we are trying to get        
 
 Review comment:
   You get an exception
 
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services