StringSink.java

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

import java.io.IOException;

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 StringSink {
    
    private static final String SERVER = "localhost:3333";
    private static final String SINK_NAME = "GetSting";
    private static final String SOURCE_NAME = "Command";
    private static final String CHANNEL_NAME = "CommandData";
    
    private String server = SERVER;
    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) {
        StringSink w = new StringSink(args);
        w.exec();
        w.startThread();
    }
    
    private void printUsage() {
        System.out.println("StringSink: usage is...");      
        System.out.println("StringSink ");
        System.out.println("[-s server_hostname *" + SERVER + "] ");
        System.out.println("[-k Sink Name *" + SINK_NAME + " ]");
        System.out.println("[-n source_name *" + SOURCE_NAME + "] ");
        System.out.println("[-c channel_name *" + CHANNEL_NAME + "] ");
    }
    
    public StringSink(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) server=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("StringSink argument exception "+e.getMessage());
            e.printStackTrace();
            RBNBProcess.exit(0);
        }

        requestPath = sourceName + "/" + channelName;
        
        System.out.println("Starting StringSink on " + server + " as " + sinkName);
        System.out.println("  Requesting " + requestPath);
        System.out.println("  Use StringSink -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("StringSink: 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("StringSink: Started thread.");
    }

    public void stopThread()
    {
        runit = false;
        stringDataThread.interrupt();
        System.out.println("StringSink: Stopped thread.");
    }
    
    private void runWork ()
    {
        try {
            while(isRunning())
            {
                ChannelMap m = sink.Fetch(-1);
                if (m == null)
                {
                    System.out.println("Data fetch failed.");
                    continue;
                }
                String[] st = m.GetDataAsString(index);
                System.out.println("Command(s) Received: ");
                for (int i = 0; i < st.length; i++)
                {
                    System.out.println(st[i]);
                }
            }
        } catch (SAPIException se) {
            se.printStackTrace(); 
        }
        stringDataThread = null;
    }
    
    public boolean isRunning()
    {
        return (connected && runit);
    }


}

---