ds.granules.streaming.core

Class StreamBase

java.lang.Object

ds.granules.operation.OperationBase

ds.granules.operation.MapReduceBase

ds.granules.streaming.core.StreamBase

All Implemented Interfaces:

ds.funnel.data.format.FormatMarshaller, Map, MapReduce, Operation, OperationConfiguration, Reduce

Direct Known Subclasses:

StreamProcessor, StreamSource

public abstract class StreamBase
extends MapReduceBase

This class is the base class for StreamSource and StreamProcessor. Extends MapReduceBase and provides a more abstract streaming API.

Author:

: Thilina Buddhika

Nested Class Summary

Nested classes/interfaces inherited from interface ds.granules.operation.Operation

Operation.OperationType

Field Summary

Fields

Modifier and Type	Field and Description
protected String	identifier
protected Map<String,String>	incomingStreamTypes
protected boolean	isBufferingAccess
protected Map<String,String>	outGoingStreamTypes
protected StreamDataset	streamDataset

Fields inherited from class ds.granules.operation.MapReduceBase

defaultOutputStream, mapReduceDataset, MODULE

Constructor Summary

Constructors

Constructor and Description
StreamBase()

Method Summary

Methods

Modifier and Type	Method and Description
protected void	addStreamConsumer(ds.funnel.topic.Topic topic, StreamBase destination)
protected abstract void	declareOutputStreams() The classes that extend StreamProcessor and StreamSource should implement this method in order to declare output streams.
protected void	declareStream(String streamId, String streamType) Declare an output stream from the Stream operator.
protected String	getDatasetIdentifier()
protected StreamDataset	getDefaultStreamDataset()
protected String	getStreamIdentifier(String streamIdentifier)
void	initialize(Properties providedDirectives) Initialize the operation based on the specified processing directives.
void	marshall(ds.funnel.data.format.FormatWriter mFormat)
void	unmarshall(ds.funnel.data.format.FormatReader mFormat) Unmarshall the operation from a set of marshalled bytes.
protected void	writeToStream(String streamId, StreamEvent event) Write a stream data element to a Stream.

Methods inherited from class ds.granules.operation.MapReduceBase

addMap, addMapper, addReduce, addReducer, createResults, getAwaitingOutputsFrom, getDatasetCollection, getMapIdentifiers, getMapReduceDataset, getNumberOfMaps, getNumberOfReducers, getReduceIdentifiers, isAwaitingOutputs, removeMap, removeMapper, removeReduce, removeReducer, resetOutputTracker, setReceivedOutputFrom, writeResults, writeResults, writeResults, writeResults

Methods inherited from class ds.granules.operation.OperationBase

changeToExecuteFixedNumberOfTimes, changeToExecuteOnce, changeToExecutePeriodically, changeToExecuteWhenDataAvailable, changeToStayAlive, dispose, execute, getDomain, getExecutionProfile, getInstanceIdentifier, getOperationIdentifier, getProcessingDirectives, getVersionInformation, hasDatasetCollection, hasExecutionProfile, hasProcessingDirectives, isInitialized, markInitializationAsComplete, needsInitialization, setAsExecuteContinuously, setAsExecuteFixedNumberOfTimes, setAsExecuteOnce, setAsExecutePeriodically, setAsExecuteWhenDataAvailable, setAsStayAlive, setDatasetCollection, setDomain, setInitializationComplete, setInitializationNeeded, setInstanceIdentifier, setOperationIdentifier, setProcessingDirectives, setTeminationConditionReached, setVersionInformation, terminationConditionReached, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface ds.granules.operation.Operation

dispose, execute, getDomain, getExecutionProfile, getInstanceIdentifier, getOperationIdentifier, getProcessingDirectives, getVersionInformation, hasDatasetCollection, hasExecutionProfile, hasProcessingDirectives, isInitialized, needsInitialization, terminationConditionReached, toString

Field Detail

streamDataset

protected StreamDataset streamDataset

identifier

protected String identifier

outGoingStreamTypes

protected Map<String,String> outGoingStreamTypes

incomingStreamTypes

protected Map<String,String> incomingStreamTypes

isBufferingAccess

protected boolean isBufferingAccess

Constructor Detail

StreamBase

public StreamBase()

Method Detail

initialize

public void initialize(Properties providedDirectives)
                throws ProcessingException

Description copied from interface: Operation

Initialize the operation based on the specified processing directives. The initialization will throw a ProcessingException is there are problems.

Specified by:

initialize in interface Operation

Specified by:

initialize in class OperationBase

Parameters:

providedDirectives - The specified processing directives

Throws:

ProcessingException - If there are problems initializing based on the specified processing directives.

unmarshall

public void unmarshall(ds.funnel.data.format.FormatReader mFormat)
                throws ds.funnel.data.format.FormatException

Description copied from interface: Operation

Unmarshall the operation from a set of marshalled bytes.

Specified by:

unmarshall in interface Operation

Overrides:

unmarshall in class MapReduceBase

Parameters:

mFormat - The byte[] representation of the operation.

Throws:

ds.funnel.data.format.FormatException

marshall

public void marshall(ds.funnel.data.format.FormatWriter mFormat)
              throws ds.funnel.data.format.FormatException

Specified by:

marshall in interface ds.funnel.data.format.FormatMarshaller

Overrides:

marshall in class MapReduceBase

Throws:

ds.funnel.data.format.FormatException

declareOutputStreams

protected abstract void declareOutputStreams()
                                      throws StreamingGraphConfigurationException

The classes that extend StreamProcessor and StreamSource should implement this method in order to declare output streams.

Throws:

StreamingGraphConfigurationException - Error in the stream processing graph.

getDatasetIdentifier

protected String getDatasetIdentifier()

getStreamIdentifier

protected String getStreamIdentifier(String streamIdentifier)

declareStream

protected void declareStream(String streamId,
                 String streamType)
                      throws StreamingGraphConfigurationException

Declare an output stream from the Stream operator.

Parameters:

streamId - An Identifier for the stream. This should be unique within a topology.

streamType - Type of the Stream event which should be an implementation of StreamEvent.

Throws:

StreamingGraphConfigurationException - Error in the stream processing graph.

addStreamConsumer

protected void addStreamConsumer(ds.funnel.topic.Topic topic,
                     StreamBase destination)
                          throws StreamingGraphConfigurationException,
                                 StreamingDatasetException

Throws:

StreamingGraphConfigurationException

StreamingDatasetException

writeToStream

protected void writeToStream(String streamId,
                 StreamEvent event)
                      throws StreamingDatasetException

Write a stream data element to a Stream.

Parameters:

streamId - Identifier of the stream to which the data is written

event - Stream data element

Throws:

StreamingDatasetException - Error when writing to the stream

getDefaultStreamDataset

protected StreamDataset getDefaultStreamDataset()