A receiver is able to collect Dripline message chunks and reassemble them into a complete Dripline message. More...

#include <receiver.hh>

Inheritance diagram for receiver:

Public Member Functions
	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...

virtual void	process_message (message_ptr_t a_message)

	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)

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

Detailed Description

A receiver is able to collect Dripline message chunks and reassemble them into a complete Dripline message.

Author: N.S. Oblath

This is a mix-in class for synchronously receiving and processing Dripline messages.

Dripline messages can be broken up into multiple chunks, each of which is transported as an AMQP message. A receiver is responsible for handling message chunks, storing incomplete Dripline messages, and eventually processing complete Dripline messages.

The receiver class contains an interface specifically for users waiting to receive reply messages: wait_for_reply().

When the first message chunk for a message is received, one of two things happens:

if the message comprises one chunk, then the message is processed immediately;
if the message comprises multiple chunks, then a separate thread is spun up to wait for the remaining chunks.

Incomplete messages are stored in the incoming-message map. Message chunks for a given message can be received in any order. The receiver will wait single_message_wait_ms ms for all of the chunks of a message to arrive before timing out processing the incomplete message.

The actual assembly of message chunks into complete messages is done in message.

The receiver class itself does not know how to process a message. This must be implemented by the class derived from receiver. The default implementation of process_message() will throw a dripline_error.

Definition at line 76 of file receiver.hh.

Constructor & Destructor Documentation

◆ receiver() [1/3]

receiver ( )

Definition at line 47 of file receiver.cc.

◆ receiver() [2/3]

receiver ( const receiver & a_orig )

delete

◆ receiver() [3/3]

receiver ( receiver && a_orig )

Definition at line 53 of file receiver.cc.

◆ ~receiver()

~receiver ( )

virtual

Definition at line 61 of file receiver.cc.

Member Function Documentation

◆ handle_message_chunk()

void handle_message_chunk ( amqp_envelope_ptr a_envelope )

Processes a message chunk: starts a new message pack if it's the first of multiple messages, or puts the chunk in the correct existing message pack. For single-chunk messages, processes the message immediately.

Definition at line 73 of file receiver.cc.

◆ operator=() [1/2]

receiver& operator= ( const receiver & a_orig )

delete

◆ operator=() [2/2]

receiver & operator= ( receiver && a_orig )

Definition at line 64 of file receiver.cc.

◆ process_message()

void process_message ( message_ptr_t a_message )

virtual

Processes a single Dripline message. This is the default implementation that always throws a dripline_error.

Reimplemented in concurrent_receiver.

Definition at line 212 of file receiver.cc.

◆ process_message_pack()

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.

Definition at line 184 of file receiver.cc.

◆ process_received_reply()

reply_ptr_t process_received_reply	(	incoming_message_pack &	a_pack,
		const std::string &	a_message_id
	)

protected

Definition at line 328 of file receiver.cc.

◆ snake_case_mv_accessible()

snake_case_mv_accessible	(	unsigned	,
		single_message_wait_ms
	)

Wait time for message chunks from a single message.

◆ snake_case_mv_referrable()

snake_case_mv_referrable	(	incoming_message_map	,
		incoming_messages
	)

Stores the incomplete messages.

◆ wait_for_message()

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

Waits for messages for a set amount of time (single_message_wait_ms), and submits the message pack for processing. Intended to be used in a separate thread for each message pack.

Definition at line 149 of file receiver.cc.

◆ wait_for_reply() [1/2]

reply_ptr_t wait_for_reply	(	const sent_msg_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 217 of file receiver.cc.

◆ wait_for_reply() [2/2]

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