relayer Class Reference

Asynchronous message sending. More...

#include <relayer.hh>

Inheritance diagram for relayer:

Classes
struct	message_and_reply
struct	wait_for_send_pkg
	Extended sent-message package that adds thread synchronization objects. More...

Public Types
typedef std::shared_ptr< wait_for_send_pkg >	wait_for_send_pkg_ptr
Public Types inherited from core
enum class	post_listen_status {   unknown , message_received , timeout , soft_error ,   hard_error }

Public Member Functions
	relayer (const scarab::param_node &a_config, const scarab::authentication &a_auth)
	relayer (const relayer &)=delete
	relayer (relayer &&)=default
virtual	~relayer ()=default
relayer &	operator= (const relayer &)=delete
relayer &	operator= (relayer &&)
void	execute_relayer ()
	Thread execution function: sends any messages that are submitted via the send functions. More...
wait_for_send_pkg_ptr	send_async (request_ptr_t a_request) const
wait_for_send_pkg_ptr	send_async (alert_ptr_t a_alert) const
reply_ptr_t	wait_for_reply (const wait_for_send_pkg_ptr a_receive_reply, int a_timeout_ms=0)
reply_ptr_t	wait_for_reply (const wait_for_send_pkg_ptr a_receive_reply, core::post_listen_status &a_status, int a_timeout_ms=0)
Public Member Functions inherited from core
	core (const scarab::param_node &a_config=dripline_config(), const scarab::authentication &a_auth=scarab::authentication(), const bool a_make_connection=true)
	core (const core &a_orig)=default
	core (core &&a_orig)=default
virtual	~core ()=default
core &	operator= (const core &a_orig)=default
core &	operator= (core &&a_orig)=default
virtual sent_msg_pkg_ptr	send (request_ptr_t a_request, amqp_channel_ptr a_channel=amqp_channel_ptr()) const
virtual sent_msg_pkg_ptr	send (reply_ptr_t a_reply, amqp_channel_ptr a_channel=amqp_channel_ptr()) const
virtual sent_msg_pkg_ptr	send (alert_ptr_t a_alert, amqp_channel_ptr a_channel=amqp_channel_ptr()) const
	snake_case_mv_referrable (std::string, address)
	snake_case_mv_accessible (unsigned, port)
	snake_case_mv_referrable (std::string, username)
	snake_case_mv_referrable (std::string, password)
	snake_case_mv_referrable (std::string, requests_exchange)
	snake_case_mv_referrable (std::string, alerts_exchange)
	snake_case_mv_referrable (std::string, heartbeat_routing_key)
	snake_case_mv_accessible (unsigned, max_payload_size)
	snake_case_mv_accessible (bool, make_connection)
	snake_case_mv_accessible (unsigned, max_connection_attempts)

Private Types
typedef std::shared_ptr< message_and_reply >	mar_ptr

Private Member Functions
void	do_cancellation (int a_code)
	snake_case_mv_referrable (receiver, msg_receiver)

Private Attributes
scarab::concurrent_queue< mar_ptr >	f_queue

Additional Inherited Members
Static Public Member Functions inherited from core
static void	listen_for_message (amqp_envelope_ptr &a_envelope, post_listen_status &a_status, amqp_channel_ptr a_channel, const std::string &a_consumer_tag, int a_timeout_ms=0, bool a_do_ack=true)
	listen for a single AMQP message More...
Static Public Attributes inherited from core
static bool	s_offline = false
Protected Member Functions inherited from core
sent_msg_pkg_ptr	do_send (message_ptr_t a_message, const std::string &a_exchange, bool a_expect_reply, amqp_channel_ptr a_channel=amqp_channel_ptr()) const
amqp_channel_ptr	send_withreply (message_ptr_t a_message, std::string &a_reply_consumer_tag, const std::string &a_exchange) const
bool	send_noreply (message_ptr_t a_message, const std::string &a_exchange) const
amqp_channel_ptr	open_channel () const
Static Protected Member Functions inherited from core
static bool	setup_exchange (amqp_channel_ptr a_channel, const std::string &a_exchange)
static bool	setup_queue (amqp_channel_ptr a_channel, const std::string &a_queue_name)
static bool	bind_key (amqp_channel_ptr a_channel, const std::string &a_exchange, const std::string &a_queue_name, const std::string &a_routing_key)
static std::string	start_consuming (amqp_channel_ptr a_channel, const std::string &a_queue_name)
static bool	stop_consuming (amqp_channel_ptr a_channel, std::string &a_consumer_tag)
static bool	remove_queue (amqp_channel_ptr a_channel, const std::string &a_queue_name)

Detailed Description

Asynchronous message sending.

Author

N.S. Oblath

This class allows a user to send message asynchronously. Messages are sent in a dedicated thread that runs the function execute_relayer(). Asynchronous operation is achieved using a concurrent queue to store messages that are submitted.

The primary user interface consists of send_async() and wait_for_reply(). The former immediately returns a special sent-message package that includes thread-synchonization objects. That package is then passed to the latter, which blocks while waiting for a reply.

Definition at line 37 of file relayer.hh.

Member Typedef Documentation

mar_ptr

typedef std::shared_ptr< message_and_reply > mar_ptr

private

Definition at line 107 of file relayer.hh.

wait_for_send_pkg_ptr

typedef std::shared_ptr< wait_for_send_pkg > wait_for_send_pkg_ptr

Definition at line 72 of file relayer.hh.

Constructor & Destructor Documentation

relayer() [1/3]

relayer	(	const scarab::param_node &	a_config,
		const scarab::authentication &	a_auth
	)

Definition at line 16 of file relayer.cc.

relayer() [2/3]

relayer ( const relayer &  )

delete

relayer() [3/3]

relayer ( relayer &&  )

default

~relayer()

virtual ~relayer ( )

virtualdefault

Member Function Documentation

do_cancellation()

void do_cancellation ( int  a_code )

private

Definition at line 86 of file relayer.cc.

execute_relayer()

void execute_relayer

(

)

Thread execution function: sends any messages that are submitted via the send functions.

Definition at line 34 of file relayer.cc.

operator=() [1/2]

relayer& operator= ( const relayer &  )

delete

operator=() [2/2]

relayer & operator=

(

relayer &&

a_orig

)

Definition at line 23 of file relayer.cc.

send_async() [1/2]

relayer::wait_for_send_pkg_ptr send_async

(

alert_ptr_t

a_alert

)

const

Asynchronously send an alert message Returns immediately, without blocking for send

Definition at line 112 of file relayer.cc.

send_async() [2/2]

relayer::wait_for_send_pkg_ptr send_async

(

request_ptr_t

a_request

)

const

Asynchronously send a request message Returns immediately, without blocking for send

Definition at line 93 of file relayer.cc.

snake_case_mv_referrable()

snake_case_mv_referrable ( receiver  , msg_receiver    )

private

wait_for_reply() [1/2]

reply_ptr_t wait_for_reply	(	const wait_for_send_pkg_ptr	a_receive_reply,
		core::post_listen_status &	a_status,
		int	a_timeout_ms = 0
	)

User interface for waiting for a reply message. This can be called multiple times to receive multiple replies.

Parameters

[in]	a_receive_reply	The sent-message package from the request.
[in]	a_timeout_ms	Timeout for waiting for a reply; if it's 0, there will be no timeout.
[out]	a_status	Returns the output status after receiving a message (or failing to do so).

Returns

Reply message

Definition at line 137 of file relayer.cc.

wait_for_reply() [2/2]

reply_ptr_t wait_for_reply	(	const wait_for_send_pkg_ptr	a_receive_reply,
		int	a_timeout_ms = 0
	)

User interface for waiting for a reply message. This can be called multiple times to receive multiple replies.

Parameters

a_receive_reply	The sent-message package from the request.
a_timeout_ms	Timeout for waiting for a reply; if it's 0, there will be no timeout.

Returns

Reply message

Definition at line 131 of file relayer.cc.

Member Data Documentation

f_queue

scarab::concurrent_queue< mar_ptr > f_queue

mutableprivate

Definition at line 109 of file relayer.hh.

---

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

• relayer.hh
• relayer.cc