bug6278: harmonized listeners and pending action counting;

introduced abstract base class for AsyncActionsExecutor and
TimeRangeActionsExecutor

java/com.sap.sse.gwt/src/com/sap/sse/gwt/client/async/AbstractActionsExecutor.java

@@ -0,0 +1,39 @@
+package com.sap.sse.gwt.client.async;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Can execute actions and keeps track of the number of pending actions.
+ * Listeners can register to get informed about changes in the number of
+ * pending actions.
+ * 
+ * @author Axel Uhl (d043530)
+ *
+ */
+public abstract class AbstractActionsExecutor {
+    protected int numberOfPendingActions;
+    
+    public static interface Listener {
+        void onNumberOfPendingCallsChanged(int newNumberOfPendingCalls);
+    }
+    
+    private final Set<Listener> listeners;
+
+    protected AbstractActionsExecutor() {
+        this.listeners = new HashSet<>();
+    }
+
+    public void addListener(Listener listener) {
+        listeners.add(listener);
+    }
+    
+    public void removeListener(Listener listener) {
+        listeners.remove(listener);
+    }
+    
+    protected void notifyListeners() {
+        listeners.forEach(l->l.onNumberOfPendingCallsChanged(numberOfPendingActions));
+    }
+
+}

java/com.sap.sse.gwt/src/com/sap/sse/gwt/client/async/AsyncActionsExecutor.java

@@ -27,13 +27,9 @@ import com.sap.sse.common.impl.MillisecondsTimePoint;
  * 
  * @author c5163874, Simon Marcel Pamies
  */
-public class AsyncActionsExecutor {
+public class AsyncActionsExecutor extends AbstractActionsExecutor {
     private final static Logger logger = Logger.getLogger(AsyncActionsExecutor.class.getName());
     
-    public static interface Listener {
-        void onNumberOfPendingCallsChanged(int newNumberOfPendingCalls);
-    }
-    
     private class ExecutionJob<T> implements AsyncCallback<T> {
         private final AsyncAction<T> action;
         private final String category;
@@ -104,8 +100,6 @@ public class AsyncActionsExecutor {
     private final Map<String, Set<Runnable>> runAfterLastActionForCategoryReturned;
     private final Set<Runnable> runAfterLastActionReturned;
     
-    private int numPendingCalls = 0;
-    private final Set<Listener> listeners;
     private TimePoint timePointOfFirstExecutorInit = null;
 
     public AsyncActionsExecutor() {
@@ -114,7 +108,6 @@ public class AsyncActionsExecutor {
     }
     
     public AsyncActionsExecutor(int maxPendingCalls, int maxPendingCallsPerType, Duration durationAfterToResetQueue) {
-        this.listeners = new HashSet<>();
         if (maxPendingCalls < maxPendingCallsPerType) {
             throw new RuntimeException("The number of max pending calls can not be lower than the number of max pending calls per type.");
         }
@@ -129,14 +122,6 @@ public class AsyncActionsExecutor {
         this.timePointOfFirstExecutorInit = MillisecondsTimePoint.now(); // triggering duration to reset
     }
     
-    public void addListener(Listener listener) {
-        listeners.add(listener);
-    }
-    
-    public void removeListener(Listener listener) {
-        listeners.remove(listener);
-    }
-    
     /**
      * If there are no calls for the {@code category} whose results are still outstanding, the {@code callback}
      * is invoked immediately. Otherwise, the {@code callback} is stored and will be invoked once there are
@@ -178,12 +163,12 @@ public class AsyncActionsExecutor {
     }
     
     public int getNumberOfPendingActions() {
-        return numPendingCalls;
+        return numberOfPendingActions;
     }
     
     private void execute(ExecutionJob<?> job) {
         Integer numActionsOfType = actionsPerTypeCounter.computeIfAbsent(job.getType(), j->Integer.valueOf(0));
-        if (numPendingCalls >= maxPendingCalls || (numActionsOfType >= maxPendingCallsPerType)) {
+        if (numberOfPendingActions >= maxPendingCalls || (numActionsOfType >= maxPendingCallsPerType)) {
             final TimePoint now = MillisecondsTimePoint.now();
             final TimePoint timePointToInspectForResetDecision = timePointOfTypeLastBeingExecuted.get(job.getType()) != null ?
                     timePointOfTypeLastBeingExecuted.get(job.getType()) : timePointOfFirstExecutorInit;
@@ -191,7 +176,7 @@ public class AsyncActionsExecutor {
                     now.minus(durationAfterToResetQueue).after(timePointToInspectForResetDecision)) {
                 logger.info("Resetting number of pending calls after "+durationAfterToResetQueue+" of no response");
                 // reset the number of pending calls
-                numPendingCalls = 0;
+                numberOfPendingActions = 0;
                 // reset number of pending actions per type - 0 is fine as checkForEmptyCallQueue
                 // will check for a number less than maxPendingCallsPerType to send out the
                 // last job pending for a given type
@@ -216,15 +201,11 @@ public class AsyncActionsExecutor {
             }
         }
         actionsPerTypeCounter.put(job.getType(), numActionsOfType+1);
-        numPendingCalls++;
+        numberOfPendingActions++;
         job.execute();
         notifyListeners();
     }
 
-    private void notifyListeners() {
-        listeners.forEach(l->l.onNumberOfPendingCallsChanged(numPendingCalls));
-    }
-
     private void callCompleted(ExecutionJob<?> job) {
         String type = job.getType();
         Integer numActionsPerType = actionsPerTypeCounter.get(type);
@@ -238,12 +219,12 @@ public class AsyncActionsExecutor {
             }
 
         }
-        if (numPendingCalls > 0) { // don't let it go negative; we may have reset the queue
+        if (numberOfPendingActions > 0) { // don't let it go negative; we may have reset the queue
             // and then one or more pending requests may have returned late
-            numPendingCalls--;
+            numberOfPendingActions--;
         }
         notifyListeners();
-        if (numPendingCalls == 0) {
+        if (numberOfPendingActions == 0) {
             runAfterLastActionReturned.forEach(callback->callback.run());
         }
         timePointOfTypeLastBeingExecuted.put(type, MillisecondsTimePoint.now());

java/com.sap.sse.gwt/src/com/sap/sse/gwt/client/async/TimeRangeActionsExecutor.java

@@ -48,11 +48,7 @@ import com.sap.sse.common.Util.Pair;
  *
  * @author Tim Hessenmüller (D062243)
  */
-public class TimeRangeActionsExecutor<Result, SubResult, Key> {
-    public static interface Listener {
-        void onNumberOfPendingCallsChanged(int newNumberOfPendingCalls);
-    }
-    
+public class TimeRangeActionsExecutor<Result, SubResult, Key> extends AbstractActionsExecutor {
     /**
      * Callback called by {@link TimeRangeActionsExecutor#executor} upon receiving an answer from the server for a
      * potentially trimmed request. The compound {@code Result} is {@link TimeRangeAsyncCallback#unzipResult(Object)
@@ -151,14 +147,6 @@ public class TimeRangeActionsExecutor<Result, SubResult, Key> {
 
     private final Map<Key, TimeRangeResultCache<SubResult>> cacheMap = new HashMap<>();
     
-    private int numberOfPendingActions;
-    
-    private final Set<Listener> listeners;
-
-    public TimeRangeActionsExecutor() {
-        this.listeners = new HashSet<>();
-    }
-
     /**
      * Executes a {@link TimeRangeAsyncAction} and returns the results to a {@link TimeRangeAsyncCallback}. Calls
      * {@link #execute(TimeRangeAsyncAction, AsyncCallback, boolean)} with {@code forceTimeRange} set to {@code false},
@@ -223,16 +211,4 @@ public class TimeRangeActionsExecutor<Result, SubResult, Key> {
     private TimeRangeResultCache<SubResult> getSubResultCache(Key key) {
         return cacheMap.computeIfAbsent(key, k->new TimeRangeResultCache<>());
     }
-
-    public void addListener(Listener listener) {
-        listeners.add(listener);
-    }
-    
-    public void removeListener(Listener listener) {
-        listeners.remove(listener);
-    }
-    
-    private void notifyListeners() {
-        listeners.forEach(l->l.onNumberOfPendingCallsChanged(numberOfPendingActions));
-    }
 }