Refactor some code in worker model

java/base/src/main/java/base/work/Listen.java

@@ -6,15 +6,19 @@ import java.lang.invoke.MethodType;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
-import base.worker.DirectListener;
-import base.worker.ThreadListener;
+import base.exception.worker.ActivateException;
+import base.worker.ForegroundListener;
+import base.worker.BackgroundListener;
 import base.worker.Worker;
 import base.worker.pool.Listener;
 import base.worker.pool.ListenerPool;
 import base.worker.pool.PooledListener;
 
-public abstract class Listen<E> extends Work {
+public abstract class Listen<E> extends Work implements Listener<E> {
+	protected static final Worker.Type WORKER_TYPE = Worker.Type.DIRECT;
+
     protected Listener<E> listener;
+    protected Worker.Type workerType;
     public Queue<E> queue;
 
     public Listen() {
@@ -22,13 +26,15 @@ public abstract class Listen<E> extends Work {
     }
 
 	protected Listen(Worker.Type workerType) {
+		this.workerType = workerType;
 		switch (workerType) {
 			case DIRECT:
-				listener = new DirectListener<E>(this);
+				return;
+			case FOREGROUND:
+				listener = new ForegroundListener<E>(this);
 				break;
 			default:
-			case THREAD:
-				listener = new ThreadListener<E>(this);
+				listener = new BackgroundListener<E>(this);
 				break;
 		}
         queue = new ConcurrentLinkedQueue<E>();
@@ -46,10 +52,27 @@ public abstract class Listen<E> extends Work {
 	}
 
     public synchronized void add(E element) {
-    	listener.add(element);
+    	if (workerType.equals(Worker.Type.DIRECT)) {
+    		input(element);
+    	} else {
+    		listener.add(element);
+    	}
+    }
+
+    public void start() {
+    	if (workerType.equals(Worker.Type.DIRECT)) {
+    		try {
+				activate();
+			} catch (ActivateException e) {
+				logger.error("Failed to start directly", e);
+			}
+    	} else {
+    		super.start();
+    	}
     }
 
 	public void work() {
+		System.err.println(this.getClass().getName());
         while (!queue.isEmpty()) {
             input(queue.poll());
         }
@@ -61,12 +84,15 @@ public abstract class Listen<E> extends Work {
 	}
 
 	public void input(Object object) {
+		// This lookup should be cached
 		MethodType methodType = MethodType.methodType(void.class, object.getClass());
 		MethodHandles.Lookup lookup = MethodHandles.lookup();
 		MethodHandle methodHandle;
 		try {
 			methodHandle = lookup.findVirtual(getClass(), "input", methodType);
 			methodHandle.invoke(this, object);
+		} catch (Exception e) {
+			logger.error("", e);
 		} catch (Throwable e) {
 			logger.error("", e);
 		}

java/base/src/main/java/base/work/Work.java

@@ -12,8 +12,8 @@ import base.worker.pool.PooledWorker;
 import base.worker.pool.WorkerPool;
 
 public abstract class Work {
-	protected static final Worker.Type WORKER_TYPE = Worker.Type.THREAD;
-	
+	protected static final Worker.Type WORKER_TYPE = Worker.Type.BACKGROUND;
+
     protected Logger logger = LoggerFactory.getLogger(getClass());
 
 	protected Worker worker;
@@ -24,11 +24,10 @@ public abstract class Work {
 
 	protected Work(Worker.Type workerType) {
 		switch (workerType) {
-			case DIRECT:
+			case FOREGROUND:
 				worker = new DirectWorker(this);
 				break;
 			default:
-			case THREAD:
 				worker = new ThreadWorker(this);
 				break;
 		}
@@ -48,6 +47,7 @@ public abstract class Work {
 	}
 
 	public void start() {
+		logger.debug("Start work");
 		worker.start();
 	}
 
@@ -61,7 +61,7 @@ public abstract class Work {
 	}
 
 	public void exit() {
-		worker.exit();		
+		worker.exit();
 	}
 
 	public void activate() throws ActivateException {}

java/base/src/main/java/base/worker/BackgroundListener.java

@@ -3,15 +3,15 @@ package base.worker;
 import base.work.Listen;
 import base.worker.pool.Listener;
 
-public class ThreadListener<E> extends ThreadWorker implements Listener<E> {
+public class BackgroundListener<E> extends ThreadWorker implements Listener<E> {
 	protected Listen<E> listen;
 
-	public ThreadListener(Listen<E> listen) {
+	public BackgroundListener(Listen<E> listen) {
 		super(listen);
 		this.listen = listen;
 	}
 
-	public ThreadListener(Listen<E> listen, boolean start) {
+	public BackgroundListener(Listen<E> listen, boolean start) {
 		super(listen);
 	}
 

java/base/src/main/java/base/worker/DirectListener.java

@@ -1,16 +0,0 @@
-package base.worker;
-
-import base.work.Listen;
-import base.worker.pool.Listener;
-
-public class DirectListener<E> extends ThreadListener<E> implements Listener<E> {
-	public DirectListener(Listen<E> listen) {
-		super(listen, false);
-	}
-
-	@Override
-	public void add(Object element) {
-		// TODO Auto-generated method stub
-		
-	}
-}

java/base/src/main/java/base/worker/ForegroundListener.java

@@ -0,0 +1,10 @@
+package base.worker;
+
+import base.work.Listen;
+import base.worker.pool.Listener;
+
+public class ForegroundListener<E> extends BackgroundListener<E> implements Listener<E> {
+	public ForegroundListener(Listen<E> listen) {
+		super(listen, false);
+	}
+}

java/base/src/main/java/base/worker/IntervalWork.java

@@ -13,11 +13,11 @@ public abstract class IntervalWork extends Work {
 
 	protected IntervalWork(Worker.Type workerType) {
 		switch (workerType) {
-			case DIRECT:
+			case FOREGROUND:
 				worker = new DirectIntervalWorker(this);
 				break;
 			default:
-			case THREAD:
+			case BACKGROUND:
 				worker = new ThreadIntervalWorker(this);
 				break;
 		}
@@ -25,11 +25,11 @@ public abstract class IntervalWork extends Work {
 
 	protected IntervalWork(Worker.Type workerType, int interval) {
 		switch (workerType) {
-			case DIRECT:
+			case FOREGROUND:
 				worker = new DirectIntervalWorker(this, interval);
 				break;
 			default:
-			case THREAD:
+			case BACKGROUND:
 				worker = new ThreadIntervalWorker(this, interval);
 				break;
 		}

java/base/src/main/java/base/worker/ThreadWorker.java

@@ -40,9 +40,6 @@ public class ThreadWorker extends Worker implements Runnable {
 
     public synchronized void stop() {
     	super.stop();
-        if (active) {
-            deactivate = true;
-        }
         notifyAll();
     }
 

java/base/src/main/java/base/worker/Worker.java

@@ -9,7 +9,7 @@ import base.work.Work;
 
 public abstract class Worker {
 	public enum Type {
-		DIRECT, THREAD, POOLED
+		DIRECT, FOREGROUND, BACKGROUND, POOLED
 	}
 
     public static final int SLEEP = 100;
@@ -28,10 +28,10 @@ public abstract class Worker {
 	}
 
     public boolean active() {
-        return active;
+        return deactivate || active;
     }
 
-    public void run() {
+    public final void run() {
         while (run || deactivate) {
     		runActivate();
     		runDeactivate();
@@ -51,7 +51,7 @@ public abstract class Worker {
             }
         }
     }
-    
+
     public void runDeactivate() {
     	if (deactivate && active) {
             try {
@@ -96,8 +96,10 @@ public abstract class Worker {
 	public abstract void start();
 
 	public void stop() {
-		logger.debug("Stop worker");
-		activate = false;
+        if (active && !activate) {
+            deactivate = true;
+        }
+        activate = false;
 	}
 
     abstract public void exit();