#include <StreamAligner.hpp>

Inheritance diagram for aggregator::StreamAligner:

Classes
class	Stream

Public Types
typedef std::vector< StreamBase * >	stream_vector

Public Member Functions
	StreamAligner (base::Time timeout=base::Time::fromSeconds(1))

virtual	~StreamAligner ()

void	copyState (const StreamAligner &other)

void	setTimeout (const base::Time &t)

void	disableStream (int idx)

void	enableStream (int idx)

bool	isStreamActive (int idx) const

void	unregisterStream (int idx)

template<class T >
int	registerStream (typename Stream< T >::callback_t callback, int bufferSize, base::Time period, int priority=-1, const std::string &name=std::string())

template<class T >
void	push (int idx, const base::Time &ts, const T &data)
	Push new data into the stream. More...

template<class T >
bool	getNextSample (int idx, std::pair< base::Time, T > &sample) const

bool	step ()

void	clear ()

base::Time	getTimeOut () const

base::Time	getLatency () const

base::Time	getCurrentTime () const

base::Time	getLatestTime () const

int	getStreamSize () const

const StreamStatus &	getBufferStatus (int idx) const

const StreamAlignerStatus &	getStatus () const

Static Public Member Functions
static bool	compareStreams (const StreamBase b1, const StreamBase b2)

Public Attributes
stream_vector	streams

base::Time	timeout

base::Time	latest_ts

base::Time	current_ts

double	buffer_size_factor

StreamAlignerStatus	status

Friends
std::ostream &	operator<< (std::ostream &stream, const aggregator::StreamAligner::StreamBase &base)

std::ostream &	operator<< (std::ostream &stream, const aggregator::StreamAligner &re)

Member Typedef Documentation

◆ stream_vector

typedef std::vector<StreamBase*> aggregator::StreamAligner::stream_vector

Constructor & Destructor Documentation

◆ StreamAligner()

aggregator::StreamAligner::StreamAligner ( base::Time timeout = base::Time::fromSeconds(1) )

inlineexplicit

◆ ~StreamAligner()

virtual aggregator::StreamAligner::~StreamAligner ( )

inlinevirtual

Member Function Documentation

◆ clear()

void aggregator::StreamAligner::clear ( )

inline

clears all samples in all streams, resets the statistics and resets the playback times but leaves the stream setup intact.

◆ compareStreams()

static bool aggregator::StreamAligner::compareStreams	(	const StreamBase *	b1,
		const StreamBase *	b2
	)

inlinestatic

◆ copyState()

void aggregator::StreamAligner::copyState ( const StreamAligner & other )

inline

will take the state of other StreamAligner and make it the state of this object. State constitutes current_time and latest_time as well as all the stream content, but not the configuration.

◆ disableStream()

void aggregator::StreamAligner::disableStream ( int idx )

inline

Will disable the stream with the given index.

All data left in the stream will still be played out, however the stream will be ignored for lookahead and timeout calculation. A stream, which is disabled can be enabled through the enableStream() call, or if new data in this stream arrives.

The functionality is needed for cases where streams might be optional, so that the other streams won't be delayed up to the maximum time out value.

◆ enableStream()

void aggregator::StreamAligner::enableStream ( int idx )

inline

Enables a stream which has been disabled previously.

All streams are enabled by default. Does not have any effect on streams which are already enabled.

◆ getBufferStatus()

const StreamStatus& aggregator::StreamAligner::getBufferStatus ( int idx ) const

inline

return the buffer status as a std::pair. first element in pair is the current buffer fill and the second element is the buffer size

◆ getCurrentTime()

base::Time aggregator::StreamAligner::getCurrentTime ( ) const

inline

return the time of the last data item that went out

◆ getLatency()

base::Time aggregator::StreamAligner::getLatency ( ) const

inline

latency is the time difference between the latest data item that has come in, and the latest data item that went out

◆ getLatestTime()

base::Time aggregator::StreamAligner::getLatestTime ( ) const

inline

return the time of the last data item that came in

◆ getNextSample()

template<class T >

bool aggregator::StreamAligner::getNextSample	(	int	idx,
		std::pair< base::Time, T > &	sample
	)		const

inline

◆ getStatus()

const StreamAlignerStatus& aggregator::StreamAligner::getStatus ( ) const

inline

Returns: the current status of the StreamAligner this is mainly used for debug purposes

◆ getStreamSize()

int aggregator::StreamAligner::getStreamSize ( ) const

inline

return the number of streams

◆ getTimeOut()

base::Time aggregator::StreamAligner::getTimeOut ( ) const

inline

Get the time the Estimator will wait for an expected reading on any of the streams. This number effectively puts an upper limit to the lag that can be created due to delay or missing values on the channels.

◆ isStreamActive()

bool aggregator::StreamAligner::isStreamActive ( int idx ) const

inline

See if a stream is enabled or disabled.

Returns: true if the stream is enabled (active)

◆ push()

template<class T >

void aggregator::StreamAligner::push	(	int	idx,
		const base::Time &	ts,
		const T &	data
	)

inline

Push new data into the stream.

Note that if the stream was previously inactive, this call will make it active implicetely.

Parameters

ts	- the timestamp of the data item
data	- the data added to the stream

◆ registerStream()

template<class T >

int aggregator::StreamAligner::registerStream	(	typename Stream< T >::callback_t	callback,
		int	bufferSize,
		base::Time	period,
		int	priority = `-1`,
		const std::string &	name = `std::string()`
	)

inline

Will register a stream with the aggregator.

Parameters

callback	- will be called for data gone through the synchronization process
period	- time between sensor readings. This will be used to estimate when the next reading should arrive, so out of order arrivals are possible. Set to 0 if not a periodic stream. When set to a negative value, the calculation of the buffer is performed for that period, however no lookahead is set.
bufferSize	- The size of the internal FIFO buffer. This should be at least the amount of samples that can occur in a timeout period. If no value is provided, the bufferSize is calculated from the period and timeout values provided, with an additional safety factor.
priority	- if streams have data with equal timestamps, the one with the lower priority value will be pushed first.
name	- name of the stream. This is only for debug purposes

Returns: - stream index, which is used to identify the stream (e.g. for push).

◆ setTimeout()

void aggregator::StreamAligner::setTimeout ( const base::Time & t )

inline

Set the time the Estimator will wait for an expected reading on any of the streams. This number effectively puts an upper limit to the lag that can be created due to delay or missing values on the channels.

◆ step()

bool aggregator::StreamAligner::step ( )

inline

This will go through the available streams and look for the oldest available data. The data can be either existing are predicted through the period.

There are three different cases that can happen:

The data is already available. In this case that data is forwarded to the callback.
The data is not yet available, and the time difference between oldest data and newest data is below the timeout threshold. In this case no data is called.
The data is not yet available, and the timeout is reached. In this case, the oldest data (which is obviously non-available) is ignored, and only newer data is considered.

Returns: - true if a callback was called and more data might be available

◆ unregisterStream()

void aggregator::StreamAligner::unregisterStream ( int idx )

inline

This function will remove the stream with the given index from the stream aligner.

Parameters

idx	index of the stream that should be unregistered

Friends And Related Function Documentation

◆ operator<< [1/2]

std::ostream& operator<<	(	std::ostream &	stream,
		const aggregator::StreamAligner::StreamBase &	base
	)

friend

◆ operator<< [2/2]

std::ostream& operator<<	(	std::ostream &	stream,
		const aggregator::StreamAligner &	re
	)

friend

Member Data Documentation

◆ buffer_size_factor

double aggregator::StreamAligner::buffer_size_factor

◆ current_ts

base::Time aggregator::StreamAligner::current_ts

time of the last sample that went out

◆ latest_ts

base::Time aggregator::StreamAligner::latest_ts

time of the last sample that came in

◆ status

StreamAlignerStatus aggregator::StreamAligner::status

mutable

temporary object that gets returned by getStatus, in order to avoid dynamic allocation on each call

◆ streams

stream_vector aggregator::StreamAligner::streams

◆ timeout

base::Time aggregator::StreamAligner::timeout

The documentation for this class was generated from the following file:

/build/rock-master-18.09-drivers-aggregator-0.20160914/src/StreamAligner.hpp

Classes

Public Types

Public Member Functions

Static Public Member Functions

Public Attributes

Friends

Member Typedef Documentation

◆ stream_vector

Constructor & Destructor Documentation

◆ StreamAligner()

◆ ~StreamAligner()

Member Function Documentation

◆ clear()

◆ compareStreams()

◆ copyState()

◆ disableStream()

◆ enableStream()

◆ getBufferStatus()

◆ getCurrentTime()

◆ getLatency()

◆ getLatestTime()

◆ getNextSample()

◆ getStatus()

◆ getStreamSize()

◆ getTimeOut()

◆ isStreamActive()

◆ push()

◆ registerStream()

◆ setTimeout()

◆ step()

◆ unregisterStream()

Friends And Related Function Documentation

◆ operator<< [1/2]

◆ operator<< [2/2]

Member Data Documentation

◆ buffer_size_factor

◆ current_ts

◆ latest_ts

◆ status

◆ streams

◆ timeout