Service class aimed at adding a Dripline API to an existing codebase. More...

#include <hub.hh>

Inheritance diagram for hub:

Public Member Functions
	hub (const scarab::param_node &a_config=scarab::param_node(), const std::string &a_queue_name="", const std::string &a_broker_address="", unsigned a_port=0, const std::string &a_auth_file="", const bool a_make_connection=true)

virtual	~hub ()

void	set_run_handler (const handler_func_t &a_func)
	Sets the run request handler function. More...

void	register_get_handler (const std::string &a_key, const handler_func_t &a_func)
	Sets a get request handler function. More...

void	register_set_handler (const std::string &a_key, const handler_func_t &a_func)
	Sets a set request handler function. More...

void	register_cmd_handler (const std::string &a_key, const handler_func_t &a_func)
	Sets a command request handler function. More...

void	remove_get_handler (const std::string &a_key)
	Removes a get request handler function. More...

void	remove_set_handler (const std::string &a_key)
	Removes a set request handler function. More...

void	remove_cmd_handler (const std::string &a_key)
	Removes a command request handler function. More...

Public Member Functions inherited from service
	service (const scarab::param_node &a_config=scarab::param_node(), const std::string &a_queue_name="", const std::string &a_broker_address="", unsigned a_port=0, const std::string &a_auth_file="", const bool a_make_connection=true)

	service (const bool a_make_connection, const scarab::param_node &a_config=scarab::param_node())

	service (const service &)=delete

	service (service &&a_orig)

virtual	~service ()

service &	operator= (const service &)=delete

service &	operator= (service &&a_orig)

	snake_case_mv_accessible (status, status)

	snake_case_mv_accessible (bool, enable_scheduling)

bool	add_child (endpoint_ptr_t a_endpoint_ptr)
	Add a synchronous child endpoint. More...

bool	add_async_child (endpoint_ptr_t a_endpoint_ptr)
	Add an asynchronous child endpoint. More...

virtual sent_msg_pkg_ptr	send (request_ptr_t a_request) const
	Sends a request message and returns a channel on which to listen for a reply. More...

virtual sent_msg_pkg_ptr	send (reply_ptr_t a_reply) const
	Sends a reply message. More...

virtual sent_msg_pkg_ptr	send (alert_ptr_t a_alert) const
	Sends an alert message. More...

bool	start ()

bool	listen ()

bool	stop ()

virtual bool	listen_on_queue ()
	Waits for AMQP messages arriving on the channel. More...

virtual void	submit_message (message_ptr_t a_message)
	Submit a message for direct processing. More...

virtual void	send_reply (reply_ptr_t a_reply) const
	Sends a reply message. More...

	snake_case_mv_accessible (uuid_t, id)

	snake_case_mv_referrable (sync_map_t, sync_children)

	snake_case_mv_referrable (async_map_t, async_children)

	snake_case_mv_referrable (std::string, broadcast_key)

Public Member Functions inherited from core
	core (const scarab::param_node &a_config=scarab::param_node(), const std::string &a_broker_address="", unsigned a_port=0, const std::string &a_auth_file="", const bool a_make_connection=true)

	core (const bool a_make_connection, const scarab::param_node &a_config=scarab::param_node())

	core (const core &a_orig)

	core (core &&a_orig)

virtual	~core ()

core &	operator= (const core &a_orig)

core &	operator= (core &&a_orig)

	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)

Public Member Functions inherited from endpoint
	endpoint (const std::string &a_name)

	endpoint (const endpoint &a_orig)

	endpoint (endpoint &&a_orig)

virtual	~endpoint ()

endpoint &	operator= (const endpoint &a_orig)

endpoint &	operator= (endpoint &&a_orig)

	snake_case_mv_referrable (std::string, name)

	snake_case_mv_referrable (service_ptr_t, service)

reply_ptr_t	submit_request_message (const request_ptr_t a_request)
	Directly submit a request message to this endpoint. More...

void	submit_reply_message (const reply_ptr_t a_reply)
	Directly submit a reply message to this endpoint. More...

void	submit_alert_message (const alert_ptr_t a_alert)
	Directly submit an alert message to this endpoint. More...

virtual void	on_reply_message (const reply_ptr_t a_reply)

virtual void	on_alert_message (const alert_ptr_t a_alert)

void	sort_message (const message_ptr_t a_request)

uuid_t	enable_lockout (const scarab::param_node &a_tag)
	enable lockout with randomly-generated key More...

uuid_t	enable_lockout (const scarab::param_node &a_tag, uuid_t a_key)
	enable lockout with user-supplied key More...

bool	disable_lockout (const uuid_t &a_key, bool a_force=false)

bool	is_locked () const

bool	check_key (const uuid_t &a_key) const

Public Member Functions inherited from listener_receiver
	listener_receiver ()

	listener_receiver (const listener_receiver &)=delete

	listener_receiver (listener_receiver &&a_orig)

listener_receiver &	operator= (const listener_receiver &)=delete

listener_receiver &	operator= (listener_receiver &&a_orig)

Public Member Functions inherited from listener
	listener ()

	listener (const listener &)=delete

	listener (listener &&a_orig)

virtual	~listener ()

listener &	operator= (const listener &)=delete

listener &	operator= (listener &&a_orig)

	snake_case_mv_referrable (amqp_channel_ptr, channel)

	snake_case_mv_referrable (std::string, consumer_tag)

	snake_case_mv_accessible (unsigned, listen_timeout_ms)

	snake_case_mv_referrable (std::thread, listener_thread)

Public Member Functions inherited from concurrent_receiver
	concurrent_receiver ()

	concurrent_receiver (const concurrent_receiver &)=delete

	concurrent_receiver (concurrent_receiver &&a_orig)

virtual	~concurrent_receiver ()

concurrent_receiver &	operator= (const concurrent_receiver &)=delete

concurrent_receiver &	operator= (concurrent_receiver &&a_orig)

virtual void	process_message (message_ptr_t a_message)
	Deposits the message in the concurrent queue (called by the listener) More...

void	execute ()
	Handles messages that appear in the concurrent queue by calling `submit_message()`. More...

Public Member Functions inherited from receiver
	receiver ()

	receiver (const receiver &a_orig)=delete

	receiver (receiver &&a_orig)

virtual	~receiver ()

receiver &	operator= (const receiver &a_orig)=delete

receiver &	operator= (receiver &&a_orig)

void	handle_message_chunk (amqp_envelope_ptr a_envelope)

void	wait_for_message (incoming_message_pack &a_pack, const std::string &a_message_id)

void	process_message_pack (incoming_message_pack &a_pack, const std::string &a_message_id)
	Converts a message pack into a Dripline message, and then submits the message for processing. More...

	snake_case_mv_referrable (incoming_message_map, incoming_messages)
	Stores the incomplete messages. More...

	snake_case_mv_accessible (unsigned, single_message_wait_ms)
	Wait time for message chunks from a single message. More...

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

reply_ptr_t	wait_for_reply (const sent_msg_pkg_ptr a_receive_reply, bool &a_chan_valid, int a_timeout_ms=0)

Public Member Functions inherited from heartbeater
	heartbeater (service_ptr_t a_service=service_ptr_t())
	Primary constructor. A service pointer is required to be able to send messages. More...

	heartbeater (const heartbeater &)=delete

	heartbeater (heartbeater &&a_orig)

virtual	~heartbeater ()

heartbeater &	operator= (const heartbeater &)=delete

heartbeater &	operator= (heartbeater &&a_orig)

void	execute (const std::string &a_name, uuid_t a_id, const std::string &a_routing_key)

	snake_case_mv_accessible (unsigned, heartbeat_interval_s)
	Interval between heartbeat alerts (default: 60 s) More...

	snake_case_mv_accessible (unsigned, check_timeout_ms)
	Timing interval for the internal loop (default: 1000 ms) More...

	snake_case_mv_referrable (service_ptr_t, service)

	snake_case_mv_atomic (bool, stop)

Public Member Functions inherited from scheduler<>
	scheduler ()

	scheduler (const scheduler &)=delete

	scheduler (scheduler &&)

virtual	~scheduler ()

scheduler &	operator= (const scheduler &)=delete

scheduler &	operator= (scheduler &&)

int	schedule (executable_t an_executable, time_point_t an_exe_time)

int	schedule (executable_t an_executable, duration_t an_interval, time_point_t an_exe_time=std::chrono::system_clock ::now())

void	unschedule (int an_id)
	Unschedule an event using the event's ID. More...

void	execute ()
	Main execution loop for the scheduler. More...

	snake_case_mv_accessible (duration_t, exe_buffer)
	The time difference from "now" that determines whether an event is executed. More...

	snake_case_mv_accessible (duration_t, cycle_time)
	Main thread cycle time. More...

	snake_case_mv_referrable_const (simple_executor, the_executor)
	The executor used to execute events. More...

	snake_case_mv_referrable_const (events_map_t, events)
	The scheduled events, stored in a map sorted by execution time. More...

Private Types
typedef std::function< reply_ptr_t(const dripline::request_ptr_t) >	handler_func_t

typedef std::unordered_map< std::string, handler_func_t >	handler_funcs_t

Private Member Functions
virtual reply_ptr_t	do_run_request (const request_ptr_t a_request)

virtual reply_ptr_t	do_get_request (const request_ptr_t a_request)

virtual reply_ptr_t	do_set_request (const request_ptr_t a_request)

virtual reply_ptr_t	do_cmd_request (const request_ptr_t a_request)

Private Attributes
handler_func_t	f_run_handler

handler_funcs_t	f_get_handlers

handler_funcs_t	f_set_handlers

handler_funcs_t	f_cmd_handlers

Additional Inherited Members
Public Types inherited from service
typedef std::map< std::string, endpoint_ptr_t >	sync_map_t

typedef std::map< std::string, lr_ptr_t >	async_map_t

Public Types inherited from scheduler<>
using	clock_t = std::chrono::system_clock

using	time_point_t = typename std::chrono::system_clock ::time_point

using	duration_t = typename std::chrono::system_clock ::duration

using	executable_t = std::function< void() >

typedef std::multimap< time_point_t, event >	events_map_t

Static Public Member Functions inherited from core
static bool	listen_for_message (amqp_envelope_ptr &a_envelope, amqp_channel_ptr a_channel, const std::string &a_consumer_tag, int a_timeout_ms=0, bool a_do_ack=true)
	return: if false, channel is no longer useable; if true, may be reused More...

Public Attributes inherited from scheduler<>
snake_case_mv_accessible_static(int, curr_id) protected std::recursive_mutex	f_scheduler_mutex
	The ID to be used for the next scheduled event. More...

std::mutex	f_executor_mutex

std::condition_variable_any	f_cv

std::thread	f_scheduler_thread

Static Public Attributes inherited from core
static bool	s_offline = false

Protected Types inherited from service
enum	status { nothing = 0, channel_created = 10, exchange_declared = 20, queue_declared = 30, queue_bound = 40, consuming = 50, listening = 60, processing = 70 }

Protected Member Functions inherited from service
virtual bool	open_channels ()

virtual bool	setup_queues ()

virtual bool	bind_keys ()

virtual bool	start_consuming ()

virtual bool	stop_consuming ()

virtual bool	remove_queue ()

virtual reply_ptr_t	on_request_message (const request_ptr_t a_request)
	Default request handler; passes request to initial request functions. More...

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) 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

Protected Member Functions inherited from endpoint
bool	authenticate (const uuid_t &a_key) const
	Returns true if the server is unlocked or if it's locked and the key matches the lockout key; returns false otherwise. More...

	snake_case_mv_referrable (scarab::param_node, lockout_tag)

	snake_case_mv_accessible (uuid_t, lockout_key)

Protected Member Functions inherited from concurrent_receiver
	snake_case_mv_referrable (scarab::concurrent_queue< message_ptr_t >, message_queue)

	snake_case_mv_referrable (std::thread, receiver_thread)

Protected Member Functions inherited from receiver
reply_ptr_t	process_received_reply (incoming_message_pack &a_pack, const std::string &a_message_id)

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)

Protected Attributes inherited from heartbeater
std::thread	f_heartbeat_thread

Detailed Description

Service class aimed at adding a Dripline API to an existing codebase.

Author: N.S. Oblath

Hub is a tool to a Dripline API onto existing codebase. Message-handler functions in the codebase are mapped to Dripline run, command, get, and set requests.

The handler functions need to have the signature reply_ptr_t( const request_ptr_t ). Typically those handler functions will wrap another function in the codebase and provide the tranlation between dripline message and the input/output of the function.

The key for the request–>function mapping for any given request message are given in the message's specifier.

As an example of adding a Dripline API, say you had class foo, which had a function int bar( int ):

class foo
{
    int bar( int a_input )
    {
        return 2 * a_input;
    }
};

To add a Dripline API for this class you first need a message-handling function:

class foo
{
    [...]
    dripline::reply_ptr_t handle_bar( dripline::request_ptr_t a_request )
    {
        param_ptr_t t_reply_payload( new param_node() );
        t_reply_payload->as_node().add( "value", bar( a_request->payload()["values"][0]().as_int() );
        return a_request->reply( dl_success(), "Bar request succeeded", std::move(t_reply_payload) );
    }
};

Then you would create a hub as part of the codebase, and either in the hub or somewhere else you would register the bar handler function:

register_cmd_handler( "bar", std::bind( &foo::handle_bar, this, _1 ) );

In this case we choose to make it a Dripline command request, since that seems to fit the best with performing the foo::bar operation.

Definition at line 70 of file hub.hh.

Member Typedef Documentation

◆ handler_func_t

typedef std::function< reply_ptr_t( const dripline::request_ptr_t ) > handler_func_t

private

Definition at line 73 of file hub.hh.

◆ handler_funcs_t

typedef std::unordered_map< std::string, handler_func_t > handler_funcs_t

private

Definition at line 107 of file hub.hh.

Constructor & Destructor Documentation

◆ hub()

hub	(	const scarab::param_node &	a_config = `scarab::param_node()`,
		const std::string &	a_queue_name = `""`,
		const std::string &	a_broker_address = `""`,
		unsigned	a_port = `0`,
		const std::string &	a_auth_file = `""`,
		const bool	a_make_connection = `true`
	)

Definition at line 22 of file hub.cc.

◆ ~hub()

~hub ( )

virtual

Definition at line 32 of file hub.cc.

Member Function Documentation

◆ do_cmd_request()

reply_ptr_t do_cmd_request ( const request_ptr_t a_request )

privatevirtual

Reimplemented from endpoint.

Definition at line 140 of file hub.cc.

◆ do_get_request()

reply_ptr_t do_get_request ( const request_ptr_t a_request )

privatevirtual

Reimplemented from endpoint.

Definition at line 108 of file hub.cc.

◆ do_run_request()

reply_ptr_t do_run_request ( const request_ptr_t a_request )

privatevirtual

Reimplemented from endpoint.

Definition at line 103 of file hub.cc.

◆ do_set_request()

reply_ptr_t do_set_request ( const request_ptr_t a_request )

privatevirtual

Reimplemented from endpoint.

Definition at line 124 of file hub.cc.

◆ register_cmd_handler()

void register_cmd_handler	(	const std::string &	a_key,
		const handler_func_t &	a_func
	)

Sets a command request handler function.

Definition at line 57 of file hub.cc.

◆ register_get_handler()

void register_get_handler	(	const std::string &	a_key,
		const handler_func_t &	a_func
	)

Sets a get request handler function.

Definition at line 43 of file hub.cc.

◆ register_set_handler()

void register_set_handler	(	const std::string &	a_key,
		const handler_func_t &	a_func
	)

Sets a set request handler function.

Definition at line 50 of file hub.cc.

◆ remove_cmd_handler()

void remove_cmd_handler ( const std::string & a_key )

Removes a command request handler function.

Definition at line 90 of file hub.cc.

◆ remove_get_handler()

void remove_get_handler ( const std::string & a_key )

Removes a get request handler function.

Definition at line 64 of file hub.cc.

◆ remove_set_handler()

void remove_set_handler ( const std::string & a_key )

Removes a set request handler function.

Definition at line 77 of file hub.cc.

◆ set_run_handler()

void set_run_handler ( const handler_func_t & a_func )

Sets the run request handler function.

Definition at line 36 of file hub.cc.

Member Data Documentation

◆ f_cmd_handlers

handler_funcs_t f_cmd_handlers

private

Definition at line 110 of file hub.hh.

◆ f_get_handlers

handler_funcs_t f_get_handlers

private

Definition at line 108 of file hub.hh.

◆ f_run_handler

handler_func_t f_run_handler

private

Definition at line 105 of file hub.hh.

◆ f_set_handlers

handler_funcs_t f_set_handlers

private

Definition at line 109 of file hub.hh.

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

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ handler_func_t

◆ handler_funcs_t

Constructor & Destructor Documentation

◆ hub()

◆ ~hub()

Member Function Documentation

◆ do_cmd_request()

◆ do_get_request()

◆ do_run_request()

◆ do_set_request()

◆ register_cmd_handler()

◆ register_get_handler()

◆ register_set_handler()

◆ remove_cmd_handler()

◆ remove_get_handler()

◆ remove_set_handler()

◆ set_run_handler()

Member Data Documentation

◆ f_cmd_handlers

◆ f_get_handlers

◆ f_run_handler

◆ f_set_handlers