Implementing a blocking source operator with the Java Operator API

The sample operator com.ibm.streams.operator.samples.sources.SystemPropertySource is a simple source operator that uses a Thread to read Java system properties and deliver them as a stream of two ustring attributes, name and value. SystemPropertySource extends the sample pattern abstract class ProcessTupleProducer, which allows subclasses to implement a process() method to fetch data and submit tuples, thus hiding the details of creating a Thread and implementing Runnable. This use of a single process() method is similar to the C++ operator runtime API style of using a process() function as the entry point for one or more threads. ProcessTupleProducer itself extends TupleProducer to allow a delay to be specified before tuples are submitted. Looking at the ProcessTupleProducer.initialize() method, see that it creates a new thread using the operator's ThreadFactory and that thread calls the abstract no-argument process() method when it is run.

public synchronized void initialize(OperatorContext context) throws Exception {
        super.initialize(context);
        
        /*
         * Create the thread at initialize time but do not start it.
         * The thread will be started by startProcessing() which will
         * be called at allPortsReady() time if no delay is specified,
         * otherwise it will be called once the delay has expired.
         */
        processThread = getOperatorContext().getThreadFactory().newThread(
                new Runnable() {

                    @Override
                    public void run() {
                        try {
                            process();
                        } catch (Exception e) {
                            trace.log(TraceLevel.ERROR, e.getMessage(), e);
                        }
                    }

                });
        
        /*
         * Set the thread not to be a daemon to ensure that the SPL runtime
         * will wait for the thread to complete before determining the
         * operator is complete.
         */
        processThread.setDaemon(false);
    }

The implementation of Runnable that calls process is provided by an anonymous inner class that calls process() and logs any thrown exception to the error log. The thread is set to a non-daemon thread to ensure that the IBM Streams instance waits for the thread to complete before it marks the operator as complete. When initialize() completes, the thread exists but is not started, ProcessTupleProducer then implements startProcessing() to start the thread during or after allPortsReady() (if a delay was specified) is called.

protected synchronized void startProcessing() {
        processThread.start();
    }

ProcessTupleProducer thus provides the framework for subclasses to implement the no-argument process() to loop and fetch data. SystemPropertySource has a simple process() method that fetches the Java system properties and submits a tuple for each property.

protected void process() throws Exception {
        
        final StreamingOutput<OutputTuple> out = getOutput(0);
        for (Enumeration<?> e = System.getProperties().propertyNames();
            e.hasMoreElements(); )
        {
            String name = (String) e.nextElement();
            String value = System.getProperty(name);
            
            OutputTuple tuple = out.newTuple();
            
            tuple.setString("name", name);
            tuple.setString("value", value);
            
            out.submit(tuple);
        }
        
        // Finite set of properties so send a final mark.
        out.punctuate(Punctuation.FINAL_MARKER);
    }

Since the properties are a finite set, the operator submits a final punctuation mark when all the properties are submitted as tuples and then returns, which completes the thread and the operator.