Removing finalize method altogether. (#388)

* Removing finalize method altogether.

* Chaging vesion.

* Addin finalize only for MessagingFactory as connection needs to be disposed properly.

* Another fix to avoid stalling in finalize waiting for object to close.

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.17</version>
+    <version>1.2.18</version>
     <licenses>
         <license>
             <name>The MIT License (MIT)</name>

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

@@ -15,6 +15,7 @@
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
 
 import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
 import org.apache.qpid.proton.amqp.transport.SenderSettleMode;
@@ -52,7 +53,8 @@ class MessageReceiver extends InitializableEntity implements IMessageReceiver, I
     private CoreMessageReceiver internalReceiver = null;
     private boolean isInitialized = false;
     private MessageBrowser browser = null;
-    private int messagePrefetchCount;    
+    private int messagePrefetchCount;
+    private ScheduledFuture<?> requestResponseLockTokenPruner = null;
 
     private final ConcurrentHashMap<UUID, Instant> requestResponseLockTokensToLockTimesMap;
 
@@ -425,6 +427,9 @@ protected CompletableFuture<Void> onClose() {
             } else {
                 TRACE_LOGGER.info("Closing MessageReceiver to entity '{}'", this.entityPath);
             }
+            if (this.requestResponseLockTokenPruner != null) {
+            	this.requestResponseLockTokenPruner.cancel(false);
+            }
             CompletableFuture<Void> closeReceiverFuture = this.internalReceiver.closeAsync();
 
             return closeReceiverFuture.thenComposeAsync((v) ->
@@ -624,8 +629,12 @@ public CompletableFuture<Collection<IMessage>> peekBatchAsync(long fromSequenceN
 
     private void schedulePruningRequestResponseLockTokens() {
         // Run it every 1 hour
-        Timer.schedule(new Runnable() {
+        this.requestResponseLockTokenPruner = Timer.schedule(new Runnable() {
             public void run() {
+            	if (MessageReceiver.this.getIsClosed())	{
+            		MessageReceiver.this.requestResponseLockTokenPruner.cancel(true);
+		    		return;
+		    	}
                 Instant systemTime = Instant.now();
                 Entry<UUID, Instant>[] copyOfEntries = (Entry<UUID, Instant>[]) MessageReceiver.this.requestResponseLockTokensToLockTimesMap.entrySet().toArray();
                 for (Entry<UUID, Instant> entry : copyOfEntries) {

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

@@ -138,17 +138,10 @@ protected final void throwIfClosed(Throwable cause)
 	}
 
 	@Override
-	protected void finalize()
-	{
-		try {
-			if(!this.isClosed)
-			{
-				this.close();	
-			}
-			
-			super.finalize();
-		} catch (Throwable e) {
-			//Ignore
-		}		
-	}
+    protected void finalize() throws Throwable {
+    	if (!this.getIsClosingOrClosed()) {
+    		this.closeAsync();
+    	}
+    	super.finalize();
+    }
 }

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

@@ -1051,8 +1051,12 @@ public void onEvent() {
 
             this.cancelSASTokenRenewTimer();
             this.closeRequestResponseLink();
-            this.updateStateRequestsTimeoutChecker.cancel(false);
-            this.returnMessagesLoopRunner.cancel(false);
+            if (this.updateStateRequestsTimeoutChecker != null) {
+            	this.updateStateRequestsTimeoutChecker.cancel(false);
+            }
+            if (this.returnMessagesLoopRunner != null) {
+            	this.returnMessagesLoopRunner.cancel(false);
+            }
         }
 	}
 

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

@@ -555,7 +555,9 @@ public void onEvent()
 	protected CompletableFuture<Void> onClose() {
 	    TRACE_LOGGER.info("Closing requestresponselink to {} by closing both internal sender and receiver links.", this.linkPath);
 	    this.cancelSASTokenRenewTimer();
-		return this.amqpSender.closeAsync().thenComposeAsync((v) -> this.amqpReceiver.closeAsync(), MessagingFactory.INTERNAL_THREAD_POOL);
+	    CompletableFuture<Void> senderCloseFuture = this.amqpSender.closeAsync();
+        CompletableFuture<Void> receiverCloseFuture = this.amqpReceiver.closeAsync();
+        return CompletableFuture.allOf(senderCloseFuture, receiverCloseFuture);
 	}
 
 	private static void scheduleLinkCloseTimeout(CompletableFuture<Void> closeFuture, Duration timeout, String linkName)

pom.xml

@@ -13,7 +13,7 @@
 		<proton-j-version>0.31.0</proton-j-version>
 	  	<junit-version>4.12</junit-version>
 		<slf4j-version>1.7.0</slf4j-version>
-		<client-current-version>1.2.17</client-current-version>		
+		<client-current-version>1.2.18</client-current-version>		
 	</properties>
 
 	<modules>