org.apache.streams.local.tasks

Class BaseStreamsTask

java.lang.Object

org.apache.streams.local.tasks.BaseStreamsTask

All Implemented Interfaces:

Runnable, StreamsTask

Direct Known Subclasses:

StreamsMergeTask, StreamsPersistWriterTask, StreamsProcessorTask, StreamsProviderTask

public abstract class BaseStreamsTask
extends Object
implements StreamsTask

BaseStreamsTask is the primary abstract StreamsTask.

Field Summary

Fields

Modifier and Type	Field and Description
protected org.apache.streams.config.StreamsConfiguration	streamConfig

Constructor Summary

Constructors

Constructor and Description
BaseStreamsTask(org.apache.streams.config.StreamsConfiguration config)

Method Summary

Modifier and Type	Method and Description
void	addInputQueue(BlockingQueue<StreamsDatum> inputQueue) Add an input Queue for this task.
void	addOutputQueue(BlockingQueue<StreamsDatum> outputQueue) Add an output Queue for this task.
protected void	addToOutgoingQueue(StreamsDatum datum) Adds a StreamDatum to the outgoing queues.
protected StreamsDatum	cloneStreamsDatum(StreamsDatum datum) //TODO LOCAL MODE HACK.
List<BlockingQueue<StreamsDatum>>	getInputQueues() Returns the input queues that have been set for this task.
protected StreamsDatum	getNextDatum() Deprecated.
List<BlockingQueue<StreamsDatum>>	getOutputQueues() Returns the output queues that have been set for this task
long	getStartedAt()
String	getStreamIdentifier()
boolean	isWaiting() Returns true if the task is waiting on more data to process
void	setStartedAt()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.streams.local.tasks.StreamsTask

isRunning, setStreamConfig, setStreamsTaskCounter, stopTask

Methods inherited from interface java.lang.Runnable

run

Field Detail

streamConfig

protected org.apache.streams.config.StreamsConfiguration streamConfig

Constructor Detail

BaseStreamsTask

public BaseStreamsTask(org.apache.streams.config.StreamsConfiguration config)

Method Detail

addInputQueue

public void addInputQueue(BlockingQueue<StreamsDatum> inputQueue)

Description copied from interface: StreamsTask

Add an input Queue for this task.

Specified by:

addInputQueue in interface StreamsTask

addOutputQueue

public void addOutputQueue(BlockingQueue<StreamsDatum> outputQueue)

Description copied from interface: StreamsTask

Add an output Queue for this task.

Specified by:

addOutputQueue in interface StreamsTask

getInputQueues

public List<BlockingQueue<StreamsDatum>> getInputQueues()

Description copied from interface: StreamsTask

Returns the input queues that have been set for this task.

Specified by:

getInputQueues in interface StreamsTask

Returns:

list of input queues

getOutputQueues

public List<BlockingQueue<StreamsDatum>> getOutputQueues()

Description copied from interface: StreamsTask

Returns the output queues that have been set for this task

Specified by:

getOutputQueues in interface StreamsTask

Returns:

list of output queues

getNextDatum

@Deprecated
protected StreamsDatum getNextDatum()

Deprecated.

SHOULD NOT BE NECCESARY, WILL REMOVE. Round Robins through input queues to get the next StreamsDatum. If all input queues are empty, it will return null.

Returns:

the next StreamsDatum or null if all input queues are empty.

addToOutgoingQueue

protected void addToOutgoingQueue(StreamsDatum datum)
                           throws InterruptedException

Adds a StreamDatum to the outgoing queues. If there are multiple queues, it uses serialization to create clones of the datum and adds a new clone to each queue.

Parameters:

datum -

Throws:

InterruptedException

isWaiting

public boolean isWaiting()

Description copied from interface: StreamsTask

Returns true if the task is waiting on more data to process

Specified by:

isWaiting in interface StreamsTask

Returns:

true, if waiting on more data to process

cloneStreamsDatum

protected StreamsDatum cloneStreamsDatum(StreamsDatum datum)

//TODO LOCAL MODE HACK. Need to fix In order for our data streams to ported to other data flow frame works(Storm, Hadoop, Spark, etc) we need to be able to enforce the serialization required by each framework. This needs some thought and design before a final solution is made. In order to be able to copy/clone StreamDatums the orginal idea was to force all StreamsDatums to be java serializable. This was seen as unacceptable for local mode. So until we come up with a solution to enforce serialization and be compatiable across multiple frame works, this hack is in place. If datum.document is Serializable, we use serialization to clone a new copy. If it is not Serializable we attempt different methods using an com.fasterxml.jackson.databind.ObjectMapper to copy/clone the StreamsDatum. If the object is not clonable by these methods, an error is reported to the logging and a NULL object is returned.

Parameters:

datum -

Returns:

getStartedAt

public long getStartedAt()

setStartedAt

public void setStartedAt()

getStreamIdentifier

public String getStreamIdentifier()