Fixing the infinite loop that runs to renew SAS token after 20 minutes when a connection string contains the SAS token instead of SAS key. (#342)

Author: yvgopal

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CommonRequestResponseOperations.java

@@ -80,7 +80,6 @@ static CompletableFuture<Void> sendCBSTokenAsync(RequestResponseLink requestResp
         Message requestMessage = RequestResponseUtils.createRequestMessageFromValueBody(ClientConstants.REQUEST_RESPONSE_PUT_TOKEN_OPERATION, securityToken.getTokenValue(), Util.adjustServerTimeout(operationTimeout));
         requestMessage.getApplicationProperties().getValue().put(ClientConstants.REQUEST_RESPONSE_PUT_TOKEN_TYPE, securityToken.getTokenType().toString());
         requestMessage.getApplicationProperties().getValue().put(ClientConstants.REQUEST_RESPONSE_PUT_TOKEN_AUDIENCE, securityToken.getTokenAudience());
-        requestMessage.getApplicationProperties().getValue().put(ClientConstants.REQUEST_RESPONSE_PUT_TOKEN_EXPIRATION, securityToken.getValidUntil().toEpochMilli());
         CompletableFuture<Message> responseFuture = requestResponseLink.requestAysnc(requestMessage, operationTimeout);
         return responseFuture.thenComposeAsync((responseMessage) -> {
             CompletableFuture<Void> returningFuture = new CompletableFuture<Void>();

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java

@@ -1167,6 +1167,7 @@ private synchronized CompletableFuture<Void> ensureLinkIsOpen()
 	                	Throwable cause = ExceptionUtil.extractAsyncCompletionCause(sendTokenEx);
         				TRACE_LOGGER.error("Sending SAS Token to '{}' failed.", this.receivePath, cause);
 	                    this.receiveLinkReopenFuture.completeExceptionally(sendTokenEx);
+	                    this.clearAllPendingWorkItems(sendTokenEx);
 	                }
 	                else
 	                {
@@ -1214,7 +1215,7 @@ private void completePendingUpdateStateWorkItem(Delivery delivery, String delive
 		this.pendingUpdateStateRequests.remove(deliveryTagAsString);
 	}
 
-	private void clearAllPendingWorkItems(Exception exception)
+	private void clearAllPendingWorkItems(Throwable exception)
 	{
 	    TRACE_LOGGER.info("Completeing all pending receive and updateState operation on the receiver to '{}'", this.receivePath);
 		final boolean isTransientException = exception == null ||

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java

@@ -553,7 +553,7 @@ else if (outcome instanceof Released)
 		}
 	}
 
-	private void clearAllPendingSendsWithException(Exception failureException)
+	private void clearAllPendingSendsWithException(Throwable failureException)
 	{
 	    synchronized (this.pendingSendLock)
         {
@@ -567,7 +567,7 @@ private void clearAllPendingSendsWithException(Exception failureException)
         }
 	}
 
-	private void cleanupFailedSend(final SendWorkItem<Void> failedSend, final Exception exception)
+	private void cleanupFailedSend(final SendWorkItem<Void> failedSend, final Throwable exception)
 	{
 		failedSend.cancelTimeoutTask(false);		
 		ExceptionUtil.completeExceptionally(failedSend.getWork(), exception, this, true);
@@ -730,6 +730,7 @@ private synchronized CompletableFuture<Void> ensureLinkIsOpen()
                     	Throwable cause = ExceptionUtil.extractAsyncCompletionCause(sendTokenEx);
         				TRACE_LOGGER.error("Sending SAS Token to '{}' failed.", this.sendPath, cause);
                         this.sendLinkReopenFuture.completeExceptionally(sendTokenEx);
+                        this.clearAllPendingSendsWithException(sendTokenEx);
                     }
                     else
                     {

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/ExceptionUtil.java

@@ -121,7 +121,7 @@ else if (errorCondition.getCondition() == ClientConstants.ENTITY_ALREADY_EXISTS_
 		return new ServiceBusException(ClientConstants.DEFAULT_IS_TRANSIENT, errorCondition.toString());
 	}	
 
-	static <T> void completeExceptionally(CompletableFuture<T> future, Exception exception, IErrorContextProvider contextProvider, boolean completeAsynchronously)
+	static <T> void completeExceptionally(CompletableFuture<T> future, Throwable exception, IErrorContextProvider contextProvider, boolean completeAsynchronously)
 	{
 		if (exception != null && exception instanceof ServiceBusException)
 		{

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java

@@ -627,9 +627,17 @@ private CompletableFuture<Instant> generateAndSendSecurityToken(String sasTokenA
 
 	private static ScheduledFuture<?> scheduleRenewTimer(Instant currentTokenValidUntil, Runnable validityRenewer)
 	{
-	    // It will eventually expire. Renew it
-        int renewInterval = Util.getTokenRenewIntervalInSeconds((int)Duration.between(Instant.now(), currentTokenValidUntil).getSeconds());
-        return Timer.schedule(validityRenewer, Duration.ofSeconds(renewInterval), TimerType.OneTimeRun);
+		if(currentTokenValidUntil == Instant.MAX)
+		{
+			// User provided token or will never expire
+			return null;
+		}
+		else
+		{
+			// It will eventually expire. Renew it
+	        int renewInterval = Util.getTokenRenewIntervalInSeconds((int)Duration.between(Instant.now(), currentTokenValidUntil).getSeconds());
+	        return Timer.schedule(validityRenewer, Duration.ofSeconds(renewInterval), TimerType.OneTimeRun);
+		}
 	}
 
 	CompletableFuture<RequestResponseLink> obtainRequestResponseLinkAsync(String entityPath, MessagingEntityType entityType)

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/RequestResponseLink.java

@@ -343,6 +343,7 @@ private CompletableFuture<Void> recreateInternalLinks()
                 Throwable cause = ExceptionUtil.extractAsyncCompletionCause(sasTokenEx);
                 TRACE_LOGGER.error("Sending SAS Token failed. RequestResponseLink path:{}", this.linkPath, cause);
                 recreateInternalLinksFuture.completeExceptionally(cause);
+                this.completeAllPendingRequestsWithException(cause);
             }
             else
             {
@@ -405,7 +406,7 @@ public void run()
 		return recreateInternalLinksFuture;
 	}
 
-	private void completeAllPendingRequestsWithException(Exception exception)
+	private void completeAllPendingRequestsWithException(Throwable exception)
 	{
 	    TRACE_LOGGER.warn("Completing all pending requests with exception in request response link to {}", this.linkPath);
 		for(RequestResponseWorkItem workItem : this.pendingRequests.values())

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/Util.java

@@ -425,7 +425,7 @@ public static ClientSettings getClientSettingsFromConnectionStringBuilder(Connec
         }
         else
         {
-            tokenProvider = new SharedAccessSignatureTokenProvider(builder.getSharedAccessSignatureToken(), Instant.now().plus(Duration.ofSeconds(SecurityConstants.DEFAULT_SAS_TOKEN_VALIDITY_IN_SECONDS)));
+            tokenProvider = new SharedAccessSignatureTokenProvider(builder.getSharedAccessSignatureToken(), Instant.MAX); // Max validity as we will not generate another token
         }
 
         return new ClientSettings(tokenProvider, builder.getRetryPolicy(), builder.getOperationTimeout());