NeesControl.java

/*
 * Created on Apr 29, 2004
 */
package org.nees.rbnb;

import java.io.InputStreamReader;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.File;
import java.io.FileInputStream;

import java.util.Properties;

import java.util.Iterator;
import java.util.List;
import java.util.Vector;
import java.util.Hashtable;
import java.util.Enumeration;
import java.util.Date;
import java.util.TimeZone;
import java.text.SimpleDateFormat;

import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
import org.w3c.dom.Node;

import com.rbnb.sapi.*;

//import COM.Creare.Utility.ArgHandler; //for argument parsing
import com.rbnb.utility.ArgHandler; //for argument parsing

public class NeesControl
{

    private static final String DEFAULT_PARAMETER_FILE_PATH = "./NeesControl.properties";
    private static final String SERVER_NAME = "localhost";
    private static final String SERVER_PORT = "3333";
    private static final String SOURCE_NAME = "neesPop";
    private static final String SOURCE_CHANNEL_NAME = "commandFeedback";
    private static final String SINK_NAME = "neesPop_CommandInput";
    private static final String SINK_SOURCE_PATH = "neesPopClient/command";
    private static final long CONNECT_IDLE_TIME = 1000*60*10; // ten minutes in milliseconds
    private static final long NON_CONNECT_IDLE_TIME = 1000*30; // 30 seconds
    private static final long LONG_THREAD_IDLE_TIME = 1000*60*60; // on hour; interupted

    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("MMM d, yyyy h:mm aa");
    private static final SimpleDateFormat LOG_FORMAT = new SimpleDateFormat("yyyy.MM.dd-HH:mm:ss z");
    private static final TimeZone TZ = TimeZone.getTimeZone("GMT");

    static
    {
        DATE_FORMAT.setTimeZone(TZ);
    }
        
    private String server = SERVER_NAME + ":" + SERVER_PORT;

    private Thread backgroundThread;
    private boolean backgroundRunning = false;
    
    private Source source = null;
    private String sourceName = SOURCE_NAME;
    private String sourceChannelName = SOURCE_CHANNEL_NAME;
    private int sourceChannelIndex;
    private ChannelMap sourceMap;
    private Thread sourceCommandThread;
    private boolean sourceConnected = false;
    private boolean sourceCommandRunning = false;

    private Sink sink = null;
    private String sinkName = SINK_NAME;
    private ChannelMap sinkMap;
    private int sinkIndex;
    private String sinkRequestPath = SINK_SOURCE_PATH;
    private Thread sinkCommandThread;
    private boolean sinkConnected = false;
    private boolean sinkCommandRunning = false;

    private Sink metadataSink = null;
    private ChannelMap metadataSinkMap;
    private Thread metadataSinkThread;
    private boolean metadataSinkConnected = false;
    private boolean metadataSinkRunning = false;
    
    private Thread commandThread;
    private boolean commandRunning = false;

    private MetaDataPropertyList pList = new MetaDataPropertyList();
    private Vector commands = new Vector();
    private Vector replies = new Vector();
    private NeesControlMetadata metadata = new NeesControlMetadata();
    
    private String propertiesFilePath = DEFAULT_PARAMETER_FILE_PATH;
    private Properties prop = new Properties();

    private RBNBBase[] widget =
        { (new WalkerSource()), (new PictureSource()), (new AxisSource()),
        (new NumberSubSample()), (new DataVideoGather()), (new DaqToRbnb()),
        (new JPEGThumbnailer()) };
    private String[] widgetType = {
        NeesControlMetadata.DATA_TYPE,
        NeesControlMetadata.IMAGE_TYPE,
        NeesControlMetadata.IMAGE_TYPE,
        NeesControlMetadata.DATA_TYPE,
        NeesControlMetadata.MOVIE_TYPE,
        NeesControlMetadata.DATA_TYPE,
        NeesControlMetadata.IMAGE_TYPE
        };

    public static void main(String[] args) {
        NeesControl n = new NeesControl();
        n.exec(args);
    }
        
    private void exec(String[] args)
    {
        if (!setArgs(args)) return;

        // set up template list
        for (int i = 0; i < widget.length; i++)
        {
            addTemplate(widget[i], widgetType[i]);
        }
        
        // Use this inner class to hide the public run method
        Runnable r = new Runnable() {
            public void run() {
              runBackground();
            }
        };
        backgroundRunning = true;
        backgroundThread = new Thread(r, "Background");
        log("NeesControl: Starting Background Thread.");
        backgroundThread.start();
    }
    
    private void addTemplate(RBNBBase base, String type) {
        // put template into metaData and on local list
        String name = base.getTemplateName();
        metadata.addToTemplateList(name,type);
        // get paprameters from base and add to template
        metadata.insertParametersToTempateFromBase(name, base);
    }

    protected void runBackground()
    {
        while(true)
        {
            try{
                log("Starting... ");
        
                // attempt to set up the initial conditions and then test if they worked
                if (connectSource()) startSourceThread();
                if (connectSink()) startSinkThread();
        
                // this runs until the program is killed from the outside
                // all clean up is done by closing the sockets and timing out
                while (true)
                {
                    log("Background Thread Loop");
                    
                    if (!commandIsRunning()) startCommandThread();
                    
                    if (sinkIsRunning() && sourceIsRunning())
                    {
                        // everything is a-ok... go idle for a long time
                        try {
                            log("Background idle");
                            Thread.sleep(CONNECT_IDLE_TIME);
                        } catch (InterruptedException e) {
                            // log interupt and retry
                            log("Background thread interupted (ok condition)...");
                        }
                    }
                    else
                    {
                        if (!sourceConnected) connectSource();
                        if (!sourceIsRunning()) 
                        {
                            stopSourceThread();
                            startSourceThread();
                        } 
                        if (!sinkConnected) connectSink();
                        if (!sinkIsRunning())
                        {
                            stopSinkThread();
                            startSinkThread();
                        }
                        if (!sinkIsRunning() || !sourceIsRunning())
                        {
                            // still problems ... go idle for a short time and retry
                            log("problems with connections...");
                            try {
                                Thread.sleep(NON_CONNECT_IDLE_TIME);
                            } catch (InterruptedException e) {
                                // log interupt and retry
                                log("Background thread interupted (problem condition)...");
                            }               
                            if (!sinkIsRunning()) log ("  command sink is not running");
                            if (!sourceIsRunning()) log("  response source is not running");
                            log("...retyring");
                        }
                    }
                } // while (true)
            } catch (Throwable t)
            {
                log("Error in background thread = " + t);
                t.printStackTrace();
            } // try
            log("Restarting....");
        } // (outer) while (true)
    } // runBackground

    private void printUsage() {
        System.out.println("NeesControl: usage is...");     
        System.out.println("NeesControl ");
        System.out.println("[-f (PorpertiesFilePat) file path run time properties *"
            + propertiesFilePath +"]");
    } // printUsage
    
    private boolean setArgs(String[] args)
    {
        try {
            ArgHandler ah=new ArgHandler(args);
            if (ah.checkFlag('h')) {
                printUsage();
                return false;
            }
            if (ah.checkFlag('f')) {
                String a=ah.getOption('f');
                if (a!=null) propertiesFilePath=a;
            }
        } catch (Exception e) {
            log("Failed to process command line args: " + e);
            log("(use NeesControl -h to see command line options)");
            return false;
        }
        
        return 
            processPropertiesFile(propertiesFilePath)
            && setPropertiesFromArgs(args)
            && setVarablesFromProperties();
    } // setArgs
    
    private boolean processPropertiesFile(String propertiesFilePath)
    {
        if (propertiesFilePath == null) return true;
        if (propertiesFilePath.length() == 0) return true;
        File f = new File(propertiesFilePath);
        if (!f.exists())
        {
            log("Properties file does not exist: " + propertiesFilePath);
            return true;
        }
        
        try
        {
            prop.load(new FileInputStream(f));
            log("Sucessfully loaded properties file: " + propertiesFilePath);
        }
        catch (IOException e)
        {
            log("Can not process properties file, " + propertiesFilePath
                + ", exception = " + e);
            return false;
        }
        return true;
    } // processPropertiesFile

    private boolean setPropertiesFromArgs(String[] args) {
        // TODO Auto-generated method stub
        return true;
    } // setPropertiesFromArgs

    private boolean setVarablesFromProperties() {
        // TODO Auto-generated method stub
        return true;
    } // setVarablesFromProperties

    protected boolean connectSource()
    {
        try {
            // Create a source and connect:
            source=new Source();
            source.OpenRBNBConnection(server,sourceName);
            sourceMap = new ChannelMap();
            sourceConnected = true;
            log("NeesControl: Source connection made to server = "
                + server + " as " + sourceName + ".");
        }
        catch (SAPIException se)
        {
            log("NeesControl: Cannot connect source to "
            + server + " as " + sourceName + ": " + se);
        }
        return sourceConnected;
    } // connectSource
    
    protected boolean connectSink()
    {
        try {
            // Create a sink and connect:
            sink=new Sink();
            sink.OpenRBNBConnection(server,sinkName);
            sinkMap = new ChannelMap();
            sinkIndex = sinkMap.Add(sinkRequestPath);
            sink.Subscribe(sinkMap,"newest");
            sinkConnected = true;
            log("NeesControl: Sink connection made to server = "
                + server + " as " + sinkName + ".");
        }
        catch (SAPIException se)
        {
            log("NeesControl: Cannot connect sink to "
            + server + " as " + sinkName + ": " + se);
        }
        return sinkConnected;
    } // connectSink
        
    protected void startSourceThread()
    {
        
        if (!sourceConnected) return;
        
        // Use this inner class to hide the public run method
        Runnable r = new Runnable() {
            public void run() {
              runSourceWork();
            }
        };
        sourceCommandRunning = true;
        sourceCommandThread = new Thread(r, "Command");
        sourceCommandThread.start();
        log("NeesControl: Started Command Respond thread.");
    }

    protected void stopSourceThread()
    {
        sourceCommandRunning = false;
        if (sourceCommandThread != null)
            sourceCommandThread.interrupt();
        log("NeesControl: Stopped Command Respond thread.");
    }
    
    protected void runSourceWork ()
    {
        try {
            while(sourceIsRunning())
            {
                log("Source thread loop");
                String replyString = null;
                synchronized (replies)
                {
                    if (replies.size() > 0)
                    {
                        replyString = (String)replies.firstElement();
                        replies.removeElementAt(0);
                    }
                }
                
                if (replyString != null)
                {
                    //log("Reply sent is: " + replyString);
                    log ("Reply send");
                    // send command
                    sourceMap.Clear();
                    sourceChannelIndex = sourceMap.Add(sourceChannelName);
                    sourceMap.PutTimeAuto("timeofday");
                    sourceMap.PutDataAsString(sourceChannelIndex,replyString);
                    source.Flush(sourceMap);
                }
                try {
                    Thread.sleep(LONG_THREAD_IDLE_TIME);
                } catch (InterruptedException ignore) {}
            }
        } catch (SAPIException se) {
            log("Source Connection failed " + se);
        }
        source.CloseRBNBConnection();
        sourceConnected = false;
        sourceCommandThread = null;
        backgroundThread.interrupt();
    }
    
    protected boolean sourceIsRunning()
    {
        return ( (sourceCommandThread != null) &&
        sourceConnected && sourceCommandRunning);
    }

    protected void startSinkThread()
    {
        
        if (!sinkConnected) return;
        
        // Use this inner class to hide the public run method
        Runnable r = new Runnable() {
            public void run() {
              runSinkWork();
            }
        };
        sinkCommandRunning = true;
        sinkCommandThread = new Thread(r, "Listen");
        sinkCommandThread.start();
        log("NeesControl: Started Command Listen thread.");
    }

    protected void stopSinkThread()
    {
        sinkCommandRunning = false;
        if (sinkCommandThread != null)
            sinkCommandThread.interrupt();
        log("NeesControl: Stopped Command Listen thread.");
    }
    
    protected void runSinkWork()
    {
        try {
            while(sinkIsRunning())
            {
                log("Sink thread loop");
                ChannelMap m = sink.Fetch(-1);
                if (m == null)
                {
                    log("Data fetch failed.");
                    continue;
                }
                String[] st = m.GetDataAsString(sinkIndex);
                log("MCP: Command(s) Received(count = "+ st.length + "): ");
                for (int i = 0; i < st.length; i++)
                {
                    log(st[i]);
                    synchronized (commands)
                    {
                        commands.add(st[i]);
                    }
                }
                log("end command list - starting command thread");
                if ((commandThread != null)
                        && commandIsRunning())
                    commandThread.interrupt();
            }
        } catch (SAPIException se) {
            log("Sink Connection failed " + se);
        }
        sink.CloseRBNBConnection();
        sinkConnected = false;
        sinkCommandThread = null;
        backgroundThread.interrupt();
    }
    
    protected boolean sinkIsRunning()
    {
        return ( (sinkCommandThread != null) &&
        sinkConnected && sinkCommandRunning);
    }

    protected void startCommandThread()
    {
        
        // Use this inner class to hide the public run method
        Runnable r = new Runnable() {
            public void run() {
              runCommandWork();
            }
        };
        commandRunning = true;
        commandThread = new Thread(r, "Command");
        commandThread.start();
        log("NeesControl: Started Command Processing thread.");
    }

    protected void stopCommandThread()
    {
        commandRunning = false;
        if (commandThread != null)
            commandThread.interrupt();
        log("NeesControl: Stopped Command Processing thread.");
    }
    
    protected void runCommandWork()
    {
        String command = "";

        while(commandIsRunning())
        {
            try {
                Thread.sleep(LONG_THREAD_IDLE_TIME);
            } catch (InterruptedException ignore) {}
            log("Command thread loop");
            // decant commands and then sleep
            while (commands.size() > 0)
            {
                command = "unrecognized";
                try {
                    synchronized (commands)
                    {
                        command = (String)commands.elementAt(0);
                        commands.removeElementAt(0);
                    }
                    processCommand(command);    
                }
                catch (Throwable t)
                {
                    log("Problems processing command (" + command + ") = " + t);
                }
            }
        }
        commandThread = null;
    }
    
    protected boolean commandIsRunning()
    {
        return commandRunning;
    }

    protected void processCommand(String command)
    {
        Document doc = null;
        try{
            doc = Xml.readDocumentFromString(command);
        } catch (Throwable ignore) {}
        if (doc == null)
            log("Unrecognized Command (not XML): " + command);
        else
        {
            Element root = doc.getDocumentElement();
            if (root.getTagName().equals(
                    NeesControlMetadata.XML_TAG_FOR_COMMAND))
            {
                try
                {       
                    processCommand(doc);            
                } catch (Throwable t)
                {
                    log("Message parse failed on command: " + t);
                }
            }
            else
                log("Not Command XML? " + root.getTagName());
        }
    }
    
    private void processCommand(Document doc)
    {
        String command = null;
        String template = null;
        String instance = null;
        RBNBBaseParameterHolder parameter = null;

        boolean stateChanged = false;
        
        Element root = doc.getDocumentElement();
        NodeList children = root.getChildNodes();
        final int length = children.getLength();
        for(int i = 0; i < length; i++)
        {
            Node child = children.item(i);
            if (child.getNodeType() == Node.ELEMENT_NODE)
            {
                Element element = (Element)child;
                String name = element.getTagName();
                if (name.equals(NeesControlMetadata.XML_TAG_FOR_COMMAND))
                    command = child.getChildNodes().item(0).getNodeValue();
                else
                if (name.equals(NeesControlMetadata.XML_TAG_FOR_TEMPLATE))
                    template = child.getChildNodes().item(0).getNodeValue();
                else
                if (name.equals(NeesControlMetadata.XML_TAG_FOR_INSTANCE))
                    instance = child.getChildNodes().item(0).getNodeValue();
                else
                if (name.equals(NeesControlMetadata.XML_TAG_FOR_PARAMETER))
                {
                    parameter = new RBNBBaseParameterHolder(child);
                }
            }
        }
        if (command == null)
        {
            reply(Xml.errorXml("Unprocessed command","command is null"));
        }
        else
        if (command.equals(Command.GET_METADATA))
        {
            stateChanged = true;
        }
        else if (command.equals(Command.RESTART_COMMAND))
        {
            reply(Xml.statusXml("Processed restart command"));          
            stateChanged = true;
        }
        else if (command.equals(Command.ADD_LINK))
            reply(Xml.statusXml("Unimplemented Command " + command));           
        else if (command.equals(Command.REMOVE_LINK))
            reply(Xml.statusXml("Unimplemented Command " + command));           
        else if (command.equals(Command.FOCUS_TO_LINK))
            reply(Xml.statusXml("Unimplemented Command " + command));           
        else if (command.equals(Command.CREATE_INSTANCE))
        {
            if (template == null)
            {
                reply(Xml.errorXml("Create instance command failed","no template name"));
            }
            else
            {
                createInstanceFromTempate(template);
                reply(Xml.statusXml("New instance created for " + template));           
                stateChanged = true;
            }
        }
        else if (command.equals(Command.START_INSTANCE))
        {
            stateChanged = commandStartInstance(instance);
        }
        else if (command.equals(Command.STOP_INSTANCE))
        {
            stateChanged = commandStopInstance(instance);
        }
        else if (command.equals(Command.DELETE_INSTANCE))
        {
            stateChanged = commandDeleteInstance(instance);
        }
        else if (command.equals(Command.SET_PARAMETER))
        {
            if (instance == null)
            {
                reply(Xml.errorXml("Set parameter command failed","no instance name"));
            }
            else
            if (parameter == null)
            {
                reply(Xml.errorXml("Set parameter command for "
                    + instance + "failed","no parameter name"));
            }
            else
            {
                if (commandSetParameter(instance,parameter))
                    stateChanged = true;
                else
                    reply(Xml.errorXml("Start instance command failed","can not stop instance"));
            }
        }
        else if (command.equals(Command.RESET_PARAMETER))
        {
            if (instance == null)
            {
                reply(Xml.errorXml("Reset parameter command failed","no instance name"));
            }
            else
            if (parameter == null)
            {
                reply(Xml.errorXml("Reset parameter command for "
                    + instance + "failed","no parameter name"));
            }
            else
            {
                if (commandResetParameter(instance,parameter))
                    stateChanged = true;
                else
                    reply(Xml.errorXml("Start instance command failed","can not stop instance"));
            }
        }
        else
        {
            reply(Xml.errorXml("Unprocessed command",command));
        }
        if (stateChanged) postMetadata();
    }

    /* (non-javadoc)
     * Note: assumes that instance is set to the name of the instance
     * and that the instance is an actual instance and not just a empty
     * instacne description
     */
    private boolean commandStartInstance(String instance) {
        boolean started = false;
        try
        {
            if (instance == null)
            {
                reply(Xml.errorXml("Problem Starting instance", "Null String for instance name."));
                return false;
            }
            InstanceDescription i = metadata.getInstance(instance);
            if (i == null) 
            {
                reply(Xml.errorXml("Problem Starting " + instance, "Null description."));
                return false;
            }
            started = i.startInstance();
            if (!started)
                reply(Xml.errorXml("Problem Starting " + instance, "Start failed."));
        }
        catch (Throwable t)
        {
            reply(Xml.errorXml("Exception when starting " + instance, t.toString()));
            started = false;
        }
        if (started)
            reply(Xml.statusXml("Instance " + instance + " started."));         

        return started;
    }

    /* (non-javadoc)
     * Note: assumes that instance is set to the name of the instance
     * and that the instance is an actual instance and not just a empty
     * instacne description
     */
    private boolean commandStopInstance(String instance)
    {
        boolean stopped = false;
        try
        {
            if (instance == null)
            {
                reply(Xml.errorXml("Problem Stopping instance", "Null String for instance name."));
                return false;
            }
            InstanceDescription i = metadata.getInstance(instance);
            if (i == null) 
            {
                reply(Xml.errorXml("Problem Stopping " + instance, "Null description."));
                return false;
            }
            stopped = i.stopInstance();
            if (!stopped)
                reply(Xml.errorXml("Problem Stopping " + instance, "Start failed."));
        }
        catch (Throwable t)
        {
            reply(Xml.errorXml("Problem stopping " + instance, t.toString()));
            stopped = false;
        }
        if (stopped)
            reply(Xml.statusXml("Instance " + instance + " stopped."));         

        return stopped;
    }

    /* (non-javadoc)
     * Note: assumes that instance is set to the name of the instance
     * and that the instance is an actual instance and not just a empty
     * instacne description
     */
    private boolean commandDeleteInstance(String instance)
    {
        boolean deleted = false;
        try
        {
            if (instance == null)
            {
                reply(Xml.errorXml("Problem Deleting instance", "Null String for instance name."));
                return false;
            }
            InstanceDescription i = metadata.getInstance(instance);
            if (i == null) 
            {
                reply(Xml.errorXml("Problem Deleting " + instance, "Null description."));
                return false;
            }
            if (i.isRunning())
            {
                reply(Xml.errorXml("Problem Deleting " + instance, "Instance is running; stop first."));
                return false;               
            }

            deleted = metadata.deleteInstance(instance); // may fail to actually delete!
        }
        catch (Throwable t)
        {
            reply(Xml.errorXml("Problem Deleting " + instance, t.toString()));
            deleted = false;
        }

        if (deleted)
            reply(Xml.statusXml("Instance " + instance + " deleted."));         

        return deleted;
    }

    private boolean commandSetParameter(String instance, RBNBBaseParameterHolder parameter)
    {
        boolean ret = false;
        try
        {
            if (instance == null) return false;
            InstanceDescription i = metadata.getInstance(instance);
            if (i == null) return false;
            i.setParameter(parameter.getName(),parameter.getValue());           
            ret = true;
        }
        catch (Throwable ignore) {}
        return ret;
    }

    private boolean commandResetParameter(String instance, RBNBBaseParameterHolder parameter)
    {
        boolean ret = false;
        try
        {
            if (instance == null) return false;
            InstanceDescription i = metadata.getInstance(instance);
            if (i == null) return false;
            i.setParameter(parameter.getName(),parameter.getDefaultValue());            
            ret = true;
        }
        catch (Throwable ignore) {}
        return ret;
    }

    private void createInstanceFromTempate(String templateName) {
        String name = templateName + "_01";
        int index = 2;
        Hashtable h = metadata.getInstancesHashtable();
        while (h.get(name) != null)
        {
            name = templateName 
                + ((index < 9)?("_0" + index):("_" + index));
            index++;
        }
        if (templateName.equals("WalkerSource"))
        {
            WalkerSource w = new WalkerSource();
            metadata.addToInstanceList(name,templateName,(RBNBBase)w);
        }
        else if (templateName.equals("PictureSource"))
        {
            PictureSource p = new PictureSource();
            metadata.addToInstanceList(name,templateName,(RBNBBase)p);
        }
        else if (templateName.equals("NumberSubSample"))
        {
            NumberSubSample n = new NumberSubSample();
            metadata.addToInstanceList(name,templateName,(RBNBBase)n);
        }
        else if (templateName.equals("DataVideoGather"))
        {
            DataVideoGather n = new DataVideoGather();
            metadata.addToInstanceList(name,templateName,(RBNBBase)n);
        }
        else if (templateName.equals("DaqToRbnb"))
        {
            DaqToRbnb n = new DaqToRbnb();
            metadata.addToInstanceList(name,templateName,(RBNBBase)n);
        }
        else if (templateName.equals("AxisSource"))
        {
            AxisSource n = new AxisSource();
            metadata.addToInstanceList(name,templateName,(RBNBBase)n);
        }
        else if (templateName.equals("JPEGThumbnailer"))
        {
            JPEGThumbnailer n = new JPEGThumbnailer();
            metadata.addToInstanceList(name,templateName,(RBNBBase)n);
        }
        else
        {
            log("Creating instance for unknow template type");
            metadata.addToInstanceList(name,templateName);
        }
    }

    protected void reply(String r)
    {
        synchronized (replies)
        {
            replies.add(r);
        }
        if ((sourceCommandThread != null) && sourceIsRunning())
        sourceCommandThread.interrupt();
    }
    
    protected void postMetadata()
    {
        reply(metadata.toXmlString());
    }
        
    protected void log(String message)
    {
        String time = LOG_FORMAT.format(new Date());
        System.out.println("Log (" + time + "): " + message);
    }
    
} // NeesControl
/*
 * $Header: /disks/cvs/neesgrid/turbine/src/org/nees/rbnb/NeesControl.java,v 1.15 2004/07/22 18:50:27 hubbard Exp $
 */

---