Creating Custom Java Embedded Adapters

StreamBase currently supports the property types data types shown in the table in Adding Adapter Properties. Most are simple property types that StreamBase derives automatically by looking at the signatures of the getter and setter methods in the Parameterizable class. These can be specified in the BeanInfo class with SBPropertyDescriptor objects, as in the example above.

Enum properties are string properties that can only take on a specified set of values. ResourceFile properties are similar to Enum properties, but their value must be the name of a resource. In StreamBase Studio, this is enforced by displaying a drop-down list of files available from your project's folder.

Because both Enum and ResourceFile properties are implemented as Strings, their getter and setter methods must return and expect objects of type String, respectively. Thus, these property types cannot be automatically derived by StreamBase. To use these properties, an adapter must have an accompanying BeanInfo class that returns the SBPropertyDescriptor subclasses EnumPropertyDescriptor or ResourceFilePropertyDescriptor in its list of PropertyDescriptors.

The BeanInfo class is defined in the java.beans.BeanInfo API documentation. It can be used by the Parameterizable object to:

If a BeanInfo class is present, only the properties explicitly declared in this class are exposed by StreamBase.

The easiest way to make a BeanInfo class is to extend java.beans.SimpleBeanInfo in a class whose name consists of the name of the Parameterizable class, with BeanInfo appended, residing in the same package as the Parameterizable class. For example, if the Parameterizable class is called MyAdapter, the BeanInfo class should be called MyAdapterBeanInfo.

The following example BeanInfo class makes the Clever property optional:

public class ExampleBeanInfo extends SimpleBeanInfo {
  public PropertyDescriptor [] getPropertyDescriptors() {
    try {
        SBPropertyDescriptor [] p = {
            new SBPropertyDescriptor("StartCount", Example.class),
            new SBPropertyDescriptor("Clever", Example.class).optional(),
            new SBPropertyDescriptor("Name", Example.class)
        };

        return p;
    }
    catch(IntrospectionException e) {
        System.out.println("Exception: " + e);
        e.printStackTrace();
    }
    return null;
  }
}

Note that the optional() method shown in this example simply marks the descriptor as optional.

Input adapters often need to perform operations asynchronously with the StreamBase application they are running in. For example, an input adapter may need to listen for information on a socket, or read input from a file, without interrupting the rest of the application. Though output adapters are called upon by StreamBase whenever a tuple arrives on an input port, there may be other asynchronous tasks an output adapter may wish to perform, such as periodically flushing a buffer.

To facilitate such requirements, the adapter API provides a mechanism for creating threads that are managed along with the rest of the StreamBase application. During the init method, an adapter may call registerRunnable(runnable) to register Runnable objects that will be managed in separate threads by StreamBase. When the application starts running, StreamBase will start all registered Runnables in new threads.

StreamBase-managed threads have the following general structure:

...
  public void run() {
    // Perform thread startup tasks
    while (shouldRun()) {
      // Perform tasks such as calling sendOutput
    }
    // Perform thread shutdown tasks
  }
...

The call to shouldRun ties the thread into StreamBase's thread management in the following ways:

Embedded adapters start, pause, resume and shutdown along with the StreamBase application that contains them. It is also possible to suspend and resume embedded adapters independently of their StreamBase application.

This is accomplished with the same commands that are used to suspend and resume a StreamBase application as a whole. The command sbadmin suspend can be used to suspend an application; similarly, sbadmin resume is used to resume a StreamBase application.

To suspend or resume individual adapters, you can append a list of embedded adapters to the sbadmin suspend or sbadmin resume command. If one or more strings are appended to sbadmin suspend, then the StreamBase application as a whole will not be suspended, rather the individual embedded adapters named by the appended strings will be suspended.

Note that an individual embedded adapter can be suspended or resumed only if the StreamBase application itself is running. An individual embedded adapter can be running only if the application that contains it is also running. Therefore, it is not meaningful to suspend or resume an adapter that is not currently running if the application that contains it is not running.

The sbc status command returns status information about the StreamBase Server.

The sbc status --operators command returns the status of any embedded adapters contained by the server's application. Note that sbc status --operators and sbc status are disjoint commands: sbc status returns information about the server only, not about embedded adapters contained in the server. Similarly, sbc status --operators returns information about contained embedded adapters only, and not any information about the server itself.

In this context, the status of an embedded adapter consists only of its current state. For example, if an embedded adapter has been started and is currently running, its state will be STARTED. If an embedded adapter has yet to be started, its state will be NONE. If an embedded adapter has been suspended, its state will be either SUSPENDED_AND_DROPPING_TUPLES or SUSPENDED_AND_PROCESSING_TUPLES. Lastly, if an adapter has been shut down its state will be SHUTDOWN.

By default, an embedded adapter starts along with the StreamBase application that contains it. In Studio, in the General tab for embedded adapters, there is a check box labeled Start with application. By default, this box is selected, meaning that the embedded adapter will start with the application. Clearing this check box causes the adapter to be left in the NONE state when the application starts.

A Java operator that does not start with its application will stay in the NONE state until it is explicitly started with the sbadmin resume adapterName command. Such an adapter will not start even if the application as a whole is resumed. So for example, the application as a whole may suspend and then resume; this will have no effect on an adapter that has not started with its application.

If an embedded adapter is suspended separately from the application that contains it, tuples might still arrive at the suspended adapter. You can configure the embedded adapter to handle these tuples in two different ways:

These two possibilities are represented by static Strings on the class com.streambase.sb.operator.Operator, SUSPENDED_AND_DROPPING_TUPLES and SUSPENDED_AND_PROCESSING_TUPLES, respectively. An embedded adapter is configured to either drop or process tuples by calling the method setSuspendBehavior on its instance. setSuspendBehavior takes an int argument, the value of which must be either SUSPENDED_AND_DROPPING_TUPLES or SUSPENDED_AND_PROCESSING_TUPLES.