MANIFEST.MF

@@ -1,6 +1,6 @@
 Manifest-Version: 1.0
 Ant-Version: Apache Ant 1.9.7
 Created-By: 1.8.0_191-b12 (Oracle Corporation)
-Implementation-Version: 1.11
+Implementation-Version: 1.13
 Built-By: jgregor5
 

build.properties

@@ -1,3 +1,3 @@
 
 builder=jgregor5
-version=1.11
+version=1.18

build.xml

@@ -87,6 +87,21 @@
         </jar>
     </target>
 
+    <target name="commander-src" depends="compile">
+
+        <manifest file="MANIFEST.MF">
+            <attribute name="Implementation-Version" value="${version}"/> 
+            <attribute name="Built-By" value="${builder}"/>
+        </manifest>
+
+        <jar destfile="dist/commander-src-${version}.jar" manifest="MANIFEST.MF">
+            <fileset dir="src">
+                <include name="client/**"/>
+                <include name="commander/**"/>
+            </fileset>
+        </jar>
+    </target>
+
     <target name="commander" depends="compile">
 
         <manifest file="MANIFEST.MF">

src/client/CommanderClient.java

@@ -7,6 +7,7 @@
 import java.net.Socket;
 import org.json.JSONObject;
 import client.LoggingClient.IStreamListener;
+import commander.ComponentManager;
 import commander.IEventListener;
 import commander.IManager;
 import java.io.File;
@@ -29,6 +30,7 @@ public class CommanderClient implements IManager, IStreamListener {
 
     private final String host;
 
+    private int cport, lport;
     private Socket socket;
     private BufferedReader br;
     private PrintWriter pw;
@@ -37,6 +39,8 @@ public class CommanderClient implements IManager, IStreamListener {
 
     public CommanderClient(String host) { 
         this.host = host;
+        this.cport = 0;
+        this.lport = 0;
         this.listeners = new HashSet<>();
         configureLogging();
     }
@@ -46,35 +50,54 @@ public CommanderClient(String host) {
     @Override
     public JSONObject execute(JSONObject command) {
 
+        openIfNeeded();
+
         try {
             LOGGER.log(Level.FINER, "sending {0}", command);
             this.pw.println(command);
             String response = this.br.readLine();
             LOGGER.log(Level.FINER, "received {0}", response);
             return new JSONObject(response);
         } catch (IOException ex) {
+            close(); // remove connection references
             throw new IORuntimeException(ex);
         }
     }
 
+    private void openIfNeeded() {
+
+        if (this.logging == null && this.cport > 0) {
+            connect(this.cport);
+        }
+
+        if (this.logging == null && this.lport > 0) {
+            listen(this.lport);
+        }
+    }
+
     public void listen(int port) {
-        this.logging = new LoggingClient(this.host, port, this);
-        new Thread(logging, "commander-client").start();
+        this.lport = port;
+        this.logging = new LoggingClient(this.host, this.lport, this);
+        new Thread(this.logging, "commander-client").start();
     }
 
     public void unlisten() {
         this.logging.shutdown();
+        this.logging = null;
     }
 
     public void connect(int port) {
-
+        this.cport = port;       
         try {
-            this.socket = new Socket(this.host, port);
+            this.socket = new Socket(this.host, this.cport);
             this.socket.setSoTimeout(SO_TIMEOUT_MILLIS);
             this.br = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));            
             this.pw = new PrintWriter(socket.getOutputStream(), true); // autoflush
 
         } catch (IOException ex) {
+            this.br = null;
+            this.pw = null;
+            this.socket = null;
             throw new IORuntimeException(ex);
         }
     }
@@ -88,6 +111,10 @@ public void disconnect() {
 
         } catch (IOException ex) {
             throw new IORuntimeException(ex);
+        } finally {
+            this.br = null;
+            this.pw = null;
+            this.socket = null;
         }
     }
 

src/commander/ComponentManager.java

@@ -9,7 +9,10 @@
 import java.util.Map;
 import java.util.ServiceLoader;
 import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.LogManager;
 import java.util.logging.Logger;
@@ -23,21 +26,34 @@ public class ComponentManager implements IEventListener, IManager {
 
     private final static Logger LOGGER = Logger.getLogger(ComponentManager.class.getName());
 
+    private static final long POLL_TIMEOUT = 5000;
+
     public final static int COMMAND_PORT = 9000;
     public final static int LISTEN_PORT = 9001;
 
     private static ComponentManager service;
     private ServiceLoader<IComponent> loader;
     private Map<String, IComponent> commands;
     private Set<IEventListener> listeners;
+
+    private QueueConsumer consumer;
+    private Thread queueThread;
+    private BlockingQueue<JSONObject> queue;    
 
-    private ComponentManager() {        
+    private ComponentManager() { 
+
         this.loader = ServiceLoader.load(IComponent.class);
         this.listeners = new CopyOnWriteArraySet<>();
+
+        this.queue = new ArrayBlockingQueue<>(1024);
+        this.consumer = new QueueConsumer();
+        this.queueThread = new Thread(this.consumer, "dispatcher");
+
         init();
     }
 
     public static synchronized ComponentManager getInstance() {
+
         if (service == null) {
             service = new ComponentManager();
             service.start();
@@ -55,10 +71,12 @@ private void start() {
                 try {
                     ((IInitManager) component).setManager(this);
                 } catch (Throwable t) {
-                    LOGGER.log(Level.SEVERE, "setting manager", t);
+                    LOGGER.log(Level.SEVERE, "failed to set manager", t);
                 }
             }
         }
+
+        this.queueThread.start();
     }
 
     private void init() {
@@ -68,47 +86,65 @@ private void init() {
         this.commands = new HashMap<>();
         Iterator<IComponent> components = this.loader.iterator();
         while (components.hasNext()) {
-            IComponent component = components.next();
-            for (String command: component.getCommands()) {
-                if (this.commands.put(command, component) != null) {
-                    throw new RuntimeException("command already registered: " + command);
-                }
-            }
 
-            boolean isSource = false;
-            if (component instanceof IEventSource) {
-                isSource = true;
-                ((IEventSource) component).registerListener(this);
-            }
-
-            boolean isListener = false;
-            if (component instanceof IEventListener) {
-                isListener = true;
-                registerListener((IEventListener) component);
+            try {
+                IComponent component = components.next();
+                LOGGER.log(Level.CONFIG, "configuring {0}", component.getName());
+
+                for (String command: component.getCommands()) {
+                    if (this.commands.containsKey(command)) {
+                        throw new RuntimeException("command already registered: " + command);
+                    }
+                    else {
+                        this.commands.put(command, component);
+                    }
+                }
+
+                boolean isSource = false;
+                if (component instanceof IEventSource) {
+                    isSource = true;
+                    ((IEventSource) component).registerListener(this);
+                }
+
+                boolean isListener = false;
+                if (component instanceof IEventListener) {
+                    isListener = true;
+                    registerListener((IEventListener) component);
+                }
+
+                LOGGER.log(Level.CONFIG, 
+                        "added commands {0} from {1} version {2} source:{3} listener:{4}", 
+                        new Object[]{
+                            Arrays.toString(component.getCommands()), 
+                            component.getName(), 
+                            getVersion(component),
+                            isSource? "Y":"N",
+                            isListener? "Y":"N"
+                        });                  
+
+            } catch (Throwable t) {
+                LOGGER.log(Level.SEVERE, "failed to load component", t);
             }
 
-            LOGGER.log(Level.CONFIG, 
-                    "added commands {0} from {1} version {2} source:{3} listener:{4}", 
-                    new Object[]{
-                        Arrays.toString(component.getCommands()), 
-                        component.getName(), 
-                        getVersion(component),
-                        isSource? "Y":"N",
-                        isListener? "Y":"N"
-                    });     
+
         }
+
     }
 
     @Override
     public void close() {
+
+        this.consumer.finish();
+        this.queueThread.interrupt();
+
         Iterator<IComponent> components = this.loader.iterator();
         while (components.hasNext()) {
             IComponent component = components.next();
             if (component instanceof AutoCloseable) {
                 try {
                     ((AutoCloseable) component).close();
                 } catch (Throwable t) {
-                    LOGGER.log(Level.SEVERE, "closing component", t);
+                    LOGGER.log(Level.SEVERE, "failed to close component", t);
                 }
             }
         }
@@ -145,11 +181,14 @@ public JSONObject getComponentsInfo() {
     }
 
     @Override
-    public synchronized void handleEvent(JSONObject event) {
-        // events from components to a (network server?)
-        LOGGER.log(Level.INFO, "event:{0}", event.toString(4));  
-        for (IEventListener listener: this.listeners) {
-            listener.handleEvent(event);
+    public void handleEvent(JSONObject event) {
+
+        if (event.has("source") && event.has("type")) {
+            LOGGER.log(Level.INFO, "queue event:{0}", event.toString(4));  
+            this.queue.add(event);            
+        }
+        else {
+            LOGGER.log(Level.SEVERE, "wrong event format:{0}", event.toString(4));  
         }
     }
 
@@ -166,17 +205,6 @@ public void unregisterListener(IEventListener listener) {
         LOGGER.log(Level.CONFIG, "unregistered {0}", listener.getClass().getName());
     }
 
-    /*
-    @Override
-    public String getName() {
-        return this.getClass().getSimpleName();
-    }
-
-    @Override
-    public String[] getCommands() {
-        return this.commands.keySet().toArray(new String[this.commands.size()]);
-    }*/
-
     @Override
     public JSONObject execute(JSONObject command) {
 
@@ -220,10 +248,55 @@ private static void configureLogging() {
             }
         }
     }
+
+    private class QueueConsumer implements Runnable {
+
+        private boolean finish;
+
+        public QueueConsumer() {
+            this.finish = false;
+        }
+
+        public void finish() {
+            this.finish = true;
+        }
+
+        @Override
+        public void run() {
+
+            while (!this.finish) {
+
+                try {
+                    JSONObject event = ComponentManager.this.queue.poll(
+                            POLL_TIMEOUT, TimeUnit.MILLISECONDS);
+
+                    if (event == null) {
+                        continue;
+                    }
+
+                    LOGGER.log(Level.INFO, "dispatch event:{0}", event.toString(4));  
+                    for (IEventListener listener: ComponentManager.this.listeners) {
+                        try {
+                            listener.handleEvent(event);
+                        } catch (Exception ex) {
+                            LOGGER.log(Level.SEVERE, 
+                                    "dispatching event to " + listener.getClass().getName(), ex);
+                        }
+                    }
+
+                } catch (InterruptedException ex) {
+                    LOGGER.log(Level.CONFIG, "consumer interrupted");
+                }
+            }
+
+            LOGGER.log(Level.CONFIG, "finished queue consumer");
+        }
+
+    }
 
     public static void main(String[] args) {
         // for testing purposes
-        getInstance();
+        getInstance().close();
     }
 
 }