#include <ddtDataConsumer.hpp>

Inheritance diagram for ddt::DdtDataConsumer:

Public Types
enum	SubscriberType { LOCAL , REMOTE }

Public Types inherited from ddt::DdtProducerConsumerBase
enum	NotificationType { NEW_DATA , PUB_REG , PUB_UNREG , SHM_DEL }

Public Attributes
signal_n	notification_signal

Protected Attributes
DdtMemoryAccessor *	memory_accessor

std::string	data_stream_identifier

int32_t	number_of_samples

int32_t	notification_port

std::string	publishing_uri

std::chrono::system_clock::time_point	last_received

uint64_t	total_samples = 0

uint64_t	total_bytes = 0

uint64_t	total_latency = 0

Protected Attributes inherited from ddt::DdtProducerConsumerBase
int32_t	min_port

int32_t	max_port

DdtLogger *	logger

const int	MIN_PORT_DEFAULT = 5100

const int	MAX_PORT_DEFAULT = 5500

const int	MIN_RANGE = 3

const int	LINE_LENGTH = 256

	DdtDataConsumer (const std::string &data_stream_identifier, const int32_t latency, const int32_t deadline, DdtLogger *ddt_logger)

	DdtDataConsumer (const std::string &data_stream_identifier, const std::string &subscription_uri, const int32_t latency, const int32_t deadline, DdtLogger *ddt_logger)

	~DdtDataConsumer () override

void	StartSubscription ()

void	StopSubscription ()

void	AddUuid (std::string uuid, SubscriberType type)

void	RemoveUuid (const std::string uuid)

void	Notify (const NotificationType type) override

int32_t	get_number_of_subscribers ()

int32_t	get_number_of_remote_subscribers ()

std::map< std::string, SubscriberType >	get_subscribers ()

std::string	get_remote_broker_uri () const

int32_t	get_notification_port () const

DdtStatistics	get_statistics ()

void	ResetStatistics ()

std::string	get_publishing_uri () const

void	set_remote_broker_uri (const std::string &remote_uri)

void	set_number_of_samples (const int32_t num_samples)

void	set_memory_accessor (DdtMemoryAccessor *mem_accessor)

void	set_notification_port (const int32_t noti_port)

void	set_publishing_uri (const std::string pub_uri)

void	set_originating_broker (const std::string orig_broker)

void	Init (const std::string &ds_id, DdtLogger *ddt_logger)

Additional Inherited Members
Public Member Functions inherited from ddt::DdtProducerConsumerBase
	DdtProducerConsumerBase (DdtLogger *ddt_logger)

virtual	~DdtProducerConsumerBase ()

Protected Member Functions inherited from ddt::DdtProducerConsumerBase
void	Init (DdtLogger *ddt_logger)

void	LoadDefaults ()

void	ReadIni ()

void	PrintConfigValues ()

int32_t	GetFreePort () const

Detailed Description

This class provides the functionality to subscribe to a data stream, to write the received data into shared memory and to notify DdtDataSubscribers that new data is available. There will be one data consumer object per data stream identifier and thus several DdtDataSubscribers may share one consumer object.

Member Enumeration Documentation

◆ SubscriberType

enum ddt::DdtDataConsumer::SubscriberType

To distinguish between local and remote subscribers.

Enumerator
LOCAL	Local subscriber
REMOTE	Remote subscriber

Constructor & Destructor Documentation

◆ DdtDataConsumer() [1/2]

DdtDataConsumer::DdtDataConsumer	(	const std::string &	data_stream_identifier,
		const int32_t	latency,
		const int32_t	deadline,
		DdtLogger *	ddt_logger )

Constructor

Parameters

data_stream_identifier	The data stream identifier.
latency	A MAL QoS parameter. Shall be the maximum time a sample may remain in-transit between the publisher and subscriber in [ms].
deadline	A MAL QoS parameter. Shall be the maximum age of a sample in [s].
ddt_logger	A DDT logger object (no transfer of ownership).

◆ DdtDataConsumer() [2/2]

DdtDataConsumer::DdtDataConsumer	(	const std::string &	data_stream_identifier,
		const std::string &	subscription_uri,
		const int32_t	latency,
		const int32_t	deadline,
		DdtLogger *	ddt_logger )

Constructor

Parameters

data_stream_identifier	The data stream identifier.
subscription_uri	URI for the subscription.
latency	A MAL QoS parameter. Shall be the maximum time a sample may remain in-transit between the publisher and subscriber in [ms].
deadline	A MAL QoS parameter. Shall be the maximum age of a sample in [s].
ddt_logger	A DDT logger object (no transfer of ownership).

◆ ~DdtDataConsumer()

DdtDataConsumer::~DdtDataConsumer ( )

overridedefault

Destructor

Member Function Documentation

◆ AddUuid()

void DdtDataConsumer::AddUuid	(	std::string	uuid,
		SubscriberType	type )

Adds the UUID of a DdtDataSubscriber to the subscriber_map.

Parameters

uuid	UUID of the subscriber.
type	Type of the subscriber.

◆ get_notification_port()

int32_t DdtDataConsumer::get_notification_port ( ) const

Returns notification_port.

Returns: The port used for notifications.

◆ get_number_of_remote_subscribers()

int32_t DdtDataConsumer::get_number_of_remote_subscribers ( )

Returns the number of remote DdtDataSubscribers attached to a consumer object.

Returns: Number of all remote DdtDataSubscribers.

iterate over the subscriber map count all subscribers of type REMOTE

◆ get_number_of_subscribers()

int32_t DdtDataConsumer::get_number_of_subscribers ( )

Returns the number of all DdtDataSubscribers attached to a consumer object.

Returns: Number of all DdtDataSubscribers.

◆ get_publishing_uri()

std::string DdtDataConsumer::get_publishing_uri ( ) const

Returns publishing_uri.

Returns: The publishing URI.

◆ get_remote_broker_uri()

std::string DdtDataConsumer::get_remote_broker_uri ( ) const

Returns remote_broker_uri.

Returns: The URI of the remote broker.

◆ get_statistics()

DdtStatistics DdtDataConsumer::get_statistics ( )

Returns the statistics.

Returns: A DdtStatistics object.

◆ get_subscribers()

std::map< std::string, DdtDataConsumer::SubscriberType > DdtDataConsumer::get_subscribers ( )

Returns the map with the uuids of DdtDataSubscribers attached to a consumer object.

Returns: A map with the UUIDs and the subscriber types.

◆ Init()

void DdtDataConsumer::Init	(	const std::string &	ds_id,
		DdtLogger *	ddt_logger )

protected

Initializes some member variables.

Parameters

ds_id	The data stream identifier.
ddt_logger	A DDT logger object (no transfer of ownership).

◆ Notify()

void DdtDataConsumer::Notify ( const NotificationType type )

overridevirtual

Sends a notification to local subscribers.

Parameters

type	Type of the subscriber.

Implements ddt::DdtProducerConsumerBase.

◆ RemoveUuid()

void DdtDataConsumer::RemoveUuid ( const std::string uuid )

Removes the UUID of a DdtDataSubscriber from the subscriber_map.

Parameters

uuid	UUID of the subscriber.

◆ ResetStatistics()

void DdtDataConsumer::ResetStatistics ( )

Resets the statistics to zero in case the shared memory was deleted.

◆ set_memory_accessor()

void DdtDataConsumer::set_memory_accessor ( DdtMemoryAccessor * mem_accessor )

Sets memory_accessor.

Parameters

mem_accessor A DDT memory accessor object (no transfer of ownership).

◆ set_notification_port()

void DdtDataConsumer::set_notification_port ( const int32_t noti_port )

Sets notification_port.

Parameters

noti_port The port used for notifications.

◆ set_number_of_samples()

void DdtDataConsumer::set_number_of_samples ( const int32_t num_samples )

Sets number_of_samples.

Parameters

num_samples Number of samples stored in the SHM ring buffer.

◆ set_originating_broker()

void DdtDataConsumer::set_originating_broker ( const std::string orig_broker )

Sets the originating broker_uri.

Parameters

orig_broker The originating broker URI.

◆ set_publishing_uri()

void DdtDataConsumer::set_publishing_uri ( const std::string pub_uri )

Sets the publishing_uri.

Parameters

pub_uri The publishing URI.

◆ set_remote_broker_uri()

void DdtDataConsumer::set_remote_broker_uri ( const std::string & remote_uri )

Sets remote_broker_uri.

Parameters

remote_uri THe URI of the remote broker.

◆ StartSubscription()

void DdtDataConsumer::StartSubscription ( )

Starts MAL subscription as a separate thread.

◆ StopSubscription()

void DdtDataConsumer::StopSubscription ( )

Stops MAL subscription.

Member Data Documentation

◆ data_stream_identifier

std::string ddt::DdtDataConsumer::data_stream_identifier

protected

The data stream identifier.

◆ last_received

std::chrono::system_clock::time_point ddt::DdtDataConsumer::last_received

protected

Last time when a data sample was received.

◆ memory_accessor

DdtMemoryAccessor* ddt::DdtDataConsumer::memory_accessor

protected

The DDT memory accessor object.

◆ notification_port

int32_t ddt::DdtDataConsumer::notification_port

protected

The port used for notifications.

◆ notification_signal

signal_n ddt::DdtDataConsumer::notification_signal

Boost signal to send the notification type received via MALps.

◆ number_of_samples

int32_t ddt::DdtDataConsumer::number_of_samples

protected

The Number of samples stored in the SHM ring buffer.

◆ publishing_uri

std::string ddt::DdtDataConsumer::publishing_uri

protected

The publishing URI.

◆ total_bytes

uint64_t ddt::DdtDataConsumer::total_bytes = 0

protected

Total number of bytes received in the data stream.

◆ total_latency

uint64_t ddt::DdtDataConsumer::total_latency = 0

protected

Aggregated latency in [ms].

◆ total_samples

uint64_t ddt::DdtDataConsumer::total_samples = 0

protected

Total number of complete data samples received.

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

datatransfer/databrokerlib/src/include/ddt/ddtDataConsumer.hpp
datatransfer/databrokerlib/src/ddtDataConsumer.cpp

Public Types

Public Attributes

Protected Attributes

Additional Inherited Members

Detailed Description

Member Enumeration Documentation

◆ SubscriberType

Constructor & Destructor Documentation

◆ DdtDataConsumer() [1/2]

◆ DdtDataConsumer() [2/2]

◆ ~DdtDataConsumer()

Member Function Documentation

◆ AddUuid()

◆ get_notification_port()

◆ get_number_of_remote_subscribers()

◆ get_number_of_subscribers()

◆ get_publishing_uri()

◆ get_remote_broker_uri()

◆ get_statistics()

◆ get_subscribers()

◆ Init()

◆ Notify()

◆ RemoveUuid()

◆ ResetStatistics()

◆ set_memory_accessor()

◆ set_notification_port()

◆ set_number_of_samples()

◆ set_originating_broker()

◆ set_publishing_uri()

◆ set_remote_broker_uri()

◆ StartSubscription()

◆ StopSubscription()

Member Data Documentation

◆ data_stream_identifier

◆ last_received

◆ memory_accessor

◆ notification_port

◆ notification_signal

◆ number_of_samples

◆ publishing_uri

◆ total_bytes

◆ total_latency

◆ total_samples