Adding ready() to IntentAccumulator

Change-Id: I34dc32c1a416629afbf3ba1f5b65d6a764e55fd6

@@ -42,6 +42,8 @@ public class IntentAccumulator extends AbstractAccumulator<IntentData> {
 
 
     private final IntentBatchDelegate delegate;
     private final IntentBatchDelegate delegate;
 
 
+    private volatile boolean ready;
+
     /**
     /**
      * Creates an intent operation accumulator.
      * Creates an intent operation accumulator.
      *
      *
@@ -50,13 +52,14 @@ public class IntentAccumulator extends AbstractAccumulator<IntentData> {
     protected IntentAccumulator(IntentBatchDelegate delegate) {
     protected IntentAccumulator(IntentBatchDelegate delegate) {
         super(TIMER, DEFAULT_MAX_EVENTS, DEFAULT_MAX_BATCH_MS, DEFAULT_MAX_IDLE_MS);
         super(TIMER, DEFAULT_MAX_EVENTS, DEFAULT_MAX_BATCH_MS, DEFAULT_MAX_IDLE_MS);
         this.delegate = delegate;
         this.delegate = delegate;
+        // Assume that the delegate is ready for work at the start
+        ready = true; //TODO validate the assumption that delegate is ready
     }
     }
 
 
     @Override
     @Override
     public void processItems(List<IntentData> items) {
     public void processItems(List<IntentData> items) {
+        ready = false;
         delegate.execute(reduce(items));
         delegate.execute(reduce(items));
-        // FIXME kick off the work
-        //for (IntentData data : opMap.values()) {}
     }
     }
 
 
     private Collection<IntentData> reduce(List<IntentData> ops) {
     private Collection<IntentData> reduce(List<IntentData> ops) {
@@ -67,4 +70,13 @@ public class IntentAccumulator extends AbstractAccumulator<IntentData> {
         //TODO check the version... or maybe store will handle this.
         //TODO check the version... or maybe store will handle this.
         return map.values();
         return map.values();
     }
     }
+
+    @Override
+    public boolean isReady() {
+        return ready;
+    }
+
+    public void ready() {
+        ready = true;
+    }
 }
 }

@@ -317,6 +317,7 @@ public class IntentManager
                 //FIXME
                 //FIXME
 //            batchService.removeIntentOperations(data);
 //            batchService.removeIntentOperations(data);
             }
             }
+            accumulator.ready();
         }
         }
 
 
         private List<Future<FinalIntentProcessPhase>> createIntentUpdates() {
         private List<Future<FinalIntentProcessPhase>> createIntentUpdates() {