thread pool for tasks

Author: lgajos

src/main/java/notepack/app/domain/MessageBus.java

@@ -14,6 +14,12 @@
 import notepack.app.task.TypeGui;
 import notepack.app.task.TypeNote;
 import notepack.app.task.TypeNotepad;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import notepack.app.domain.exception.MessageError;
 
 public class MessageBus {
 
@@ -24,13 +30,17 @@ public class MessageBus {
 
     private Thread dispatchThread;
     private boolean dispatcherStop = false;
+    private final AtomicInteger tasksActive = new AtomicInteger();
+    private int maxTasksActive = 3;
 
     public MessageBus() {
         tasks = new ConcurrentLinkedQueue<>();
 
         noteListeners = new ArrayList<>();
         notepadListeners = new ArrayList<>();
         guiListeners = new ArrayList<>();
+
+        tasksActive.set(0);
     }
 
     public void startDispatcher() {
@@ -53,7 +63,7 @@ public void run() {
                         tasks.add(new ShowUserMessage(e.getMessage(), ShowUserMessage.TYPE.ERROR));
                     }
                     try {
-                        Thread.sleep(10);
+                        Thread.sleep(100);
                     } catch (InterruptedException ex) {
                         break;
                     }
@@ -70,38 +80,61 @@ public void stopDispatcher() {
         dispatchThread = null;
     }
 
+    synchronized private Task getTaskToDispatch() {
+        return tasks.poll();
+    }
+
     private void dispatch() throws MessageError {
 
-        for (Task t : tasks) {
-            
-            if (t instanceof BaseTask) {
-                ((BaseTask) t).setMessageBus(this);
-            }
-            
-            tasks.remove(t);
+        int currentTasksCounter = tasksActive.get();
+        if (currentTasksCounter > maxTasksActive) {
+            return;
+        }
 
-            if (t instanceof TypeNote) {
-                t.dispatch();
+        Task t = getTaskToDispatch();
+        if (t == null) {
+            return;
+        }
+        if (t instanceof BaseTask) {
+            ((BaseTask) t).setMessageBus(this);
+        }
 
-                for (NoteListener l : noteListeners) {
-                    ((TypeNote) t).notify(l);
-                }
-            }
-            if (t instanceof TypeNotepad) {
-                t.dispatch();
+        Thread job = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                tasksActive.incrementAndGet();
 
-                for (NotepadListener l : notepadListeners) {
-                    ((TypeNotepad) t).notify(l);
-                }
-            }
-            if (t instanceof TypeGui) {
-                for (GuiListener l : guiListeners) {
-                    l.proceed((TypeGui) t);
+                try {
+                    if (t instanceof TypeNote) {
+                        t.dispatch();
+
+                        for (NoteListener l : noteListeners) {
+                            ((TypeNote) t).notify(l);
+                        }
+                    }
+                    if (t instanceof TypeNotepad) {
+                        t.dispatch();
+
+                        for (NotepadListener l : notepadListeners) {
+                            ((TypeNotepad) t).notify(l);
+                        }
+                    }
+                    if (t instanceof TypeGui) {
+                        for (GuiListener l : guiListeners) {
+                            l.proceed((TypeGui) t);
+                        }
+                    }
+                } catch (MessageError ex) {
+                    Logger.getLogger(MessageBus.class.getName()).log(Level.SEVERE, null, ex);
+                    tasks.add(new ShowUserMessage(ex.getMessage(), ShowUserMessage.TYPE.ERROR));
                 }
-            }
 
-        }
+                tasksActive.decrementAndGet();
+            }
 
+        });
+        job.setName("Task processing " + t.toString());
+        job.start();
     }
 
     public void registerNoteListener(NoteListener l) {