NumberSink.java

/*
 * Created on Mar 5, 2004
 */
package org.nees.rbnb;

import java.io.IOException;
import java.util.Date;
import java.util.TimeZone;
import java.text.SimpleDateFormat;

import com.rbnb.sapi.*;
import COM.Creare.Utility.ArgHandler; //for argument parsing
import COM.Creare.Utility.RBNBProcess; //alternative to System.exit, so

                                       //don't bring down servlet engine

public class NumberSink {
    
    private static final String SERVER_NAME = "localhost";
    private static final String SERVER_PORT = "3333";
    private static final String SINK_NAME = "GetNumber";
    private static final String SOURCE_NAME = "RandomWalk";
    private static final String CHANNEL_NAME = "RandomWalkData";
    
    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("MMM d, yyyy h:mm aa");
    private static final TimeZone TZ = TimeZone.getTimeZone("GMT");

    static
    {
        DATE_FORMAT.setTimeZone(TZ);
    }

    private String serverName = SERVER_NAME;
    private String serverPort = SERVER_PORT;
    private String server = serverName + ":" + serverPort;
    private String sinkName = SINK_NAME;
    private String sourceName = SOURCE_NAME;
    private String channelName = CHANNEL_NAME;
    private String requestPath = sourceName + "/" + channelName;
     
    Sink sink = null;
    ChannelMap sMap;
    int index;
    boolean connected = false;
    
    Thread stringDataThread;
    boolean runit = false;
    
    public static void main(String[] args) {
        NumberSink s = new NumberSink(args);
        s.exec();
        s.startThread();
    }
    
    private void printUsage() {
        System.out.println("NumberSink: usage is...");      
        System.out.println("NumberSink ");
        System.out.println("[-s Server Hostname *" + SERVER_NAME + "] ");
        System.out.println("[-o Server Port Number *" + SERVER_PORT + "] ");
        System.out.println("[-k Sink Name *" + SINK_NAME + " ]");
        System.out.println("[-n Source Name *" + SOURCE_NAME + "] ");
        System.out.println("[-c Source Channel Name *" + CHANNEL_NAME + "] ");
    }
    
    public NumberSink(String[] args) {
        //parse args
        try {
            ArgHandler ah=new ArgHandler(args);
            if (ah.checkFlag('h')) {
                printUsage();
                RBNBProcess.exit(0);                
            }
            if (ah.checkFlag('s')) {
                String a=ah.getOption('s');
                if (a!=null) serverName=a;
            }
            if (ah.checkFlag('p')) {
                String a=ah.getOption('p');
                if (a!=null) serverPort=a;
            }
            if (ah.checkFlag('n')) {
                String a=ah.getOption('n');
                if (a!=null) sourceName=a;
            }
            if (ah.checkFlag('c')) {
                String a=ah.getOption('c');
                if (a!=null) channelName=a;
            }
            if (ah.checkFlag('k')) {
                String a=ah.getOption('k');
                if (a!=null) sinkName=a;
            }
        } catch (Exception e) {
            System.err.println("NumberSink argument exception "+e.getMessage());
            e.printStackTrace();
            RBNBProcess.exit(0);
        }

        requestPath = sourceName + "/" + channelName;
        server = serverName + ":" + serverPort;
        
        System.out.println("Starting NumberSink on " + server + " as " + sinkName);
        System.out.println("  Requesting " + requestPath);
        System.out.println("  Use NumberSink -h to see optional parameters");
    }
    
    public void exec()
    {
        try {
            // Create a sink and connect:
            sink=new Sink();
            sink.OpenRBNBConnection(server,sinkName);
            sMap = new ChannelMap();
            index = sMap.Add(requestPath);
            sink.Subscribe(sMap,"newest");
            connected = true;
            System.out.println("NumberSink: Connection made to sever = "
                + server + " as " + sinkName 
                + " requesting " + requestPath + ".");
        } catch (SAPIException se) { se.printStackTrace(); }
    }
    
    public void startThread()
    {
        
        if (!connected) return;
        
        // Use this inner class to hide the public run method
        Runnable r = new Runnable() {
            public void run() {
              runWork();
            }
        };
        runit = true;
        stringDataThread = new Thread(r, "StringData");
        stringDataThread.start();
        System.out.println("NumberSink: Started thread.");
    }

    public void stopThread()
    {
        runit = false;
        stringDataThread.interrupt();
        System.out.println("NumberSink: Stopped thread.");
    }
    
    private void runWork ()
    {
        try {
            while(isRunning())
            {
                ChannelMap m = sink.Fetch(-1);
                if (m == null)
                {
                    System.out.println("Data fetch failed.");
                    continue;
                }
                double timeStamp = m.GetTimeStart(index);
                double[] data = m.GetDataAsFloat64(index);
                long unixTime = (long)(timeStamp * 1000.0); // convert sec to millisec
                String time = DATE_FORMAT.format(new Date(unixTime));
                System.out.println("Data Received (for " + time + " GMT): ");
                for (int i = 0; i < data.length; i++)
                {
                    System.out.println("  " + i + ": " + data[i]);
                }
            }
        } catch (SAPIException se) {
            se.printStackTrace(); 
        }
        stringDataThread = null;
    }
    
    public boolean isRunning()
    {
        return (connected && runit);
    }


}

---