Removing smarts of filtering expired messages. This change had been removed due to merge conflicts

Author: makam

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java

@@ -356,7 +356,7 @@ public CompletableFuture<IMessage> receiveAsync() {
 
     @Override
     public CompletableFuture<IMessage> receiveAsync(Duration serverWaitTime) {
-        CompletableFuture<IMessage> receiveFuture = this.internalReceiver.receiveAsync(1, serverWaitTime).thenApplyAsync(c ->
+        return this.internalReceiver.receiveAsync(1, serverWaitTime).thenApplyAsync(c ->
         {
             if (c == null)
                 return null;
@@ -365,8 +365,6 @@ else if (c.isEmpty())
             else
                 return MessageConverter.convertAmqpMessageToBrokeredMessage(c.toArray(new MessageWithDeliveryTag[0])[0]);
         });
-        
-        return this.filterLockExpiredMessage(receiveFuture, serverWaitTime);
     }
 
     @Override
@@ -376,7 +374,7 @@ public CompletableFuture<Collection<IMessage>> receiveBatchAsync(int maxMessageC
 
     @Override
     public CompletableFuture<Collection<IMessage>> receiveBatchAsync(int maxMessageCount, Duration serverWaitTime) {
-        CompletableFuture<Collection<IMessage>> receiveFuture = this.internalReceiver.receiveAsync(maxMessageCount, serverWaitTime).thenApplyAsync(c ->
+        return this.internalReceiver.receiveAsync(maxMessageCount, serverWaitTime).thenApplyAsync(c ->
         {
             if (c == null)
                 return null;
@@ -385,8 +383,7 @@ else if (c.isEmpty())
             else
                 return convertAmqpMessagesWithDeliveryTagsToBrokeredMessages(c);
         });
-        
-        return this.filterLockExpiredMessages(receiveFuture, maxMessageCount, serverWaitTime);
+
     }
 
     @Override
@@ -503,116 +500,6 @@ private void ensurePeekLockReceiveMode() {
             throw new UnsupportedOperationException("Operations Complete/Abandon/DeadLetter/Defer cannot be called on a receiver opened in ReceiveAndDelete mode.");
         }
     }
-    
-    private boolean isMessageLockExpired(IMessage msg)
-    {
-        return msg.getLockedUntilUtc().isBefore(Instant.now());
-    }
-    
-    private CompletableFuture<IMessage> filterLockExpiredMessage(CompletableFuture<IMessage> receivedFuture, Duration serverWaitTime)
-    {
-        if(this.receiveMode == ReceiveMode.RECEIVEANDDELETE)
-        {
-            return receivedFuture;
-        }
-        else
-        {
-            Instant startTime = Instant.now();
-            return receivedFuture.thenCompose((msg) -> {
-                if(msg == null)
-                {
-                    return receivedFuture;
-                }
-                else
-                {
-                    if(isMessageLockExpired(msg))
-                    {
-                        // Message lock already expired. Receive another message
-                        TRACE_LOGGER.warn("Lock of the prefetched message with sequence number '{}' and id '{}' from '{}' already expired", msg.getSequenceNumber(), msg.getMessageId(), this.getEntityPath());
-                        Duration remainingWaitTime = serverWaitTime.minus(Duration.between(startTime, Instant.now()));
-                        if(remainingWaitTime.isNegative() || remainingWaitTime.isZero())
-                        {
-                            return CompletableFuture.completedFuture(null);
-                        }
-                        else
-                        {
-                            TRACE_LOGGER.debug("Ignored the lock exipred message and receiving again from '{}'", this.getEntityPath());
-                            return this.receiveAsync(remainingWaitTime);
-                        }
-                    }
-                    else
-                    {
-                        return CompletableFuture.completedFuture(msg);
-                    }
-                }
-            });
-        }
-    }
-    
-    private CompletableFuture<Collection<IMessage>> filterLockExpiredMessages(CompletableFuture<Collection<IMessage>> receivedFuture, int maxMessageCount, Duration serverWaitTime)
-    {
-        if(this.receiveMode == ReceiveMode.RECEIVEANDDELETE)
-        {
-            return receivedFuture;
-        }
-        else
-        {
-            Instant startTime = Instant.now();
-            return receivedFuture.thenCompose((messages) -> {
-                if(messages == null || messages.size() == 0)
-                {
-                    return receivedFuture;
-                }
-                else
-                {
-                    boolean areMessagesRemoved = false;
-                    Iterator<IMessage> msgIterator = messages.iterator();
-                    while(msgIterator.hasNext())
-                    {
-                        IMessage msg = msgIterator.next();
-                        if(isMessageLockExpired(msg))
-                        {
-                            // Message lock already expired. remove it
-                            TRACE_LOGGER.warn("Lock of the prefetched message with sequence number '{}' and id '{}' from '{}' already expired. Removing it from the list of messages returned to the caller.", msg.getSequenceNumber(), msg.getMessageId(), this.getEntityPath());
-                            msgIterator.remove();
-                            areMessagesRemoved = true;
-                        }
-                        else
-                        {
-                            // Break the loop. As next messages in the list are received from the entity after current message and are definitely not expired. No need to check them
-                            break;
-                        }
-                    }
-                    
-                    if(areMessagesRemoved)
-                    {
-                        if(messages.size() > 0)
-                        {
-                            // There are some messages still in the list.. Just return them to the caller
-                            return CompletableFuture.completedFuture(messages);
-                        }
-                        else
-                        {
-                            Duration remainingWaitTime = serverWaitTime.minus(Duration.between(startTime, Instant.now()));
-                            if(remainingWaitTime.isNegative() || remainingWaitTime.isZero())
-                            {
-                                return CompletableFuture.completedFuture(null);
-                            }
-                            else
-                            {
-                                TRACE_LOGGER.debug("All messages in the received list are lock expired. So receiving again from '{}'", this.getEntityPath());
-                                return this.receiveBatchAsync(maxMessageCount, remainingWaitTime);
-                            }
-                        }
-                    }
-                    else
-                    {
-                        return receivedFuture;
-                    }
-                }
-            });
-        }
-    }
 
     private CompletableFuture<Boolean> checkIfValidRequestResponseLockTokenAsync(UUID lockToken) {
         CompletableFuture<Boolean> future = new CompletableFuture<Boolean>();