Merge pull request Beh01der#1 from maximdim/master

Refactoring synchronized/notify constructs to CountdownLatch

Author: Beh01der

.gitignore

@@ -0,0 +1,4 @@
+/target
+/.classpath
+/.project
+/.settings

src/main/java/au/com/ds/ef/EasyFlow.java

@@ -9,7 +9,7 @@
 
 
 public class EasyFlow<C extends StatefulContext> {
-    private static Logger log = LoggerFactory.getLogger(EasyFlow.class);
+    static Logger log = LoggerFactory.getLogger(EasyFlow.class);
     protected State<C> startState;
     private C context;
     private Executor executor;
@@ -63,7 +63,6 @@ public void start(final C context) {
         }
     }
 
-    @SuppressWarnings("unchecked")
     protected void setCurrentState(final State<C> state, final C context) {
         execute(new Runnable() {
             @Override
@@ -176,11 +175,7 @@ protected void callOnFinalState(final State<C> state, final C context) {
                     log.debug("when final state {} for {} >>>", state, context);
             }
 
-            synchronized (context) {
-                callOnTerminate(context);
-                context.notifyAll();
-            }
-
+            callOnTerminate(context);
         } catch (Exception e) {
             callOnError(new ExecutionError(state, null, e,
                 "Execution Error in [EasyFlow.whenFinalState] handler", context));
@@ -202,13 +197,7 @@ public void waitForCompletion() {
     }
 
     public void waitForCompletion(C context) {
-        try {
-            synchronized (context) {
-                context.wait();
-            }
-        } catch (InterruptedException e) {
-            log.error("Error", e);
-        }
+      context.awaitTermination();
     }
 
     public EasyFlow<C> executor(Executor executor) {
@@ -229,11 +218,7 @@ protected void callOnError(final ExecutionError error) {
         if (onError != null) {
             onError.call(error);
         }
-
-        synchronized (error.getContext()) {
-            callOnTerminate((C) error.getContext());
-            error.getContext().notifyAll();
-        }
+        callOnTerminate((C) error.getContext());
     }
 
     protected void callOnTerminate(final C context) {

src/main/java/au/com/ds/ef/StatefulContext.java

@@ -1,16 +1,18 @@
 package au.com.ds.ef;
 
 import java.io.Serializable;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 @SuppressWarnings("rawtypes")
 public class StatefulContext implements Serializable {
 	private static final long serialVersionUID = 2324535129909715649L;
 	private static long idCounter = 1;
 
-	private String id;
+	private final String id;
 	private State state;
-	private AtomicBoolean terminated = new AtomicBoolean(false);
+	private final AtomicBoolean terminated = new AtomicBoolean(false);
+	private final CountDownLatch completionLatch = new CountDownLatch(1);
 
 	public StatefulContext() {
 		id = newId() + ":" + getClass().getSimpleName();
@@ -69,8 +71,21 @@ public boolean isStarted() {
 
 	protected void setTerminated() {
 		this.terminated.set(true);
+		this.completionLatch.countDown();
 	}
 
+	/**
+	 * Block current thread until Context terminated
+	 */
+	protected void awaitTermination() {
+	  try {
+      this.completionLatch.await();
+    } 
+	  catch (InterruptedException e) {
+	    Thread.interrupted();
+	  }
+	}
+	
     @Override
     public String toString() {
         return id;

src/test/java/au/com/ds/ef/SynchronizationTest.java

@@ -0,0 +1,95 @@
+package au.com.ds.ef;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import au.com.ds.ef.call.StateHandler;
+
+/**
+ * With original synchronize()/notifyAll() approach this test never finishes
+ * (thread(s) still waiting on notification)
+ */
+public class SynchronizationTest {
+  void doRun() {
+    ExecutorService threadPool = Executors.newCachedThreadPool();
+    for(int i = 0; i < 5; i++) {
+      Node n = new Node();
+      threadPool.execute(n);
+    }
+    threadPool.shutdown();
+  }
+
+  public static void main(String[] args) {
+    new SynchronizationTest().doRun();
+  }
+  
+  static class Node implements Runnable {
+    final Event<StatefulContext> onInitialize = FlowBuilder.event("onInitialize");
+    final Event<StatefulContext> onTerminate = FlowBuilder.event("onTerminate");
+
+    final EasyFlow<StatefulContext> flow;
+    
+    Node() {
+      final State<StatefulContext> uninitialized = FlowBuilder.state("unitialized");
+      final State<StatefulContext> running = FlowBuilder.state("running");
+      final State<StatefulContext> done = FlowBuilder.state("done");
+
+      uninitialized.whenEnter(new StateHandler<StatefulContext>() {
+        @Override
+        public void call(State<StatefulContext> state, StatefulContext context) throws Exception {
+          System.out.println(getThreadName()+" unitialized:Enter");
+          unitializedEnter(state, context);
+        }
+      });
+      running.whenEnter(new StateHandler<StatefulContext>() {
+        @Override
+        public void call(State<StatefulContext> state, StatefulContext context) throws Exception {
+          System.out.println(getThreadName()+" runnig:Enter");
+          runningEnter(state, context);
+        }
+      });
+
+      this.flow = FlowBuilder.from(uninitialized)
+          .transit(onInitialize.to(running).transit(onTerminate.finish(done)));
+      this.flow.executor(new SyncExecutor());
+      
+      flow.whenFinalState(new StateHandler<StatefulContext>() {
+        @Override
+        public void call(State<StatefulContext> state, StatefulContext context) throws Exception {
+          System.out.println(getThreadName()+" final");
+        }
+      });
+    }
+
+    void unitializedEnter(State<StatefulContext> state, StatefulContext context) {
+      onInitialize.trigger(context);
+    }
+
+    void runningEnter(State<StatefulContext> state, StatefulContext context) {
+      for(int i = 0; i < 10; i++) {
+        System.out.println(getThreadName()+" running");
+        sleep(1000);
+      }
+      onTerminate.trigger(context);
+    }
+    
+    @Override
+    public void run() {
+      flow.validate().trace().start(new StatefulContext());
+      flow.waitForCompletion();
+      System.out.println("Run method completed");
+    }
+  }
+  
+  static String getThreadName() {
+    return Thread.currentThread().getName();
+  }
+  
+  static void sleep(long ms) {
+    try {
+      Thread.sleep(ms);
+    } catch (InterruptedException e) {
+      Thread.interrupted();
+    }
+  }
+}