Implementing a sink operator using the Java Operator API

A sink operator transmits information or events to an external system, such as a dashboard, web-server, mail-server, or a database. When such a system provides a Java™ API, a sink operator to that system can be implemented in Java.

1. Design the model for transmission of information from tuples to the external system.
   1. Decide whether and how to use batching. In the example, batching is configurable with timeouts and batching is implemented at the tuple level using the facilities of TupleConsumer.
   2. Design how the information from the incoming com.ibm.streams.operator.Tuple object is to be translated into the Java objects that are supported by the external system API. For HTTP POST, each attribute of the tuple is sent as a name, value pair where name is the attribute's name, and value is a textual representation of the attribute's value in the tuple, with standard encoding for HTTP.
2. Specify the parameters that are required for the operator, such as information to determine the location of the external system and any batching related configuration. For HTTP POST, a url parameter is required to specify the URL for the POST and the TupleConsumer sample class provides parameters to configure batching.
3. Design error handling for errors that are communicated with the external system.
   1. Handling errors by trying the connection to the external system again, logging the data that cannot be transmitted, or by passing the exception to the IBM® Streams instance, which causes the PE to terminate.
   2. Handling connection errors, such as failure to connect to remote database.
   3. Handling data errors, such as constraint violations for an external database.
4. Create an initial implementation of com.ibm.streams.operator.Operator, either directly, by extending com.ibm.streams.operator.AbstractOperator, TupleConsumer, or any other existing Operator implementation suitable for the wanted goal.
5. Implement the Operator.initialize() method, ensuring super.initialize() is called as required. Typically the initialize() method performs the following operations:
   • Fetches and validates the parameters that are passed in from SPL.
   • Sets up connections to the external system.
6. Implement Operator.process() to process incoming tuples. Typically, the Operator.process() performs the following operations:
   • Implements batching as required
   • Adds the tuple to the batch or processes it immediately
   • Converts information from tuples into the form that is expected by the external system's Java API
   • Transmits the information to the external system.
7. Implement the logic to process incoming punctuation. If batching is implemented, then the processing of the final punctuation (Punctuation.FINAL_MARKER) must ensure that the batch processing is completed. If the final punctuation is ignored, then the operator might be shut down while a batch is outstanding.
8. Implement Operator.shutdown() to handle any cleanup, such as closing connections to external systems. If any processing requires asynchronous processing, then ThreadFactory or ScheduledExectorService provided by the OperatorContext must be used. Use of these standard Java interfaces that are provided by the OperatorContext allows the IBM Streams instance to track which operators have outstanding work to be done and thus allow the asynchronous work to be completed when a final punctuation is received.

/**
 * URL of the HTTP server we will be posting data to.
 */
private URL url;

/**
 * Initialize by setting the URL and the batch size.
 */
@Override
public void initialize(OperatorContext context) throws Exception {
    super.initialize(context);

    url = getURL();

    setBatchSize(batchSize());
}

/**
 * Get the URL for the POST requests from the required parameter url.
 * Sub-classes may override this to set the URL another way.
 * 
 * @return URL for POST requests
 * @throws MalformedURLException
 */
protected URL getURL() throws MalformedURLException {
    String urlPath = getOperatorContext().getParameterValues("url").get(0);
    return new URL(urlPath);
}

/**
 * Get the batch size to use from the parameter batchSize using 1 if that is
 * not set. Sub-classes may override this to set the batchSize another way.
 * 
 * @return batchSize to use
 */
protected int batchSize() {
    List<String> bp = getOperatorContext().getParameterValues("batchSize");
    if (bp.isEmpty())
        return 1;

      return Integer.getInteger(bp.get(0)); 
}

@Override
protected final boolean processBatch(Queue<BatchedTuple> batch)
        throws Exception {

    StringBuilder data = new StringBuilder(1024);

    for (BatchedTuple item : batch) {

        StreamSchema schema = item.getStream().getStreamSchema();

        for (Attribute attribute : schema) {
            if (data.length() != 0)
                data.append('&');

            data.append(URLEncoder.encode(attribute.getName(), "UTF-8"));
            data.append('=');
            data.append(URLEncoder.encode(item.getTuple().getString(
                    attribute.getName()), "UTF-8"));
        }
    }

    // Send data
    URLConnection conn = url.openConnection();
    conn.setDoOutput(true);
    OutputStreamWriter wr = new OutputStreamWriter(conn.getOutputStream(),
            "UTF-8");
    wr.write(data.toString());
    wr.flush();

    // Get the response
    BufferedReader rd = new BufferedReader(new InputStreamReader(conn
            .getInputStream()));

    String line;
    while ((line = rd.readLine()) != null) {
        trace.log(TraceLevel.DEBUG, line);
    }
    wr.close();
    rd.close();

    return false;
  }
}