Memory leak, user display improvement.

Broadcast state ALLDONE added to mark the event of completing postback,
and when all threads are to be terminated.

CommEngine report of a broadcast adds service times,
and number of transactions.

src/main/java/altk/comm/engine/Broadcast.java

@@ -44,6 +44,8 @@ public abstract class Broadcast
     String haltReason;
     String stateErrorText;
     public final long receiveTime;
+    public long serviceStartTime;
+    public long serviceEndTime;
     public long changeStateTime;
     protected String activityRecordIdParamName;
     private String jobReportRootNodeName;
@@ -69,11 +71,11 @@ public abstract class Broadcast
     protected List<Service> serviceThreadPool;
 	private Object resumeFlag; // Semaphore for dispatcher threads to resume.
     protected List<Recipient> recipientList;
-    private int completedJobCount;
     
     private ScheduledExecutorService scheduler;
 	private int jobsTotal;
 	private int scheduledJobs;
+	private Integer transactions;
 	
 	private Integer serviceActivityCount;
 	
@@ -88,7 +90,8 @@ public abstract class Broadcast
         PURGED(true),     // Final state
         ABORTED(true),    // final state
         EXPIRED(true),    // final state
-        COMPLETED(true);   // Final state
+        COMPLETED,
+        ALLDONE(true);   // Final state
         
         final public boolean isFinal;
         
@@ -207,6 +210,9 @@ public abstract class Broadcast
                 BroadcastState.CANCELING, // User action
                 BroadcastState.PURGED     // User action
             ));
+        toStates.put(BroadcastState.COMPLETED,  Arrays.asList(
+        		BroadcastState.ALLDONE  // when all posting back is complete
+        		));
     }
     
     public static class StateChangeResult
@@ -224,6 +230,13 @@ public abstract class Broadcast
         }
     }
     
+    protected enum PostbackThreadActionOnEmpty
+    {
+    	CONTINUE,
+    	STOP,
+    	WAIT
+    }
+    
     protected class Service extends Thread
     {
         Object serviceProviderPeer;
@@ -319,7 +332,8 @@ public abstract class Broadcast
 	                try
 	                {
 	                	updateServiceActivityCount(1);
-	                    processJobs(batch, serviceProviderPeer, prerequisites);
+	                    int transactions = processJobs(batch, serviceProviderPeer, prerequisites);
+	                    incrementTransactions(transactions);
 	                    updateServiceActivityCount(-1);
 	                }
 	                catch (EngineException e)
@@ -363,7 +377,6 @@ public abstract class Broadcast
         this.jobReportRootNodeName = jobReportRootNodeName;
         
         postback = null;
-        completedJobCount = 0;
         sleepBetweenJobs = SLEEP_BETWEEN_JOBS_DEFAULT;
         readyQueue = new LinkedBlockingQueue<Job>();
         serviceThreadPool = new ArrayList<Service>();
@@ -373,8 +386,16 @@ public abstract class Broadcast
         
         receiveTime = System.currentTimeMillis();
         serviceActivityCount = Integer.valueOf(0);
+        transactions = Integer.valueOf(0);
     }
     
+    private void incrementTransactions(int delta)
+    {
+    	synchronized (transactions)
+    	{
+    		transactions += delta;
+    	}
+    }
     public int pendingJobs()
     {
     	return readyQueue.size() + scheduledJobs;
@@ -580,19 +601,14 @@ public abstract class Broadcast
             this.stateErrorText = stateErrorText;
             
             CommonLogger.activity.info(String.format("Broadcast %s: State transitioned from %s to %s", broadcastId, state, newState));
+            state = newState;
+            changeStateTime = System.currentTimeMillis();
+            if (state == BroadcastState.RUNNING) serviceStartTime = changeStateTime;
+            if (prev == BroadcastState.RUNNING) serviceEndTime = changeStateTime;
             if (postback != null)
             {
-            	//synchronized(postback.postQueue)
-                {
-                   postback.queueReportFirst(mkStatusReport(newState));
-                   state = newState;
-            	}
-            }
-            else
-            {
-                state = newState;
+               postback.queueReportFirst(mkStatusReport());
             }
-            changeStateTime = System.currentTimeMillis();
             return new StateChangeResult(StateChangeStatus.SUCCESS, newState, prev);
         }
         else
@@ -781,17 +797,6 @@ public abstract class Broadcast
 	 * @return
 	 */
 	protected String mkStatusReport()
-    {
-		return mkStatusReport(state);
-    }
-	
-	/**
-	 * Creates status report.  Sometimes, we need to create the report before
-	 * actually changing BroadcastState.
-	 * @param state - BroadcastState for this report, which is not necessarily the same as the class attribute state.
-	 * @return status report in XML.
-	 */
-    protected String mkStatusReport(BroadcastState state)
     {
         StringBuffer statusBf = new StringBuffer();
 		String topLevelTag = broadcastType;
@@ -804,7 +809,13 @@ public abstract class Broadcast
 		{
 			statusBf.append(" launch_record_id='" + launchRecordId + "'");
 		}
-		statusBf.append(">\r\n<state>" + state + "</state><state_change_time>" + changeStateTime
+		statusBf.append(">\r\n");
+		statusBf.append("<current_time>" + System.currentTimeMillis() + "</current_time>\r\n");
+		if (serviceStartTime > 0) statusBf.append("<service_start_time>" + serviceStartTime
+				+ "</service_start_time>\r\n");
+		if (serviceEndTime > 0) statusBf.append("<service_end_time>" + serviceEndTime
+				+ "</service_end_time>\r\n");
+		statusBf.append("<state>" + state + "</state><state_change_time>" + changeStateTime
 		        + "</state_change_time>\r\n");
 		if (state == BroadcastState.PAUSED
 		        || state == BroadcastState.ABORTED)
@@ -821,6 +832,7 @@ public abstract class Broadcast
 				        + "</error_text>");
 			}
 		}
+		statusBf.append("<transactions>" + transactions + "</transactions>");
 		statusBf.append("<job_summary completed='" + getCompletedJobCount() +
 		        "' ready='" + getPendingJobCount() + "'");
 		statusBf.append(" active='" + getActiveJobCount() + "'");
@@ -988,6 +1000,7 @@ public abstract class Broadcast
         destroyResources();
 		postback.wrapup();
 	    postback = null;
+	    //setState(BroadcastState.ALLDONE);
         myLogger.info("Broadcast " + getId() + " terminated");
     }
 	
@@ -1075,8 +1088,9 @@ public abstract class Broadcast
 	 * if all service threads are idle or terminated.
 	 * @param batch
 	 * @param prerequisites 
+	 * @return int - number of transactions employed to service these jobs.
 	 */
-	abstract protected void processJobs(List<Job> batch, Object serviceProvider, ServicePrerequisites prerequisites)
+	abstract protected int processJobs(List<Job> batch, Object serviceProvider, ServicePrerequisites prerequisites)
 	        throws EngineException;
 
 	/**
@@ -1141,7 +1155,7 @@ public abstract class Broadcast
                 )  
         {
         	// No more rescheduling on cancel, expire, or pause
-            completedJobCount++;
+            //completedJobCount++;
             //logJobCount("Completed a job");
         }
         else if (rescheduleTimeMS == 0)
@@ -1162,14 +1176,44 @@ public abstract class Broadcast
 	 */
 	private void logJobCount(String title)
     {
-        myLogger.debug(String.format("%s: completed: %d, active: %d, ready: %d, scheduled %d, total jobs: %d, remaining %d",
+		if (postback == null) {
+			myLogger.debug(title + ": postback = null");
+	        myLogger.debug(String.format("%s: state %s, completed: %d, active: %d, ready: %d, scheduled: %d, total jobs: %d, remaining: %d, postQueue: ",
+	                title,
+	                state,
+	                getCompletedJobCount(),
+	                getActiveJobCount(),
+	                readyQueue.size(),
+	                scheduledJobs,
+	                jobsTotal,
+	                getRemainingJobCount()
+	                ));
+			return;
+		}
+		if (postback.postQueue == null) {
+			myLogger.debug(title + ": postback.postQueue = null");
+	        myLogger.debug(String.format("%s: state %s, completed: %d, active: %d, ready: %d, scheduled: %d, total jobs: %d, remaining: %d, postQueue: ",
+	                title,
+	                state,
+	                getCompletedJobCount(),
+	                getActiveJobCount(),
+	                readyQueue.size(),
+	                scheduledJobs,
+	                jobsTotal,
+	                getRemainingJobCount()
+	                ));
+			return;
+		}
+        myLogger.debug(String.format("%s: state %s, completed: %d, active: %d, ready: %d, scheduled %d, total jobs: %d, remaining: %d, postQueue: %d",
                 title,
-                completedJobCount,
+                state,
+                getCompletedJobCount(),
                 getActiveJobCount(),
                 readyQueue.size(),
                 scheduledJobs,
                 jobsTotal,
-                getRemainingJobCount()
+                getRemainingJobCount(),
+                postback.postQueue.size()
                 ));
     }
 
@@ -1209,11 +1253,26 @@ public abstract class Broadcast
 
     public int getCompletedJobCount()
     {
-        return completedJobCount;
+        return jobsTotal - pendingJobs() - serviceActivityCount;
     }
 
 	public String getBroadcastType() {
 		return broadcastType;
 	}
+
+	public PostbackThreadActionOnEmpty getPostbackThreadActionOnEmpty() {
+		logJobCount("getPostbackThreadActionOnEmpty");
+		if (state==BroadcastState.CANCELED || state==BroadcastState.ALLDONE) return PostbackThreadActionOnEmpty.STOP;
+		if (setState(BroadcastState.ALLDONE).stateChangeStatus == StateChangeStatus.SUCCESS) return PostbackThreadActionOnEmpty.CONTINUE;
+		return PostbackThreadActionOnEmpty.WAIT;
+	}
+	
+	public boolean postbackThreadsShouldStop() {
+		if (state==BroadcastState.CANCELED || state==BroadcastState.ALLDONE) return true;
+	    boolean shouldContinue = state==BroadcastState.COMPLETED && setState(BroadcastState.ALLDONE).stateChangeStatus == StateChangeStatus.SUCCESS;
+		return !shouldContinue;
+		//boolean doStop = setState(BroadcastState.ALLDONE).stateChangeStatus != StateChangeStatus.SUCCESS;
+		//return doStop;
+	}
 	
 }

src/main/java/altk/comm/engine/Postback.java

@@ -43,6 +43,8 @@ import org.apache.log4j.Logger;
 import org.w3c.dom.Document;
 import org.w3c.dom.Node;
 
+import altk.comm.engine.Broadcast.PostbackThreadActionOnEmpty;
+
 
 /**
  * Queues JobReports to be posted back to attribute postBackURL.
@@ -127,6 +129,8 @@ public class Postback
                         if (report == null) break;
                         reportList.add(report);
                     }
+                    // If space in que is generated, wake up all service queues waiting for space
+                    if (reportList.size() > 0 && threadsWaitingToPost > 0) postQueue.notifyAll();
                 } // synchronized()
                 if (reportList.size() > 0)
                 {
@@ -154,22 +158,29 @@ public class Postback
                 else
                 {
                 	// empty post queue
-                    if (threadsShouldStop())
+                    PostbackThreadActionOnEmpty postbackThreadActionOnEmpty = broadcast.getPostbackThreadActionOnEmpty();
+					myLogger.debug("Action on empty queue: " + postbackThreadActionOnEmpty);
+                    switch (postbackThreadActionOnEmpty)
                     {
+                    case STOP:
                         myLogger.info(getName() + " terminating");
                         System.out.println(getName() + " terminating");
                         return;
+                    case WAIT:
+	                	synchronized (postQueue)
+	                	{
+	                		try 
+	                		{
+								postQueue.wait();
+							} catch (InterruptedException e) 
+	                		{
+								// Do nothing
+							}
+	                	}
+	                    break;
+                    case CONTINUE:
+                    default:
                     }
-                	synchronized (postQueue)
-                	{
-                		try 
-                		{
-							postQueue.wait();
-						} catch (InterruptedException e) 
-                		{
-							// Do nothing
-						}
-                	}
                 }
             }
         }
@@ -280,7 +291,6 @@ public class Postback
                 CommonLogger.alarm.warn("While decoding post back response from server: " + e);
                 return PostbackStatus.IRRECOVERABLE_ERROR;
             }
-            myLogger.debug("returned from posting");
             return PostbackStatus.SUCCESS;
         }
 
@@ -402,18 +412,17 @@ public class Postback
                 }
                 else
                 {
-                    myLogger.debug("Waiting for space - postQueue size: " + postQueue.size());
-                    try
+                     try
                     {
                         threadsWaitingToPost++; 
-                        myLogger.debug("Threads waiting to post: " + threadsWaitingToPost);
+                        myLogger.debug(String.format("Waiting to post report - postqueue size %d, total waiting %d", postQueue.size(), threadsWaitingToPost));
                         postQueue.wait(QUEUE_WAIT * 1000);
+                        threadsWaitingToPost--;
                     }
                     catch (InterruptedException e)
                     {
                         break;
                     }
-                    threadsWaitingToPost--;
                 }
             }
         }
@@ -423,7 +432,8 @@ public class Postback
     
     private boolean threadsShouldStop()
     {
-    	return (shutdownWhenDone); // && postedTransactions == broadcast.transactions);
+    	//return (shutdownWhenDone); // && postedTransactions == broadcast.transactions);
+    	return broadcast.postbackThreadsShouldStop();
     }
     
     public void wrapup()