Merge from master to dev (#212)

Author: nemakam

README.md

@@ -1,5 +1,5 @@
 <p align="center">
-  <img src="service-bus.png" alt="Microsoft Azure Relay" width="100"/>
+  <img src="service-bus.png" alt="Microsoft Azure Service Bus" width="100"/>
 </p>
 
 |Build/Package|Status|

azure-servicebus/pom.xml

@@ -82,5 +82,10 @@
 	 	    <artifactId>slf4j-log4j12</artifactId> 
 	 	    <version>${slf4j-version}</version>
 	  	</dependency>
-	</dependencies>
+	  	<!--<dependency>
+	 	    <groupId>org.slf4j</groupId> 
+	 	    <artifactId>slf4j-log4j12</artifactId> 
+	 	    <version>${slf4j-version}</version>
+	  	</dependency>-->
+	</dependencies>	 
 </project>

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) {
-        return this.internalReceiver.receiveAsync(1, serverWaitTime).thenApplyAsync(c ->
+        CompletableFuture<IMessage> receiveFuture = this.internalReceiver.receiveAsync(1, serverWaitTime).thenApplyAsync(c ->
         {
             if (c == null)
                 return null;
@@ -365,6 +365,8 @@ else if (c.isEmpty())
             else
                 return MessageConverter.convertAmqpMessageToBrokeredMessage(c.toArray(new MessageWithDeliveryTag[0])[0]);
         });
+        
+        return this.filterLockExpiredMessage(receiveFuture, serverWaitTime);
     }
 
     @Override
@@ -374,7 +376,7 @@ public CompletableFuture<Collection<IMessage>> receiveBatchAsync(int maxMessageC
 
     @Override
     public CompletableFuture<Collection<IMessage>> receiveBatchAsync(int maxMessageCount, Duration serverWaitTime) {
-        return this.internalReceiver.receiveAsync(maxMessageCount, serverWaitTime).thenApplyAsync(c ->
+        CompletableFuture<Collection<IMessage>> receiveFuture = this.internalReceiver.receiveAsync(maxMessageCount, serverWaitTime).thenApplyAsync(c ->
         {
             if (c == null)
                 return null;
@@ -383,6 +385,8 @@ else if (c.isEmpty())
             else
                 return convertAmqpMessagesWithDeliveryTagsToBrokeredMessages(c);
         });
+        
+        return this.filterLockExpiredMessages(receiveFuture, maxMessageCount, serverWaitTime);
     }
 
     @Override
@@ -499,6 +503,116 @@ 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>();

azure-servicebus/src/test/java/com/microsoft/azure/servicebus/SessionTests.java

@@ -4,6 +4,8 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.concurrent.ExecutionException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import org.junit.After;
 import org.junit.AfterClass;

pom.xml

@@ -13,7 +13,7 @@
 		<proton-j-version>0.22.0</proton-j-version>
 	  	<junit-version>4.12</junit-version>
 		<slf4j-version>1.8.0-alpha2</slf4j-version>
-		<client-current-version>1.2.4-PREVIEW</client-current-version>		
+		<client-current-version>1.2.5</client-current-version>		
 	</properties>
 
 	<modules>