org.sensorhub.impl.processing

Class AbstractStreamProcess<ConfigType extends StreamProcessConfig>

java.lang.Object

org.sensorhub.impl.module.AbstractModule<ConfigType>

org.sensorhub.impl.processing.AbstractStreamProcess<ConfigType>

Type Parameters:

ConfigType - Type of process configuration

All Implemented Interfaces:

IEventListener, IEventProducer, IDataProducerModule<ConfigType>, IModuleWithDescription, IModule<ConfigType>, IProcessModule<ConfigType>, IStreamProcessModule<ConfigType>

Direct Known Subclasses:

SMLStreamProcess

public abstract class AbstractStreamProcess<ConfigType extends StreamProcessConfig>
extends AbstractModule<ConfigType>
implements IStreamProcessModule<ConfigType>

Class providing default implementation of common stream processing API methods. This can be used as the base for most stream process implementations.
Concrete process implementations generally need to override either AbstractModule.init(ModuleConfig) or start() in order to provide actual process inputs, outputs and parameters.

Since:

Feb 22, 2015

Author:

Alex Robin

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	AbstractStreamProcess.InputData

Field Summary

Fields

Modifier and Type	Field and Description
static String	DEFAULT_ID
protected int	errorCount
protected Map<String,net.opengis.swe.v20.DataComponent>	inputs
protected long	lastUpdatedSensorDescription
protected static int	MAX_ERRORS
protected Map<String,IStreamingDataInterface>	outputInterfaces
protected Map<String,net.opengis.swe.v20.DataComponent>	outputs
protected Map<String,net.opengis.swe.v20.DataComponent>	parameters
protected boolean	paused
protected net.opengis.sensorml.v20.AbstractProcess	processDescription
protected Map<IStreamingDataInterface,AbstractStreamProcess.InputData>	streamSources

Fields inherited from class org.sensorhub.impl.module.AbstractModule

config, eventHandler

Constructor Summary

Constructors

Constructor and Description
AbstractStreamProcess()

Method Summary

Methods

Modifier and Type	Method and Description
protected void	addOutput(IStreamingDataInterface outputInterface) Helper method to make sure derived classes add outputs consistently in the different maps
void	cleanup() Cleans up all ressources used by the module when deleted All persistent resources created by the module should be cleaned when this is called
protected void	connectInput(String inputName, String dataPath, org.vast.process.DataQueue inputQueue)
Map<String,IStreamingDataInterface>	getAllOutputs() Retrieves the list of data outputs
net.opengis.sensorml.v20.AbstractProcess	getCurrentDescription() Retrieves most current SensorML description of the entity whose data is provided by this module.
net.opengis.gml.v32.AbstractFeature	getCurrentFeatureOfInterest() Retrieves the feature of interest for which this producer is currently generating data. In the case of a module generating data from multiple entities (e.g.
Map<String,net.opengis.swe.v20.DataComponent>	getInputDescriptors() Gets the list of inputs needed by this process. Note that input data may contain more than the needed input.
long	getLastDescriptionUpdate() Used to check when SensorML description was last updated.
Map<String,net.opengis.swe.v20.DataComponent>	getOutputDescriptors()
Map<String,net.opengis.swe.v20.DataComponent>	getParameters() Gets the list of parameters for this process. Values should be changed directly inside the objects returned in the map. Parameters that can be changed during processing must be marked as 'updatable' and the new set values are taken into account ASAP by the running process. Changing the value during processing of parameters that are not updatable is either silently ignored or can result in a processing exception.
void	handleEvent(Event<?> e)
void	pause() Pauses processing of the event stream. Incoming events are simply discarded and won't be processed when the process is resumed.
protected abstract void	process(DataEvent lastEvent) Method called everytime a new data event has been received on any input. Concrete implementation must override this method to launch processing when appropriate (e.g.
void	resume() Resumes normal processing of the event stream. Processing may actually resume only when the next event is received.
void	start() Starts the module with the current configuration. init() should always be called before start().
void	stop() Stops the module. All temporary resources created by the module should be cleaned when this is called (ex: memory, files, connections, etc.) stop() can be called right after init() even if start() hasn't been called.

Methods inherited from class org.sensorhub.impl.module.AbstractModule

getConfiguration, getLocalID, getName, init, isEnabled, loadState, registerListener, saveState, unregisterListener, updateConfig

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.sensorhub.api.processing.IStreamProcessModule

isPauseSupported

Methods inherited from interface org.sensorhub.api.processing.IProcessModule

isCompatibleDataSource

Methods inherited from interface org.sensorhub.api.module.IModule

getConfiguration, getLocalID, getName, init, isEnabled, loadState, registerListener, saveState, unregisterListener, updateConfig

Field Detail

DEFAULT_ID

public static final String DEFAULT_ID

See Also:

Constant Field Values

MAX_ERRORS

protected static final int MAX_ERRORS

See Also:

Constant Field Values

inputs

protected Map<String,net.opengis.swe.v20.DataComponent> inputs

outputs

protected Map<String,net.opengis.swe.v20.DataComponent> outputs

parameters

protected Map<String,net.opengis.swe.v20.DataComponent> parameters

outputInterfaces

protected Map<String,IStreamingDataInterface> outputInterfaces

streamSources

protected Map<IStreamingDataInterface,AbstractStreamProcess.InputData> streamSources

processDescription

protected net.opengis.sensorml.v20.AbstractProcess processDescription

lastUpdatedSensorDescription

protected long lastUpdatedSensorDescription

paused

protected boolean paused

errorCount

protected int errorCount

Constructor Detail

AbstractStreamProcess

public AbstractStreamProcess()

Method Detail

process

protected abstract void process(DataEvent lastEvent)
                         throws ProcessException

Method called everytime a new data event has been received on any input.
Concrete implementation must override this method to launch processing when appropriate (e.g. when enough data has been received).
Since data is accumulated in input queues, process should properly remove any data that needs to be discarded from the queues, as early as possible, even when no processing is done.

Parameters:

lastEvent - data event triggering this call

Throws:

ProcessException

addOutput

protected void addOutput(IStreamingDataInterface outputInterface)

Helper method to make sure derived classes add outputs consistently in the different maps

Parameters:

outputInterface -

getInputDescriptors

public Map<String,net.opengis.swe.v20.DataComponent> getInputDescriptors()

Description copied from interface: IProcessModule

Gets the list of inputs needed by this process.
Note that input data may contain more than the needed input. To confirm that a given input is acceptable, call IProcessModule.isCompatibleDataSource(DataSourceConfig)

Specified by:

getInputDescriptors in interface IProcessModule<ConfigType extends StreamProcessConfig>

Returns:

map of descriptors of needed inputs

getOutputDescriptors

public Map<String,net.opengis.swe.v20.DataComponent> getOutputDescriptors()

Specified by:

getOutputDescriptors in interface IProcessModule<ConfigType extends StreamProcessConfig>

Returns:

map of output descriptors

getParameters

public Map<String,net.opengis.swe.v20.DataComponent> getParameters()

Description copied from interface: IStreamProcessModule

Gets the list of parameters for this process.
Values should be changed directly inside the objects returned in the map.
Parameters that can be changed during processing must be marked as 'updatable' and the new set values are taken into account ASAP by the running process.
Changing the value during processing of parameters that are not updatable is either silently ignored or can result in a processing exception.

Specified by:

getParameters in interface IStreamProcessModule<ConfigType extends StreamProcessConfig>

Returns:

map of parameter descriptors

getAllOutputs

public Map<String,IStreamingDataInterface> getAllOutputs()

Description copied from interface: IDataProducerModule

Retrieves the list of data outputs

Specified by:

getAllOutputs in interface IDataProducerModule<ConfigType extends StreamProcessConfig>

Returns:

map of output names -> data interface objects

getCurrentFeatureOfInterest

public net.opengis.gml.v32.AbstractFeature getCurrentFeatureOfInterest()

Description copied from interface: IDataProducerModule

Retrieves the feature of interest for which this producer is currently generating data.

In the case of a module generating data from multiple entities (e.g. sensor network), the feature of interest currently observed by a given entity can be retrieved using IMultiSourceDataProducer.getCurrentFeatureOfInterest(String)

Specified by:

getCurrentFeatureOfInterest in interface IDataProducerModule<ConfigType extends StreamProcessConfig>

Returns:

map of entityID to feature objects

getCurrentDescription

public net.opengis.sensorml.v20.AbstractProcess getCurrentDescription()

Description copied from interface: IModuleWithDescription

Retrieves most current SensorML description of the entity whose data is provided by this module. All implementations must return an instance of AbstractProcess with a valid unique identifier.
In the case of a module generating data from multiple entities (e.g. sensor network), this returns the description of the group as a whole. Descriptions of individual entities within the group are retrived using IMultiSourceDataProducer.getCurrentDescription(String)

Specified by:

getCurrentDescription in interface IModuleWithDescription

Returns:

AbstractProcess SensorML description of the data producer or null if none is available at the time of the call

getLastDescriptionUpdate

public long getLastDescriptionUpdate()

Description copied from interface: IModuleWithDescription

Used to check when SensorML description was last updated. This is useful to avoid requesting the object when it hasn't changed.

Specified by:

getLastDescriptionUpdate in interface IModuleWithDescription

Returns:

Date/time of last description update as unix time (ms since 1970) or Long.MIN_VALUE if description was never updated.

connectInput

protected void connectInput(String inputName,
                String dataPath,
                org.vast.process.DataQueue inputQueue)
                     throws Exception

Throws:

Exception

start

public void start()
           throws SensorHubException

Description copied from interface: IModule

Starts the module with the current configuration.
init() should always be called before start().

Specified by:

start in interface IModule<ConfigType extends StreamProcessConfig>

Throws:

SensorHubException

stop

public void stop()

Description copied from interface: IModule

Stops the module.
All temporary resources created by the module should be cleaned when this is called (ex: memory, files, connections, etc.)
stop() can be called right after init() even if start() hasn't been called.

Specified by:

stop in interface IModule<ConfigType extends StreamProcessConfig>

cleanup

public void cleanup()

Description copied from interface: IModule

Cleans up all ressources used by the module when deleted All persistent resources created by the module should be cleaned when this is called

Specified by:

cleanup in interface IModule<ConfigType extends StreamProcessConfig>

pause

public void pause()

Description copied from interface: IStreamProcessModule

Pauses processing of the event stream.
Incoming events are simply discarded and won't be processed when the process is resumed.

Specified by:

pause in interface IStreamProcessModule<ConfigType extends StreamProcessConfig>

resume

public void resume()

Description copied from interface: IStreamProcessModule

Resumes normal processing of the event stream.
Processing may actually resume only when the next event is received.

Specified by:

resume in interface IStreamProcessModule<ConfigType extends StreamProcessConfig>

handleEvent

public void handleEvent(Event<?> e)

Specified by:

handleEvent in interface IEventListener