Cherrypicking fixes from Dev branch (#244)

* Adding a property needed for additional validation to check client type (#240)

* Adding a property needed for additional validation to check if the type of client being created matches the entity type.
Added unit tests too.

* Fixing a unit test

# Conflicts:
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/ClientFactory.java
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/Controller.java
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/RequestResponseLink.java
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/RequestResponseLinkCache.java
#	azure-servicebus/src/test/java/com/microsoft/azure/servicebus/QueueSendReceiveTests.java

* Renaming two public methods. From getScheduledEnqueuedTime to getScheduledEnqueueTime. (#241)

Similarly for set. They were mistakenly kept like that in the first release.
# Conflicts:
#	azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java

* Chaning version to 1.2.7

Author: yvgopal

azure-servicebus/azure-servicebus.pom

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.microsoft.azure</groupId>
     <artifactId>azure-servicebus</artifactId>
-    <version>1.2.6</version>
+    <version>1.2.7</version>
     <licenses>
         <license>
             <name>The MIT License (MIT)</name>

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

@@ -10,6 +10,7 @@
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 
+import com.microsoft.azure.servicebus.primitives.MessagingEntityType;
 import com.microsoft.azure.servicebus.primitives.MessagingFactory;
 import com.microsoft.azure.servicebus.primitives.ServiceBusException;
 
@@ -19,8 +20,8 @@
 final class BrowsableMessageSession extends MessageSession {
     private static final String INVALID_OPERATION_ERROR_MESSAGE = "Unsupported operation on a browse only session.";
 
-    BrowsableMessageSession(String sessionId, MessagingFactory messagingFactory, String entityPath) {
-        super(messagingFactory, entityPath, sessionId, ReceiveMode.PEEKLOCK);
+    BrowsableMessageSession(String sessionId, MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType) {
+        super(messagingFactory, entityPath, entityType, sessionId, ReceiveMode.PEEKLOCK);
 //		try {
 //			this.initializeAsync().get();
 //		} catch (InterruptedException | ExecutionException e) {

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

@@ -7,6 +7,7 @@
 import java.util.concurrent.CompletableFuture;
 
 import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
+import com.microsoft.azure.servicebus.primitives.MessagingEntityType;
 import com.microsoft.azure.servicebus.primitives.MessagingFactory;
 import com.microsoft.azure.servicebus.primitives.ServiceBusException;
 import com.microsoft.azure.servicebus.primitives.Util;
@@ -48,6 +49,10 @@ public static IMessageSender createMessageSenderFromConnectionStringBuilder(Conn
         return Utils.completeFuture(createMessageSenderFromConnectionStringBuilderAsync(amqpConnectionStringBuilder));
     }
 
+    static IMessageSender createMessageSenderFromConnectionStringBuilder(ConnectionStringBuilder amqpConnectionStringBuilder, MessagingEntityType entityType) throws InterruptedException, ServiceBusException {
+        return Utils.completeFuture(createMessageSenderFromConnectionStringBuilderAsync(amqpConnectionStringBuilder, entityType));
+    }
+    
     /**
      * Creates a message sender to the entity using the client settings.
      * @param namespaceName namespace of entity
@@ -73,10 +78,26 @@ public static IMessageSender createMessageSenderFromEntityPath(String namespaceN
     public static IMessageSender createMessageSenderFromEntityPath(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings) throws InterruptedException, ServiceBusException {
         return Utils.completeFuture(createMessageSenderFromEntityPathAsync(namespaceEndpointURI, entityPath, clientSettings));
     }
+    
+    static IMessageSender createMessageSenderFromEntityPath(URI namespaceEndpointURI, String entityPath, MessagingEntityType entityType, ClientSettings clientSettings) throws InterruptedException, ServiceBusException {
+        return Utils.completeFuture(createMessageSenderFromEntityPathAsync(namespaceEndpointURI, entityPath, entityType, clientSettings));
+    }
 
-    static IMessageSender createMessageSenderFromEntityPath(MessagingFactory messagingFactory, String entityPath) throws InterruptedException, ServiceBusException {
+    /**
+     * Creates a message sender to the entity.
+     * @param messagingFactory messaging factory (which represents a connection) on which sender needs to be created
+     * @param entityPath path of entity
+     * @return IMessageSender instance
+     * @throws InterruptedException if the current thread was interrupted while waiting
+     * @throws ServiceBusException if the sender cannot be created
+     */
+    public static IMessageSender createMessageSenderFromEntityPath(MessagingFactory messagingFactory, String entityPath) throws InterruptedException, ServiceBusException {
         return Utils.completeFuture(createMessageSenderFromEntityPathAsync(messagingFactory, entityPath));
     }
+    
+    static IMessageSender createMessageSenderFromEntityPath(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType) throws InterruptedException, ServiceBusException {
+        return Utils.completeFuture(createMessageSenderFromEntityPathAsync(messagingFactory, entityPath, entityType));
+    }
 
     /**
      * Create message sender asynchronously from connection string with <a href="https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-sas">Shared Access Signatures</a>
@@ -100,6 +121,11 @@ public static CompletableFuture<IMessageSender> createMessageSenderFromConnectio
         return createMessageSenderFromEntityPathAsync(amqpConnectionStringBuilder.getEndpoint(), amqpConnectionStringBuilder.getEntityPath(),  Util.getClientSettingsFromConnectionStringBuilder(amqpConnectionStringBuilder));
     }
 
+    static CompletableFuture<IMessageSender> createMessageSenderFromConnectionStringBuilderAsync(ConnectionStringBuilder amqpConnectionStringBuilder, MessagingEntityType entityType) {
+        Utils.assertNonNull("amqpConnectionStringBuilder", amqpConnectionStringBuilder);
+        return createMessageSenderFromEntityPathAsync(amqpConnectionStringBuilder.getEndpoint(), amqpConnectionStringBuilder.getEntityPath(), entityType, Util.getClientSettingsFromConnectionStringBuilder(amqpConnectionStringBuilder));
+    }
+    
     /**
      * Creates a message sender asynchronously to the entity using the client settings.
      * @param namespaceName namespace name of entity
@@ -122,15 +148,30 @@ public static CompletableFuture<IMessageSender> createMessageSenderFromEntityPat
      * @return a CompletableFuture representing the pending creating of IMessageSender instance
      */
     public static CompletableFuture<IMessageSender> createMessageSenderFromEntityPathAsync(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings)
+    {
+        return createMessageSenderFromEntityPathAsync(namespaceEndpointURI, entityPath, null, clientSettings);
+    }
+    
+    static CompletableFuture<IMessageSender> createMessageSenderFromEntityPathAsync(URI namespaceEndpointURI, String entityPath, MessagingEntityType entityType, ClientSettings clientSettings)
     {
         Utils.assertNonNull("namespaceEndpointURI", namespaceEndpointURI);
-        MessageSender sender = new MessageSender(namespaceEndpointURI, entityPath, clientSettings);
+        MessageSender sender = new MessageSender(namespaceEndpointURI, entityPath, entityType, clientSettings);
         return sender.initializeAsync().thenApply((v) -> sender);
     }
 
-    static CompletableFuture<IMessageSender> createMessageSenderFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath) {
+    /**
+     * Creates a message sender asynchronously to the entity using the {@link MessagingFactory}
+     * @param messagingFactory messaging factory (which represents a connection) on which sender needs to be created
+     * @param entityPath path of entity
+     * @return a CompletableFuture representing the pending creating of IMessageSender instance
+     */
+    public static CompletableFuture<IMessageSender> createMessageSenderFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath) {
+        return createMessageSenderFromEntityPathAsync(messagingFactory, entityPath, null);
+    }
+    
+    static CompletableFuture<IMessageSender> createMessageSenderFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType) {
         Utils.assertNonNull("messagingFactory", messagingFactory);
-        MessageSender sender = new MessageSender(messagingFactory, entityPath);
+        MessageSender sender = new MessageSender(messagingFactory, entityPath, entityType);
         return sender.initializeAsync().thenApply((v) -> sender);
     }
 
@@ -252,6 +293,10 @@ static IMessageReceiver createMessageReceiverFromEntityPath(MessagingFactory mes
     static IMessageReceiver createMessageReceiverFromEntityPath(MessagingFactory messagingFactory, String entityPath, ReceiveMode receiveMode) throws InterruptedException, ServiceBusException {
         return Utils.completeFuture(createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, receiveMode));
     }
+    
+    static IMessageReceiver createMessageReceiverFromEntityPath(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, ReceiveMode receiveMode) throws InterruptedException, ServiceBusException {
+        return Utils.completeFuture(createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, entityType, receiveMode));
+    }
 
     /**
      * Create {@link IMessageReceiver} in default {@link ReceiveMode#PEEKLOCK} mode asynchronously from connection string with <a href="https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-sas">Shared Access Signatures</a>
@@ -343,7 +388,7 @@ public static CompletableFuture<IMessageReceiver> createMessageReceiverFromEntit
     public static CompletableFuture<IMessageReceiver> createMessageReceiverFromEntityPathAsync(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings, ReceiveMode receiveMode) {
         Utils.assertNonNull("namespaceEndpointURI", namespaceEndpointURI);
         Utils.assertNonNull("entityPath", entityPath);
-        MessageReceiver receiver = new MessageReceiver(namespaceEndpointURI, entityPath, clientSettings, receiveMode);
+        MessageReceiver receiver = new MessageReceiver(namespaceEndpointURI, entityPath, null, clientSettings, receiveMode);
         return receiver.initializeAsync().thenApply((v) -> receiver);
     }
 
@@ -352,8 +397,12 @@ static CompletableFuture<IMessageReceiver> createMessageReceiverFromEntityPathAs
     }
 
     static CompletableFuture<IMessageReceiver> createMessageReceiverFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, ReceiveMode receiveMode) {
+        return createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, null, receiveMode);
+    }
+    
+    static CompletableFuture<IMessageReceiver> createMessageReceiverFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, ReceiveMode receiveMode) {
         Utils.assertNonNull("messagingFactory", messagingFactory);
-        MessageReceiver receiver = new MessageReceiver(messagingFactory, entityPath, receiveMode);
+        MessageReceiver receiver = new MessageReceiver(messagingFactory, entityPath, entityType, receiveMode);
         return receiver.initializeAsync().thenApply((v) -> receiver);
     }
 
@@ -575,7 +624,7 @@ public static CompletableFuture<IMessageSession> acceptSessionFromEntityPathAsyn
     public static CompletableFuture<IMessageSession> acceptSessionFromEntityPathAsync(URI namespaceEndpointURI, String entityPath, String sessionId, ClientSettings clientSettings, ReceiveMode receiveMode) {
         Utils.assertNonNull("namespaceEndpointURI", namespaceEndpointURI);
         Utils.assertNonNull("entityPath", entityPath);
-        MessageSession session = new MessageSession(namespaceEndpointURI, entityPath, sessionId, clientSettings, receiveMode);
+        MessageSession session = new MessageSession(namespaceEndpointURI, entityPath, null, sessionId, clientSettings, receiveMode);
         return session.initializeAsync().thenApply((v) -> session);
     }
 
@@ -584,8 +633,12 @@ static CompletableFuture<IMessageSession> acceptSessionFromEntityPathAsync(Messa
     }
 
     static CompletableFuture<IMessageSession> acceptSessionFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, String sessionId, ReceiveMode receiveMode) {
+        return acceptSessionFromEntityPathAsync(messagingFactory, entityPath, null, sessionId, receiveMode);
+    }
+    
+    static CompletableFuture<IMessageSession> acceptSessionFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, String sessionId, ReceiveMode receiveMode) {
         Utils.assertNonNull("messagingFactory", messagingFactory);
-        MessageSession session = new MessageSession(messagingFactory, entityPath, sessionId, receiveMode);
+        MessageSession session = new MessageSession(messagingFactory, entityPath, entityType, sessionId, receiveMode);
         return session.initializeAsync().thenApply((v) -> session);
     }
 }

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

@@ -137,16 +137,41 @@ public interface IMessage {
      *
      * @return the instant at which the message will be enqueued in Azure Service Bus
      * @see <a href="https://docs.microsoft.com/azure/service-bus-messaging/message-sequencing">Message Sequencing and Timestamps</a> 
+     * @deprecated Replaced by {@link #getScheduledEnqueueTimeUtc()
      */
+    @Deprecated
     public Instant getScheduledEnqueuedTimeUtc();
 
     /**
      * Sets the scheduled enqueue time of this message.  
      * 
      * @param scheduledEnqueueTimeUtc the instant at which this message should be enqueued in Azure Service Bus
-     * @see #getScheduledEnqueuedTimeUtc()
+     * @see #getScheduledEnqueueTimeUtc()
+     * @deprecated Replaced by {@link #setScheduledEnqueueTimeUtc(Instant)()
      */
+    @Deprecated
     public void setScheduledEnqueuedTimeUtc(Instant scheduledEnqueueTimeUtc);
+    
+    /**
+     * Gets the scheduled enqueue time of this message. 
+     * 
+     * This value is used for delayed message availability. The message is safely added to 
+     * the queue, but is not considered active and therefore not retrievable until the 
+     * scheduled enqueue time. Mind that the message may not be activated (enqueued) at the exact given 
+     * instant; the actual activation time depends on the queue's workload and its state.
+     *
+     * @return the instant at which the message will be enqueued in Azure Service Bus
+     * @see <a href="https://docs.microsoft.com/azure/service-bus-messaging/message-sequencing">Message Sequencing and Timestamps</a> 
+     */
+    public Instant getScheduledEnqueueTimeUtc();
+
+    /**
+     * Sets the scheduled enqueue time of this message.  
+     * 
+     * @param scheduledEnqueueTimeUtc the instant at which this message should be enqueued in Azure Service Bus
+     * @see #getScheduledEnqueueTimeUtc()
+     */
+    public void setScheduledEnqueueTimeUtc(Instant scheduledEnqueueTimeUtc);
 
     /**
      * Gets the unique number assigned to a message by Service Bus.

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

@@ -254,15 +254,27 @@ public void setReplyToSessionId(String replyToSessionId) {
 		this.replyToSessionId = replyToSessionId;		
 	}
 
+	@Deprecated
 	@Override
 	public Instant getScheduledEnqueuedTimeUtc() {
-		return this.scheduledEnqueueTimeUtc;
+		return this.getScheduledEnqueueTimeUtc();
 	}
 
+	@Deprecated
 	@Override
 	public void setScheduledEnqueuedTimeUtc(Instant scheduledEnqueueTimeUtc) {
+		this.setScheduledEnqueueTimeUtc(scheduledEnqueueTimeUtc);
+	}
+	
+	@Override
+	public Instant getScheduledEnqueueTimeUtc() {
+		return this.scheduledEnqueueTimeUtc;
+	}
+
+	@Override
+	public void setScheduledEnqueueTimeUtc(Instant scheduledEnqueueTimeUtc) {
 		this.scheduledEnqueueTimeUtc = scheduledEnqueueTimeUtc;		
-	}	
+	}
 
 	@Override
 	public String getPartitionKey() {
@@ -301,5 +313,4 @@ void setDeliveryTag(byte[] deliveryTag)
 	{
 		this.deliveryTag = deliveryTag;
 	}
-	
 }

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

@@ -16,6 +16,7 @@
 
 import com.microsoft.azure.servicebus.primitives.ExceptionUtil;
 import com.microsoft.azure.servicebus.primitives.MessageLockLostException;
+import com.microsoft.azure.servicebus.primitives.MessagingEntityType;
 import com.microsoft.azure.servicebus.primitives.MessagingFactory;
 import com.microsoft.azure.servicebus.primitives.OperationCancelledException;
 import com.microsoft.azure.servicebus.primitives.ServiceBusException;
@@ -36,6 +37,7 @@ class MessageAndSessionPump extends InitializableEntity implements IMessageAndSe
     private final MessagingFactory factory;
     private final String entityPath;
     private final ReceiveMode receiveMode;
+    private final MessagingEntityType entityType;
     private IMessageReceiver innerReceiver;
 
     private boolean handlerRegistered = false;
@@ -45,10 +47,11 @@ class MessageAndSessionPump extends InitializableEntity implements IMessageAndSe
     private SessionHandlerOptions sessionHandlerOptions;
     private int prefetchCount;
 
-    public MessageAndSessionPump(MessagingFactory factory, String entityPath, ReceiveMode receiveMode) {
+    public MessageAndSessionPump(MessagingFactory factory, String entityPath, MessagingEntityType entityType, ReceiveMode receiveMode) {
         super(StringUtil.getShortRandomString());
         this.factory = factory;
         this.entityPath = entityPath;
+        this.entityType = entityType;
         this.receiveMode = receiveMode;
         this.openSessions = new ConcurrentHashMap<>();
         this.prefetchCount = UNSET_PREFETCH_COUNT;
@@ -66,7 +69,7 @@ public void registerMessageHandler(IMessageHandler handler, MessageHandlerOption
         this.messageHandler = handler;
         this.messageHandlerOptions = handlerOptions;
 
-        this.innerReceiver = ClientFactory.createMessageReceiverFromEntityPath(this.factory, this.entityPath, this.receiveMode);
+        this.innerReceiver = ClientFactory.createMessageReceiverFromEntityPath(this.factory, this.entityPath, this.entityType, this.receiveMode);
         TRACE_LOGGER.info("Created MessageReceiver to entity '{}'", this.entityPath);
         if(this.prefetchCount != UNSET_PREFETCH_COUNT)
         {
@@ -210,7 +213,7 @@ private void receiveAndPumpMessage() {
     private void acceptSessionAndPumpMessages() {
         if (!this.getIsClosingOrClosed()) {
             TRACE_LOGGER.debug("Accepting a session from entity '{}'", this.entityPath);
-            CompletableFuture<IMessageSession> acceptSessionFuture = ClientFactory.acceptSessionFromEntityPathAsync(this.factory, this.entityPath, null, this.receiveMode);
+            CompletableFuture<IMessageSession> acceptSessionFuture = ClientFactory.acceptSessionFromEntityPathAsync(this.factory, this.entityPath, this.entityType, null, this.receiveMode);
             acceptSessionFuture.handleAsync((session, acceptSessionEx) -> {
                 if (acceptSessionEx != null) {
                     acceptSessionEx = ExceptionUtil.extractAsyncCompletionCause(acceptSessionEx);

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

@@ -50,9 +50,9 @@ public static org.apache.qpid.proton.message.Message convertBrokeredMessageToAmq
 		amqpMessage.setGroupId(brokeredMessage.getSessionId());
 
 		Map<Symbol, Object> messageAnnotationsMap = new HashMap<Symbol, Object>();
-		if(brokeredMessage.getScheduledEnqueuedTimeUtc() != null)
+		if(brokeredMessage.getScheduledEnqueueTimeUtc() != null)
 		{
-			messageAnnotationsMap.put(Symbol.valueOf(ClientConstants.SCHEDULEDENQUEUETIMENAME), Date.from(brokeredMessage.getScheduledEnqueuedTimeUtc()));
+			messageAnnotationsMap.put(Symbol.valueOf(ClientConstants.SCHEDULEDENQUEUETIMENAME), Date.from(brokeredMessage.getScheduledEnqueueTimeUtc()));
 		}
 
 		if(!StringUtil.isNullOrEmpty(brokeredMessage.getPartitionKey()))
@@ -158,7 +158,7 @@ public static Message convertAmqpMessageToBrokeredMessage(org.apache.qpid.proton
 							brokeredMessage.setEnqueuedTimeUtc(((Date)entry.getValue()).toInstant());
 							break;
 						case ClientConstants.SCHEDULEDENQUEUETIMENAME:
-	                        brokeredMessage.setScheduledEnqueuedTimeUtc(((Date)entry.getValue()).toInstant());
+	                        brokeredMessage.setScheduledEnqueueTimeUtc(((Date)entry.getValue()).toInstant());
 	                        break;
 	                    case ClientConstants.SEQUENCENUBMERNAME:
 	                        brokeredMessage.setSequenceNumber((long)entry.getValue());