Implementing a sink operator using the Java™ Operator API

• com.ibm.streams.operator.samples.sinks.HttpPOST shows the logic specific to creating an HTTP POST request from tuples.
• com.ibm.streams.operator.samples.patterns.TupleConsumer is an abstract class that demonstrates a flexible batching of tuples that is intended for use as a sink operator.

HttpPost extends TupleConsumer: Starting with HttpPost, there are two key methods initialize() and processBatch(). The initialize() method uses the SPL parameter url to determine the URL for the POST request. A connection attempt is not made to validate the URL since the only knowledge is that the URL is suitable for a POST request and a POST request with no data might have unknown effects on the server.

/**
 * 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)); 
}

The processBatch() method is an abstract method that the super-class TupleConsumer requires and whose purpose is to take a batch of Tuple objects and transmit them to the external system. For HttpPost each attribute in the tuple is converted into a name-value pair with URL encoding. A single HTTP POST request might contain multiple tuples, with each name, value pair repeated multiple times.

@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;
  }
}