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

virtual	~receiver ()=default

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 all message chunks from a single dripline message. More...

	snake_case_mv_accessible (unsigned, reply_listen_timeout_ms)
	Listen timeout for individual message chunks when waiting for replies. 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, core::post_listen_status &a_status, 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 77 of file receiver.hh.

Constructor & Destructor Documentation

◆ receiver() [1/3]

receiver ( )

Definition at line 46 of file receiver.cc.

◆ receiver() [2/3]

receiver ( const receiver & a_orig )

delete

◆ receiver() [3/3]

receiver ( receiver && a_orig )

default

◆ ~receiver()

virtual ~receiver ( )

virtualdefault

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 62 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 53 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 201 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 173 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 364 of file receiver.cc.

◆ snake_case_mv_accessible() [1/2]

snake_case_mv_accessible	(	unsigned	,
		reply_listen_timeout_ms
	)

Listen timeout for individual message chunks when waiting for replies.

◆ snake_case_mv_accessible() [2/2]

snake_case_mv_accessible	(	unsigned	,
		single_message_wait_ms
	)

Wait time for all message chunks from a single dripline 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 139 of file receiver.cc.

◆ wait_for_reply() [1/2]

reply_ptr_t wait_for_reply	(	const sent_msg_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_chan_valid	Returns whether or not the channel is still valid for use after receiving a message.

Returns: Reply message

Definition at line 212 of file receiver.cc.

◆ wait_for_reply() [2/2]

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