ds.granules.communication.funnel

Class FunnelStreamDatasetImpl

java.lang.Object

ds.granules.communication.funnel.FunnelStreamDatasetImpl

All Implemented Interfaces:

ds.funnel.data.format.FormatMarshaller, Dataset, StreamDataset

public class FunnelStreamDatasetImpl
extends Object
implements StreamDataset

This class is a streaming implementation of the dataset. Additionally, this class implements the StreamingAccess interface which outlines methods to access the data streams that are available.

Author:

Shrideep Pallickara

Nested Class Summary

Nested classes/interfaces inherited from interface ds.granules.dataset.Dataset

Dataset.DatasetType

Field Summary

Fields

Modifier and Type	Field and Description
static String	MODULE

Constructor Summary

Constructors

Constructor and Description
FunnelStreamDatasetImpl()
FunnelStreamDatasetImpl(String datasetIdentifier, String datasetDescription) Share the instance of the client service, so that multiple connections are not created to the same broker.

Method Summary

Methods

Modifier and Type	Method and Description
int	addInputStream(ds.funnel.topic.Topic topic) Add an input stream to this dataset.
int	addInputStream(ds.funnel.topic.Topic topic, String instanceIdentifier) Add an input stream to the data set
int	addInputStreams(Collection<ds.funnel.topic.Topic> topics) Adds a set of input streams to this dataset.
int	addOutputStream(ds.funnel.topic.Topic topic) Add an output stream to this dataset.
int	addOutputStreams(Collection<ds.funnel.topic.Topic> topics) Adds a set of output streams to this dataset.
boolean	close() Close access to this dataset.
DataAvailabilityNotifier	getDataAvailabilityNotifier() Retrieve the data availability notifier for this dataset.
String	getDatasetIdentifier() Retrieve the identifier for this dataset.
Dataset.DatasetType	getDatasetType() Retrieve the type of the data set
String	getDescription() Retrieve a description of the dataset
Collection<ds.funnel.topic.Topic>	getInputStreams() Retrieve a collection of the constituent input streams.
long	getLastModificationTime() Retrieve the last update to this dataset
int	getNumberOfInputStreams() Retrieve the number of input streams in this dataset
int	getNumberOfOutputStreams() Retrieve the number of output streams in this dataset.
Collection<ds.funnel.topic.Topic>	getOutputStreams() Retrieve a collection of the constituent output streams.
ds.funnel.topic.TopicDataEvent	getStreamingData() Retrieves a portion of the streaming dataset when data is available.
boolean	hasInputStream(ds.funnel.topic.Topic topic) Check to see if this dataset has a specified input stream.
boolean	hasInputStreams() Check to see if this dataset has any registered input streams.
boolean	hasOutputStream(ds.funnel.topic.Topic topic) Check to see if this dataset has a specified output stream.
boolean	hasOutputStreams() Check to see if this dataset has any registered output streams.
boolean	initializeDataset() Invokes the initialization of the dataset.
boolean	isAvailable() A check to see if the dataset is available
boolean	isDataAvailable() This polls the dataset to see if data is available for consumption.
boolean	isInitialized() Check to see if the dataset has been initialized
void	marshall(ds.funnel.data.format.FormatWriter mFormat)
int	removeInputStream(ds.funnel.topic.Topic topic) Removes an input stream from this dataset.
int	removeInputStreams(Collection<ds.funnel.topic.Topic> topics) Removes a set of input streams from this dataset.
int	removeOutputStream(ds.funnel.topic.Topic topic) Removes an output stream from this dataset.
int	removeOutputStreams(Collection<ds.funnel.topic.Topic> topics) Removes a set of output streams from this dataset.
void	setDatasetDescription(String description) Sets the dataset description
void	setDatasetIdentifier(String identifier) Sets the dataset identifier
boolean	supportsDataAvailabilityNotifications() Indicates whether this dataset supports dynamic data availability notifications
void	unmarshall(ds.funnel.data.format.FormatReader mFormat)
void	writeStreamingData(ds.funnel.topic.Topic topic, byte[] payload) This method writes data to the output dataset.
void	writeStreamingData(ds.funnel.topic.TopicDataEvent mtdEvent) This method writes data to the output dataset.
void	writeStreamingDataAsBatch(ds.funnel.topic.Topic topic, StreamEvent event)

Methods inherited from class java.lang.Object

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

Field Detail

MODULE

public static final String MODULE

See Also:

Constant Field Values

Constructor Detail

FunnelStreamDatasetImpl

public FunnelStreamDatasetImpl()

FunnelStreamDatasetImpl

public FunnelStreamDatasetImpl(String datasetIdentifier,
                       String datasetDescription)

Share the instance of the client service, so that multiple connections are not created to the same broker.

Parameters:

datasetIdentifier - The identifier of the dataset

datasetDescription - The descriptor for this dataset.

Method Detail

getDatasetType

public Dataset.DatasetType getDatasetType()

Description copied from interface: Dataset

Retrieve the type of the data set

Specified by:

getDatasetType in interface Dataset

Returns:

type of the dataset

getDescription

public String getDescription()

Description copied from interface: Dataset

Retrieve a description of the dataset

Specified by:

getDescription in interface Dataset

Returns:

A description of the dataset.

getDatasetIdentifier

public String getDatasetIdentifier()

Description copied from interface: Dataset

Retrieve the identifier for this dataset. This would typically be a UUID which uniquely identifies the dataset in question.

Specified by:

getDatasetIdentifier in interface Dataset

Returns:

The unique identifier for this dataset.

setDatasetDescription

public void setDatasetDescription(String description)

Description copied from interface: StreamDataset

Sets the dataset description

Specified by:

setDatasetDescription in interface StreamDataset

Parameters:

description - Description

setDatasetIdentifier

public void setDatasetIdentifier(String identifier)

Description copied from interface: StreamDataset

Sets the dataset identifier

Specified by:

setDatasetIdentifier in interface StreamDataset

Parameters:

identifier - identifier

isInitialized

public boolean isInitialized()

Description copied from interface: Dataset

Check to see if the dataset has been initialized

Specified by:

isInitialized in interface Dataset

Returns:

true if the dataset is initialized; false otherwise.

initializeDataset

public boolean initializeDataset()
                          throws DatasetException

Description copied from interface: Dataset

Invokes the initialization of the dataset. This is invoked only when the node is ready to execute the operation that will operate on this dataset.

Specified by:

initializeDataset in interface Dataset

Returns:

true if the dataset has been initialized; false otherwise.

Throws:

DatasetException

isAvailable

public boolean isAvailable()

Description copied from interface: Dataset

A check to see if the dataset is available

Specified by:

isAvailable in interface Dataset

Returns:

true if the dataset is available; false otherwise.

isDataAvailable

public boolean isDataAvailable()

Description copied from interface: Dataset

This polls the dataset to see if data is available for consumption.

Specified by:

isDataAvailable in interface Dataset

Returns:

true if data is available on this dataset; false otherwise.

getLastModificationTime

public long getLastModificationTime()

Description copied from interface: Dataset

Retrieve the last update to this dataset

Specified by:

getLastModificationTime in interface Dataset

Returns:

The time at which this dataset was last updated. The time is returned in milliseconds.

supportsDataAvailabilityNotifications

public boolean supportsDataAvailabilityNotifications()

Description copied from interface: Dataset

Indicates whether this dataset supports dynamic data availability notifications

Specified by:

supportsDataAvailabilityNotifications in interface Dataset

Returns:

true if this dataset supports data availability notifications;false otherwise.

getDataAvailabilityNotifier

public DataAvailabilityNotifier getDataAvailabilityNotifier()

Description copied from interface: Dataset

Retrieve the data availability notifier for this dataset. This method will return a NULL if the underlying dataset does not support availability notifications.

Specified by:

getDataAvailabilityNotifier in interface Dataset

Returns:

The data availability notifier for this dataset.

getStreamingData

public ds.funnel.topic.TopicDataEvent getStreamingData()
                                                throws DatasetException

Description copied from interface: StreamDataset

Retrieves a portion of the streaming dataset when data is available. This method will return a NULL if data is not available. In general, this method should be used in tandem with the DataAvailabilityNotifier.

Specified by:

getStreamingData in interface StreamDataset

Returns:

A fragment of the streaming dataset

Throws:

DatasetException - If there are problems performing the operation

writeStreamingData

public void writeStreamingData(ds.funnel.topic.TopicDataEvent mtdEvent)
                        throws DatasetException

Description copied from interface: StreamDataset

This method writes data to the output dataset. This method throws an exception if the Stream producers have not been configured appropriately.

Specified by:

writeStreamingData in interface StreamDataset

Parameters:

mtdEvent - The stream fragment that needs to be written

Throws:

DatasetException - If there are problems performing the operation

writeStreamingData

public void writeStreamingData(ds.funnel.topic.Topic topic,
                      byte[] payload)
                        throws DatasetException

Description copied from interface: StreamDataset

This method writes data to the output dataset. This method throws an exception if the Stream producers have not been configured appropriately.

Specified by:

writeStreamingData in interface StreamDataset

Parameters:

topic - The stream being written to

payload - The payload for the individual stream fragments

Throws:

DatasetException - If there are problems performing the operation

writeStreamingDataAsBatch

public void writeStreamingDataAsBatch(ds.funnel.topic.Topic topic,
                             StreamEvent event)
                               throws DatasetException

Specified by:

writeStreamingDataAsBatch in interface StreamDataset

Throws:

DatasetException

addInputStream

public int addInputStream(ds.funnel.topic.Topic topic)
                   throws DatasetException

Description copied from interface: StreamDataset

Add an input stream to this dataset.

Specified by:

addInputStream in interface StreamDataset

Parameters:

topic - The topic of the stream to be added

Returns:

The total number of input streams in this dataset

Throws:

DatasetException - If there are problems adding this stream to the set of managed streams

addInputStream

public int addInputStream(ds.funnel.topic.Topic topic,
                 String instanceIdentifier)
                   throws DatasetException

Description copied from interface: StreamDataset

Add an input stream to the data set

Specified by:

addInputStream in interface StreamDataset

Parameters:

topic - Topic of the data stream to be added.

instanceIdentifier - Instance identifier of the operation base

Returns:

The total number of input streams.

Throws:

DatasetException - If there are problems adding this stream to the set of managed streams

removeInputStream

public int removeInputStream(ds.funnel.topic.Topic topic)
                      throws DatasetException

Description copied from interface: StreamDataset

Removes an input stream from this dataset.

Specified by:

removeInputStream in interface StreamDataset

Parameters:

topic - The topic of the stream to be removed

Returns:

The total number of input streams that remain in this dataset

Throws:

DatasetException - If there are problems removing this stream from the set of managed streams

addInputStreams

public int addInputStreams(Collection<ds.funnel.topic.Topic> topics)
                    throws DatasetException

Description copied from interface: StreamDataset

Adds a set of input streams to this dataset.

Specified by:

addInputStreams in interface StreamDataset

Parameters:

topics - A collection of streams to be added

Returns:

The total number of input streams in this dataset

Throws:

DatasetException - If there are problems adding these streams to the set of managed streams

removeInputStreams

public int removeInputStreams(Collection<ds.funnel.topic.Topic> topics)
                       throws DatasetException

Description copied from interface: StreamDataset

Removes a set of input streams from this dataset.

Specified by:

removeInputStreams in interface StreamDataset

Parameters:

topics - A collection of streams to be removed

Returns:

The total number of input streams in this dataset

Throws:

DatasetException - If there are problems removing these streams to the set of managed streams

hasInputStreams

public boolean hasInputStreams()

Description copied from interface: StreamDataset

Check to see if this dataset has any registered input streams.

Specified by:

hasInputStreams in interface StreamDataset

Returns:

true if this dataset has any registered input streams;false otherwise.

hasInputStream

public boolean hasInputStream(ds.funnel.topic.Topic topic)

Description copied from interface: StreamDataset

Check to see if this dataset has a specified input stream.

Specified by:

hasInputStream in interface StreamDataset

Returns:

true if this dataset has the specified input stream;false otherwise.

getNumberOfInputStreams

public int getNumberOfInputStreams()

Description copied from interface: StreamDataset

Retrieve the number of input streams in this dataset

Specified by:

getNumberOfInputStreams in interface StreamDataset

Returns:

The number of input streams in this dataset

getInputStreams

public Collection<ds.funnel.topic.Topic> getInputStreams()

Description copied from interface: StreamDataset

Retrieve a collection of the constituent input streams.

Specified by:

getInputStreams in interface StreamDataset

Returns:

The collection of the constituent input streams.

addOutputStream

public int addOutputStream(ds.funnel.topic.Topic topic)
                    throws DatasetException

Description copied from interface: StreamDataset

Add an output stream to this dataset.

Specified by:

addOutputStream in interface StreamDataset

Parameters:

topic - The topic of the stream to be added

Returns:

The total number of output streams in this dataset

Throws:

DatasetException - If there are problems adding this stream to the set of managed streams

removeOutputStream

public int removeOutputStream(ds.funnel.topic.Topic topic)
                       throws DatasetException

Description copied from interface: StreamDataset

Removes an output stream from this dataset.

Specified by:

removeOutputStream in interface StreamDataset

Parameters:

topic - The topic of the stream to be removed

Returns:

The total number of output streams that remain in this dataset

Throws:

DatasetException - If there are problems removing this stream from the set of managed streams

addOutputStreams

public int addOutputStreams(Collection<ds.funnel.topic.Topic> topics)
                     throws DatasetException

Description copied from interface: StreamDataset

Adds a set of output streams to this dataset.

Specified by:

addOutputStreams in interface StreamDataset

Parameters:

topics - A collection of streams to be added

Returns:

The total number of output streams in this dataset

Throws:

DatasetException - If there are problems adding these streams to the set of managed streams

removeOutputStreams

public int removeOutputStreams(Collection<ds.funnel.topic.Topic> topics)
                        throws DatasetException

Description copied from interface: StreamDataset

Removes a set of output streams from this dataset.

Specified by:

removeOutputStreams in interface StreamDataset

Parameters:

topics - A collection of streams to be removed

Returns:

The total number of output streams in this dataset

Throws:

DatasetException - If there are problems removing these streams from the set of managed streams

hasOutputStreams

public boolean hasOutputStreams()

Description copied from interface: StreamDataset

Check to see if this dataset has any registered output streams.

Specified by:

hasOutputStreams in interface StreamDataset

Returns:

true if this dataset has any registered output streams;false otherwise.

hasOutputStream

public boolean hasOutputStream(ds.funnel.topic.Topic topic)

Description copied from interface: StreamDataset

Check to see if this dataset has a specified output stream.

Specified by:

hasOutputStream in interface StreamDataset

Returns:

true if this dataset has the specified output stream;false otherwise.

getNumberOfOutputStreams

public int getNumberOfOutputStreams()

Description copied from interface: StreamDataset

Retrieve the number of output streams in this dataset.

Specified by:

getNumberOfOutputStreams in interface StreamDataset

Returns:

The number of output streams in this dataset.

getOutputStreams

public Collection<ds.funnel.topic.Topic> getOutputStreams()

Description copied from interface: StreamDataset

Retrieve a collection of the constituent output streams.

Specified by:

getOutputStreams in interface StreamDataset

Returns:

The collection of the constituent output streams.

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

Throws:

ds.funnel.data.format.FormatException

unmarshall

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

Specified by:

unmarshall in interface StreamDataset

Throws:

ds.funnel.data.format.FormatException

close

public boolean close()
              throws DatasetException

Description copied from interface: Dataset

Close access to this dataset. This method returns true if a graceful closure of the dataset was possible. This method will try its best to close the dataset. If it fails to close the dataset an exception, outlining the problems that were encountered will, be thrown.

Specified by:

close in interface Dataset

Returns:

true if a graceful closure of this dataset was performed;false otherwise.

Throws:

DatasetException