com.ibm.streams.operator

Interface Operator

All Known Implementing Classes:

AbstractOperator, AbstractWindowOperator

public interface Operator

Operator for processing and producing streams of tuples. An Operator is an event processor for tuples and punctuation marks that arrives on its input ports. Incoming tuples result in a call to process(StreamingInput, Tuple) and punctuation marks a call to processPunctuation(StreamingInput, StreamingData.Punctuation). These calls are invoked by the SPL runtime. Operator implementations must be thread-safe as different threads may invoke the operator's methods and tuples and punctuation marks may arrive concurrently, even on the same input port.
An implementation of Operator becomes a Java primitive operator when it is annotated with PrimitiveOperator. A Java primitive operator becomes part of an SPL toolkit and thus can be called from an SPL application. The com.ibm.streams.operator.model package contains a number of annotations that allow an implementation of Operator to fully describe itself as a Java primitive operator.

An Operator has an associated OperatorContext that describes its execution environment, including references to objects representing its input ports (StreamingInput) and output ports (StreamingOutput).

An Operator is initialized by the SPL runtime invoking initialize(OperatorContext) passing the OperatorContext for the specific SPL operator invocation. Once initialized the Operator is notified that its ports are ready by the SPL runtime invoking allPortsReady().
When the processing element(PE) that hosts the operator is being shutdown, the SPL runtime will invoke shutdown(), which may take place while tuple and punctuation processing functions are active. The Operator's implementation of its shutdown method should complete any asynchronous activity and release any resources.

Tuples are submitted to output ports using submit methods on StreamingOutput, such as StreamingOutput.submit(Tuple). Output ports are punctuated using StreamingOutput.punctuate(StreamingData.Punctuation).

An operator performs asynchronous actions using tasks with a cheduledExecutorService obtained from OperatorContext.getScheduledExecutorService() or threads created from the ThreadFactory obtained from OperatorContext.getThreadFactory() . Source operators typically initiate asynchronous actions in their implementation of allPortsReady() to produce data and submit it as tuples on their output ports.

Operators with a windowed input port specify a StreamWindowListener using StreamWindow.registerListener to handle window events. If the window is partitioned the operator specifies a StreamWindowPartitioner using StreamWindow.registerPartitioner.

A number of abstract and concrete sample operators are provided by InfoSphere Streams.

Nested Class Summary

Nested Classes

Modifier and Type	Interface and Description
static interface	Operator.TagNames Tag names for use with Operator#setTagData(String, Map.

Method Summary

Methods

Modifier and Type	Method and Description
void	allPortsReady() Called once initialization is complete and all input and output ports are connected and ready to process and submit tuples.
void	initialize(OperatorContext context) Initialize this operator.
void	process(StreamingInput<Tuple> port, Tuple tuple) Process an incoming tuple that arrived on the specified port.
void	processPunctuation(StreamingInput<Tuple> port, StreamingData.Punctuation mark) Process an incoming punctuation mark on the specified port.
void	setTagData(java.lang.String tagName, java.util.Map<java.lang.String,java.lang.String> tagValue) Set tag data for the operator.
void	shutdown() Instruct this operator to shutdown.

Method Detail

initialize

void initialize(OperatorContext context)
                throws java.lang.Exception

Initialize this operator. Called once before any tuples are processed.

Parameters:

context - OperatorContext for this operator.

Throws:

java.lang.Exception - Operator failure, will cause the enclosing PE to terminate.

allPortsReady

void allPortsReady()
                   throws java.lang.Exception

Called once initialization is complete and all input and output ports are connected and ready to process and submit tuples. Operators that submit tuples independently of incoming tuples, e.g. source type operators, must wait for this method to be called before submitting any tuples.
Since this is a notification that all input and output ports are ready, it is possible that process and processPunctuation may be called before allPortsReady, due to thread scheduling.

Implementations must not block and must return control to the caller.

This operator's OperatorContextMXBean will also issue a notification when all ports are ready.
Code may register actions to be invoked at all ports ready using the utility method addAllPortsReadyAction or by adding notification listeners to this operator's OperatorContextMXBean.

Throws:

java.lang.Exception - Operator failure, will cause the enclosing PE to terminate.

See Also:

OperatorContextMXBean, OperatorLifeCycle.addAllPortsReadyAction(OperatorContext, javax.management.NotificationListener)

process

void process(StreamingInput<Tuple> port,
           Tuple tuple)
             throws java.lang.Exception

Process an incoming tuple that arrived on the specified port.

Parameters:

port - Port the tuple is arriving on.

tuple - Object representing the incoming tuple.

Throws:

java.lang.Exception - Operator failure, will cause the enclosing PE to terminate.

processPunctuation

void processPunctuation(StreamingInput<Tuple> port,
                      StreamingData.Punctuation mark)
                        throws java.lang.Exception

Process an incoming punctuation mark on the specified port.

Parameters:

port - Port punctuation mark arrived on

mark - value of mark

Throws:

java.lang.Exception - Operator failure, will cause the enclosing PE to terminate.

shutdown

void shutdown()
              throws java.lang.Exception

Instruct this operator to shutdown. Called by the SPL runtime when the processing element (PE) hosting the operator is being shutdown. Implementations should complete or terminate any asynchronous activity and release any resources, such as closing open files.

Prior to this call, the SPL runtime initiates an orderly shutdown on the Operator's ThreadFactory and ScheduledExecutorService so that no new threads or tasks can be created.
Once this method returns the SPL Runtime completes the shutdown of the Operator's ThreadFactory (calling Thread.interrupt on any of its active threads) and ScheduledExecutorService (calling ExecutorService.shutdownNow()).

This operator's OperatorContextMXBean will also issue a shutdown notification and a shutdown complete notification.
Code may register actions to be invoked at shutdown using the utility method addShutdownAction or by adding notification listeners to this operator's OperatorContextMXBean.

Throws:

java.lang.Exception - Error while shutting down.

See Also:

OperatorContext.getThreadFactory(), OperatorContext.getScheduledExecutorService(), OperatorContextMXBean, OperatorLifeCycle.addShutdownAction(OperatorContext, javax.management.NotificationListener), OperatorLifeCycle.closeOnShutdown(OperatorContext, java.io.Closeable)

setTagData

void setTagData(java.lang.String tagName,
              java.util.Map<java.lang.String,java.lang.String> tagValue)

Set tag data for the operator. If this value is Operator.TagNames.OperatorIGC, the tags will be registered in the IG catalog. Other tag names may be defined in future releases.

Since:

InfoSphere® Streams Version 4.1