From 2cf2164548419f33a01ecfc68d960d291da887a3 Mon Sep 17 00:00:00 2001
From: Yuk-Ming Lam <ymlam@gracelam.com>
Date: Tue, 9 Feb 2021 03:40:16 -0500
Subject: [PATCH] Fix bug that service thread may get stuck on cancel.

This would lead to memory leak when SMSEngine is shutdown.
---
 src/main/java/altk/comm/engine/Broadcast.java | 29 ++++++++++++-------
 1 file changed, 19 insertions(+), 10 deletions(-)

diff --git a/src/main/java/altk/comm/engine/Broadcast.java b/src/main/java/altk/comm/engine/Broadcast.java
index 14746c6..111af67 100644
--- a/src/main/java/altk/comm/engine/Broadcast.java
+++ b/src/main/java/altk/comm/engine/Broadcast.java
@@ -263,6 +263,14 @@ public abstract class Broadcast
                             myLogger.warn("Dispatcher thread interrupted while waiting to resume");
                         }
                     }
+                	if (serviceThreadsShouldStop()) 
+                	{
+                    	// Exit thread
+                    	myLogger.info("Thread terminating");
+                    	System.out.println(getName() + " terminating");
+                    	closeServiceProvider(serviceProviderPeer);
+                    	return;
+                	}
                 }
                 
                 // Get a batch of jobs, if available
@@ -277,12 +285,10 @@ public abstract class Broadcast
                         if (job == null) break;
                         batch.add(job);
                     }
-                }
-                if (batch.size()== 0)
-                {
-                    // wait for jobs	
-                	synchronized(readyQueue)
-                	{
+	
+	                if (batch.size()== 0)
+	                {
+	                	// wait for jobs	
                 		try
                 		{
                 			myLogger.debug("Waiting for jobs");
@@ -294,7 +300,8 @@ public abstract class Broadcast
                 		}
                 	}
                 }
-                else 
+
+                if (batch.size() > 0)
                 {
 	                // Process jobs.
 	                // Mark start time
@@ -846,7 +853,7 @@ public abstract class Broadcast
 	 */
 	protected int getActiveJobCount()
 	{
-	    return jobsTotal - completedJobCount - readyQueue.size();
+	    return serviceActivityCount;
 	}
 	
 	
@@ -910,14 +917,15 @@ public abstract class Broadcast
 	    wakeUpServiceThreads();
 	}
 
-	protected void pause()
+	protected void pause(PrintWriter out)
     {
 		// Sets state to PAUSING, which is monitored by Broadcast.Service threads.
 		// EVentually, when all service activity ends, the state transitions to PAUSED
 		setState(BroadcastState.PAUSING);
+		out.write("Broadcast paused");
     }
 	
-	protected void resume()
+	protected void resume(PrintWriter out)
 	{
 	    synchronized (resumeFlag)
 	    {
@@ -927,6 +935,7 @@ public abstract class Broadcast
 	            resumeFlag.notifyAll();
 	        }
 	    }
+	    out.write("Broadcast resumed");
 	}
 	/**
 	 * Derived class may make its own Implementation of JobReport