#include <ddtDataTransferLib.hpp>

Inheritance diagram for ddt::DdtDataTransferLib:

Public Types
enum class	ConnectionState { Disconnected = 0 , Connected , Reconnecting , PublisherDisconnected }

using	ConnectionStateSignalT = boost::signals2::signal<void(ConnectionState)>

Public Member Functions
	DdtDataTransferLib (DdtLogger *ddt_logger)

	DdtDataTransferLib (log4cplus::Logger const &log4cplus_logger)

virtual	~DdtDataTransferLib ()

void	SetQoS (const int ddt_latency, const int ddt_deadline)

const std::string	VerifyPathInBrokerUri (std::string broker_uri)

int	InitMAL (const std::string broker_uri)

std::unique_ptr< datatransfer::DataBrokerRegistrationSync, std::default_delete< datatransfer::DataBrokerRegistrationSync > >	GetBrokerClient ()

virtual int	RegisterPublisher (const std::string uri, const std::string dsi, const bool compute_crc)

virtual int	UnregisterPublisher ()

virtual void	PublishData ()

virtual int	RegisterSubscriber (const std::string uri, const std::string dsi, const std::string remote_uri, const int32_t interval=10)

virtual int	UnregisterSubscriber ()

virtual DataSample *	ReadData ()

boost::signals2::connection	ConnectToConnectionState (const std::function< void(ConnectionState)> &callback)

Protected Member Functions
void	StartHeartbeat (const int32_t interval, const std::string id)

void	StopHeartbeat ()

virtual void	StopThreads ()

void	Reconnect ()

virtual void	Reregister ()

void	CheckHeartbeatTimeout (int32_t &new_reply_time)

void	NotifyConnectionState (ConnectionState state)

const std::string	GetConfigFilePath ()

void	SpawnReconnectionThread ()

Protected Attributes
int	latency = 0

int	deadline = 0

int32_t	reply_time = 0

int32_t	heartbeat_interval = 0

std::thread	hb_thread

std::thread	reconnect_thread

std::atomic< bool >	reconnect_pending {false}

std::mutex	hb_cv_mutex

std::condition_variable	hb_cv

std::atomic< bool >	heartbeat_active {false}

std::atomic< bool >	shutdown_in_progress {false}

std::unique_ptr< datatransfer::DataBrokerRegistrationSync, std::default_delete< datatransfer::DataBrokerRegistrationSync > >	client

std::atomic< bool >	connected_to_broker {false}

ConnectionStateSignalT	connection_state_signal

std::atomic< ConnectionState >	last_notified_state {ConnectionState::Disconnected}

std::string	broker_uri

elt::mal::rr::ListenerRegistration	connection_listener

DdtLogger *	logger = nullptr

DdtLogger *	my_logger = nullptr

const int32_t	REPLY_TIME_DEFAULT = 6

const int32_t	REPLY_TIME_MIN = 2

int32_t	wait_for_connection = 0

const int32_t	WAIT_FOR_CONNECTION_DEFAULT = 10

const int32_t	WAIT_FOR_CONNECTION_MIN = 2

int32_t	max_consecutive_failures = 0

const int32_t	MAX_CONSECUTIVE_FAILURES_DEFAULT = 3

const int32_t	MAX_CONSECUTIVE_FAILURES_MIN = 1

Detailed Description

Base class for DdtDataPublishers and DdtDataSubscribers.

Member Typedef Documentation

◆ ConnectionStateSignalT

using ddt::DdtDataTransferLib::ConnectionStateSignalT = boost::signals2::signal<void(ConnectionState)>

Signal type to to notify gui about connection state.

Member Enumeration Documentation

◆ ConnectionState

enum class ddt::DdtDataTransferLib::ConnectionState

strong

State of connection to the broker used for notifying gui about current state.

Enumerator
Disconnected	Connection is down.
Connected	Connection is up.
Reconnecting	Reconnection in progress.
PublisherDisconnected	Broker is reachable but the publisher of data stream is not.

Constructor & Destructor Documentation

◆ DdtDataTransferLib() [1/2]

DdtDataTransferLib::DdtDataTransferLib ( DdtLogger * ddt_logger )

explicit

Constructor

Parameters

ddt_logger A DDT logger object (no transfer of ownership).

◆ DdtDataTransferLib() [2/2]

DdtDataTransferLib::DdtDataTransferLib ( log4cplus::Logger const & log4cplus_logger )

explicit

Constructor

Parameters

log4cplus_logger A log4cplus logger object (no transfer of ownership).

◆ ~DdtDataTransferLib()

DdtDataTransferLib::~DdtDataTransferLib ( )

virtual

Destructor

Member Function Documentation

◆ CheckHeartbeatTimeout()

void DdtDataTransferLib::CheckHeartbeatTimeout ( int32_t & new_reply_time )

protected

Verifies that the heartbeat timeout retrieved from the broker is greater than reply time.

Parameters

new_reply_time MAL reply time for the case that the client needs to get recreated.

Make sure the heartbeat timeout is greater than the reply time! If not, recreate MAL client with new reply time.

reset reply_time (the connection manager makes sure that the heartbeat timeout has a valid value (at least 3s)

◆ ConnectToConnectionState()

boost::signals2::connection DdtDataTransferLib::ConnectToConnectionState ( const std::function< void(ConnectionState)> & callback )

Registers an external callback that is notified whenever the connection state to the broker changes.

Parameters

callback Slot invoked on every transition between connection states. The argument describes the new state.

Returns: Boost connection handle.

◆ GetBrokerClient()

std::unique_ptr< datatransfer::DataBrokerRegistrationSync, std::default_delete< datatransfer::DataBrokerRegistrationSync > > DdtDataTransferLib::GetBrokerClient ( )

Retrieve the smart pointer for the client created by the MAL Factory

Returns: Unique pointer for DDT Broker Client

◆ GetConfigFilePath()

const std::string DdtDataTransferLib::GetConfigFilePath ( )

protected

This function reads the environment variable DDT_TRANSFERCONFIG_PATH and returns path and name of the configuration file.

Returns: A string containing the path to the configuration file.

check if DDT_TRANSFERCONFIG_PATH is set return empty string if not set

◆ InitMAL()

int DdtDataTransferLib::InitMAL ( const std::string broker_uri )

Initializes the MAL.

Parameters

broker_uri The URI of the broker.

Returns: 1 if initialization was successfull, -1 otherwise.

determine MAL mapping from uri and load it obtain reference to CiiFactory

Synchronous client example w/ ReplyTime QoS set. DataBrokerRegistrationSync interface is requested.

◆ NotifyConnectionState()

void DdtDataTransferLib::NotifyConnectionState ( ConnectionState state )

protected

Notify subscribers of the connection-state signal that the connection state has changed.

Parameters

state New connection state.

◆ PublishData()

virtual void ddt::DdtDataTransferLib::PublishData ( )

inlinevirtual

Function for triggering the local broker to publish data over the network.

Reimplemented in ddt::DdtDataPublisher.

◆ ReadData()

virtual DataSample * ddt::DdtDataTransferLib::ReadData ( )

inlinevirtual

Function for receiving data.

Returns: The data sample.

Reimplemented in ddt::DdtDataSubscriber.

◆ Reconnect()

void DdtDataTransferLib::Reconnect ( )

protected

Reconnects to the broker after heartbeat failures.

◆ RegisterPublisher()

virtual int ddt::DdtDataTransferLib::RegisterPublisher	(	const std::string	uri,
		const std::string	dsi,
		const bool	compute_crc )

inlinevirtual

Function for registering a publisher at the local broker using a MAL client.

Parameters

uri	The MAL Uniform Resource Identifier.
dsi	The data stream identifier.
compute_crc	Used to switch on / off the CRC computation.

Returns: Success of registration.

Reimplemented in ddt::DdtDataPublisher.

◆ RegisterSubscriber()

virtual int ddt::DdtDataTransferLib::RegisterSubscriber	(	const std::string	uri,
		const std::string	dsi,
		const std::string	remote_uri,
		const int32_t	interval = 10 )

inlinevirtual

Function for registering a subscriber at the local broker using a MAL client.

Parameters

uri	The MAL Uniform Resource Identifier.
dsi	The data stream identifier.
remote_uri	The URI of the remote broker.
interval	The reading interval.

Returns: Success of registration.

Reimplemented in ddt::DdtDataSubscriber.

◆ Reregister()

virtual void ddt::DdtDataTransferLib::Reregister ( )

inlineprotectedvirtual

Reregisters at the broker after a reconnection.

◆ SetQoS()

void DdtDataTransferLib::SetQoS	(	const int	ddt_latency,
		const int	ddt_deadline )

Sets quality of service parameters.

Parameters

ddt_latency	A MAL QoS parameter. Shall be the maximum time a sample may remain in-transit between the publisher and subscriber in [ms].
ddt_deadline	A MAL QoS parameter. Shall be the maximum age of a sample in [s].

◆ SpawnReconnectionThread()

void DdtDataTransferLib::SpawnReconnectionThread ( )

protected

Spawn Reconnect() on a fresh thread if no reconnect is already in progress.

◆ StartHeartbeat()

void DdtDataTransferLib::StartHeartbeat	(	const int32_t	interval,
		const std::string	id )

protected

Starts the heartbeat thread.

Parameters

interval	The heartbeat interval in [s].
id	The data stream identifier (in case of a publisher) or the UUID (in case of a subscriber).

◆ StopHeartbeat()

void DdtDataTransferLib::StopHeartbeat ( )

protected

Stops the heartbeat thread.

◆ StopThreads()

void DdtDataTransferLib::StopThreads ( )

protectedvirtual

Stops all threads and joins them.

Reimplemented in ddt::DdtDataSubscriber.

◆ UnregisterPublisher()

virtual int ddt::DdtDataTransferLib::UnregisterPublisher ( )

inlinevirtual

Function for unregistering a publisher from the local broker.

Returns: Success of unregistration.

Reimplemented in ddt::DdtDataPublisher.

◆ UnregisterSubscriber()

virtual int ddt::DdtDataTransferLib::UnregisterSubscriber ( )

inlinevirtual

Function for unregistering a subscriber from the local broker.

Returns: Success of unregistration.

Reimplemented in ddt::DdtDataSubscriber.

◆ VerifyPathInBrokerUri()

const std::string DdtDataTransferLib::VerifyPathInBrokerUri ( std::string broker_uri )

This function verifies if the broker uri contains the path element. If not specified by the user the path element is added.

Parameters

broker_uri The URI of the broker.

Returns: The URI of the broker including a path element that is required by MAL.

Member Data Documentation

◆ broker_uri

std::string ddt::DdtDataTransferLib::broker_uri

protected

URI of the broker.

◆ client

std::unique_ptr< datatransfer::DataBrokerRegistrationSync, std::default_delete<datatransfer::DataBrokerRegistrationSync> > ddt::DdtDataTransferLib::client

protected

MAL client used to connect to data brokers.

◆ connected_to_broker

std::atomic<bool> ddt::DdtDataTransferLib::connected_to_broker {false}

protected

Stores connection state to broker.

◆ connection_listener

elt::mal::rr::ListenerRegistration ddt::DdtDataTransferLib::connection_listener

protected

Observes the connection to the broker.

◆ connection_state_signal

ConnectionStateSignalT ddt::DdtDataTransferLib::connection_state_signal

protected

Signal fired when the connection state changes.

◆ deadline

int ddt::DdtDataTransferLib::deadline = 0

protected

Maximum age of a sample [s].

◆ hb_cv

std::condition_variable ddt::DdtDataTransferLib::hb_cv

protected

◆ hb_cv_mutex

std::mutex ddt::DdtDataTransferLib::hb_cv_mutex

protected

Mutex and condition variable used to allow immediate wakeup of the heartbeat sleep when stopping.

◆ hb_thread

std::thread ddt::DdtDataTransferLib::hb_thread

protected

Heartbeat thread.

◆ heartbeat_active

std::atomic<bool> ddt::DdtDataTransferLib::heartbeat_active {false}

protected

When the HearbeatThread is stopped it waits until heartbeat_active is set to false.

◆ heartbeat_interval

int32_t ddt::DdtDataTransferLib::heartbeat_interval = 0

protected

Interval for the heartbeat in [s] retrieved from the data broker.

◆ last_notified_state

std::atomic<ConnectionState> ddt::DdtDataTransferLib::last_notified_state {ConnectionState::Disconnected}

protected

Last connection state that was broadcast via the signal.

◆ latency

int ddt::DdtDataTransferLib::latency = 0

protected

Duration between send and receive cannot be greater than 'latency' [ms].

◆ logger

DdtLogger* ddt::DdtDataTransferLib::logger = nullptr

protected

A DDT logger object.

◆ max_consecutive_failures

int32_t ddt::DdtDataTransferLib::max_consecutive_failures = 0

protected

Maximum consecutive heartbeat failures before stopping.

◆ MAX_CONSECUTIVE_FAILURES_DEFAULT

const int32_t ddt::DdtDataTransferLib::MAX_CONSECUTIVE_FAILURES_DEFAULT = 3

protected

Default value for max_consecutive_failures.

◆ MAX_CONSECUTIVE_FAILURES_MIN

const int32_t ddt::DdtDataTransferLib::MAX_CONSECUTIVE_FAILURES_MIN = 1

protected

Minimum value for max_consecutive_failures.

◆ my_logger

DdtLogger* ddt::DdtDataTransferLib::my_logger = nullptr

protected

A DDT logger object.

◆ reconnect_pending

std::atomic<bool> ddt::DdtDataTransferLib::reconnect_pending {false}

protected

Guard flag preventing concurrent reconnect attempts.

◆ reconnect_thread

std::thread ddt::DdtDataTransferLib::reconnect_thread

protected

Reconnect thread. Spawned by HeartbeatThread when consecutive heartbeat failures exceed the threshold, or by the MAL connection listener when the broker connection is reported down.

◆ reply_time

int32_t ddt::DdtDataTransferLib::reply_time = 0

protected

Configurable reply time for MAL clients in [s].

◆ REPLY_TIME_DEFAULT

const int32_t ddt::DdtDataTransferLib::REPLY_TIME_DEFAULT = 6

protected

Default value for the reply time.

◆ REPLY_TIME_MIN

const int32_t ddt::DdtDataTransferLib::REPLY_TIME_MIN = 2

protected

Minimum value for the reply time.

◆ shutdown_in_progress

std::atomic<bool> ddt::DdtDataTransferLib::shutdown_in_progress {false}

protected

Information for threads to abort operation if shutdown is pending.

◆ wait_for_connection

int32_t ddt::DdtDataTransferLib::wait_for_connection = 0

protected

Time to wait for broker connection in [s].

◆ WAIT_FOR_CONNECTION_DEFAULT

const int32_t ddt::DdtDataTransferLib::WAIT_FOR_CONNECTION_DEFAULT = 10

protected

Default value for the wait_for_connection timeout.

◆ WAIT_FOR_CONNECTION_MIN

const int32_t ddt::DdtDataTransferLib::WAIT_FOR_CONNECTION_MIN = 2

protected

Minimum value for the wait_for_connection timeout.

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

datatransfer/datatransferlib/src/include/ddt/ddtDataTransferLib.hpp
datatransfer/datatransferlib/src/ddtDataTransferLib.cpp

Public Types

Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Member Typedef Documentation

◆ ConnectionStateSignalT

Member Enumeration Documentation

◆ ConnectionState

Constructor & Destructor Documentation

◆ DdtDataTransferLib() [1/2]

◆ DdtDataTransferLib() [2/2]

◆ ~DdtDataTransferLib()

Member Function Documentation

◆ CheckHeartbeatTimeout()

◆ ConnectToConnectionState()

◆ GetBrokerClient()

◆ GetConfigFilePath()

◆ InitMAL()

◆ NotifyConnectionState()

◆ PublishData()

◆ ReadData()

◆ Reconnect()

◆ RegisterPublisher()

◆ RegisterSubscriber()

◆ Reregister()

◆ SetQoS()

◆ SpawnReconnectionThread()

◆ StartHeartbeat()

◆ StopHeartbeat()

◆ StopThreads()

◆ UnregisterPublisher()

◆ UnregisterSubscriber()

◆ VerifyPathInBrokerUri()

Member Data Documentation

◆ broker_uri

◆ client

◆ connected_to_broker

◆ connection_listener

◆ connection_state_signal

◆ deadline

◆ hb_cv

◆ hb_cv_mutex

◆ hb_thread

◆ heartbeat_active

◆ heartbeat_interval

◆ last_notified_state

◆ latency

◆ logger

◆ max_consecutive_failures

◆ MAX_CONSECUTIVE_FAILURES_DEFAULT

◆ MAX_CONSECUTIVE_FAILURES_MIN

◆ my_logger

◆ reconnect_pending

◆ reconnect_thread

◆ reply_time

◆ REPLY_TIME_DEFAULT

◆ REPLY_TIME_MIN

◆ shutdown_in_progress

◆ wait_for_connection

◆ WAIT_FOR_CONNECTION_DEFAULT

◆ WAIT_FOR_CONNECTION_MIN