Developing StreamBase Monitor Applications

Using the com.streambase.sb.monitor API, you can get the same statistics that are available in display-only mode from one of the StreamBase monitoring utilities described in Monitoring and Profiling.

The key difference is that, in your client, you can take some action based on the current performance numbers. This topic describes the classes and methods you can use to capture specific types of performance data.

Consider first an example monitoring client. The following shows a complete SimpleMonitor.java example that uses the com.streambase.sb.monitor API. Notice that this example assumes that StreamBase Server is started on port 10000.

import com.streambase.sb.*;
import com.streambase.sb.monitor.*;

class SimpleMonitor {
  public static void main(String[] argv) {

    class MyMonListener implements MonitorListener {
      public void snapshotReceived(Snapshot monSnap) {
              monSnap.prettyPrint(true,true,true,false);
      }
    }

    System.out.println("Start Simplest Monitor...");
    MyMonListener listener = new MyMonListener();
      try {
        final String sburl = "sb://localhost:10000";
        StreamBaseMonitor sbmon = new StreamBaseMonitor(sburl);
        sbmon.addMonitorListener(listener);
        sbmon.run();
        } 
      catch (StreamBaseException e) {
        System.out.println("Exception in StreamBaseMonitor run: "+ e);
      }
      System.out.println("End Simplest Monitor...");
  }
}

A running EventFlow engine produces monitoring information — everything you see with the StreamBase Monitor utility. A snapshot is generated every cycle. The cycle's period is one second. The set of tuples is punctuated by a marker tuple (with "" for the type of data). Most of the numbers are absolute counts of entities, such as the count of tuples enqueued or dequeued on an input queue.

The general idea is that a Snapshot object stores a snapshot of information from a StreamBase application. Snapshot objects are created by the StreamBaseMonitor class, and consumed by MonitorListeners. The conceptual view is illustrated here:

Now let's take a look at the classes and methods, and specifically how to get different types of performance data.

...
public static void main(String[] argv) {

class MyMonListener implements MonitorListener {
  public void snapshotReceived(Snapshot monSnap) {
    // process the monitor snapshot
    System.out.println("----> Snapshot: "+ monSnap.toString());
    SystemInfo sysInfo = monSnap.getSystemInfo();
    for (Iterator i=monSnap.operators(); i.hasNext(); ) {
      OperatorInfo opInfo = (OperatorInfo)i.next();
      System.out.println("  "+ opInfo.getName() + ": "+ opInfo.toString());
    }
    for (Iterator i=monSnap.threads(); i.hasNext(); ) {
      ThreadInfo thdInfo = (ThreadInfo)i.next();
      System.out.println("  "+ thdInfo.getName() + ": "+ thdInfo.toString());
    }
  }
}

System.out.println("Start Monitor...");
MyMonCB listener = new MyMonListener();
 try {
   StreamBaseMonitor sbmon = new StreamBaseMonitor("sb://mymachine:10000");
   sbmon.addMonitorListener(listener);
   sbmon.run();
  } 
  catch (StreamBaseException e) {
    System.out.println("Exception StreamBaseMonitor run: "+ e);
  }
}

Before you can use the client, of course you need a StreamBase application, such as the simple one illustrated here:

This StreamBase application, which we named pass.sbapp, contains an input stream with this schema:

The input stream is connected to a Map operator named Map1 that adds to these fields:

You can compile your Java program with a command such as the following examples:

As with all clients that interact with a StreamBase application, a StreamBase Server instance must be running.

Once the server has been started, you can run the client by entering commands such as the following. For best performance, we also recommend that you use the -server option with the java command, to select the server VM: