This class wraps communication via UX sockets. More...

#include <processor.hpp>

Classes
struct	EmptyData

struct	ErrorProtocolMessage

struct	MessageHeader

struct	MethodHandlers

struct	PeerInfo

struct	RegisterSignalsProtocolMessage

struct	ReturnCallbacks

struct	SignalHandlers

Public Member Functions
	Processor (epoll::EventPoll &eventPoll, const std::string &logName="", const PeerCallback &newPeerCallback=nullptr, const PeerCallback &removedPeerCallback=nullptr, const unsigned int maxNumberOfPeers=DEFAULT_MAX_NUMBER_OF_PEERS)
	Constructs the Processor, but doesn't start it. More...

	~Processor ()

	Processor (const Processor &)=delete

	Processor (Processor &&)=delete

Processor &	operator= (const Processor &)=delete

void	start ()
	Start processing. More...

bool	isStarted ()

void	stop (bool wait)
	Stops the processing thread. More...

void	setNewPeerCallback (const PeerCallback &newPeerCallback)
	Set the callback called for each new connection to a peer. More...

void	setRemovedPeerCallback (const PeerCallback &removedPeerCallback)
	Set the callback called when connection to a peer is lost. More...

PeerID	addPeer (const std::shared_ptr< Socket > &socketPtr)
	From now on socket is owned by the Processor object. More...

template<typename SentDataType , typename ReceivedDataType >
void	setMethodHandler (const MethodID methodID, const typename MethodHandler< SentDataType, ReceivedDataType >::type &process)
	Saves the callbacks connected to the method id. More...

template<typename ReceivedDataType >
void	setSignalHandler (const MethodID methodID, const typename SignalHandler< ReceivedDataType >::type &process)
	Saves the callbacks connected to the method id. More...

void	sendResult (const MethodID methodID, const PeerID &peerID, const MessageID &messageID, const std::shared_ptr< void > &data)
	Send result of the method. More...

void	sendError (const PeerID &peerID, const MessageID &messageID, const int errorCode, const std::string &message)
	Send error result of the method. More...

void	sendVoid (const MethodID methodID, const PeerID &peerID, const MessageID &messageID)
	Indicate that the method handler finished. More...

void	removeMethod (const MethodID methodID)
	Removes the callback associated with specific method id. More...

bool	isHandled (const MethodID methodID)

template<typename SentDataType , typename ReceivedDataType >
std::shared_ptr< ReceivedDataType >	callSync (const MethodID methodID, const PeerID &peerID, const std::shared_ptr< SentDataType > &data, unsigned int timeoutMS=5000)
	Synchronous method call. More...

template<typename SentDataType , typename ReceivedDataType >
MessageID	callAsync (const MethodID methodID, const PeerID &peerID, const std::shared_ptr< SentDataType > &data, const typename ResultHandler< ReceivedDataType >::type &process)
	Asynchronous method call. More...

template<typename SentDataType , typename ReceivedDataType >
MessageID	callAsyncNonBlock (const MethodID methodID, const PeerID &peerID, const std::shared_ptr< SentDataType > &data, const typename ResultHandler< ReceivedDataType >::type &process)
	The same as callAsync, but not blocking on the state mutex. More...

template<typename SentDataType >
void	signal (const MethodID methodID, const std::shared_ptr< SentDataType > &data)
	Send a signal to the peer. More...

bool	handleLostConnection (const FileDescriptor fd)
	Removes one peer. More...

bool	handleInput (const FileDescriptor fd)
	Handles input from one peer. More...

bool	handleEvent ()
	Handle one event from the internal event's queue. More...

FileDescriptor	getEventFD ()

Static Public Attributes
static const MethodID	RETURN_METHOD_ID = std::numeric_limits<MethodID>::max()
	Used to indicate a message with the return value. More...

static const MethodID	REGISTER_SIGNAL_METHOD_ID = std::numeric_limits<MethodID>::max() - 1
	Indicates an Processor's internal request/broadcast to register a Signal. More...

static const MethodID	ERROR_METHOD_ID = std::numeric_limits<MethodID>::max() - 2
	Error return message. More...

Private Types
enum	Event { Event::FINISH, Event::METHOD, Event::SIGNAL, Event::ADD_PEER, Event::REMOVE_PEER, Event::SEND_RESULT, Event::REMOVE_METHOD }

typedef std::unique_lock < std::mutex >	Lock

typedef RequestQueue< Event > ::Request	Request

typedef std::vector< PeerInfo >	Peers

Private Member Functions
template<typename SentDataType , typename ReceivedDataType >
void	setMethodHandlerInternal (const MethodID methodID, const typename MethodHandler< SentDataType, ReceivedDataType >::type &process)

template<typename ReceivedDataType >
void	setSignalHandlerInternal (const MethodID methodID, const typename SignalHandler< ReceivedDataType >::type &handler)

template<typename SentDataType >
void	signalInternal (const MethodID methodID, const PeerID &peerID, const std::shared_ptr< SentDataType > &data)

bool	onMethodRequest (MethodRequest &request)

bool	onSignalRequest (SignalRequest &request)

bool	onAddPeerRequest (AddPeerRequest &request)

bool	onRemovePeerRequest (RemovePeerRequest &request)

bool	onSendResultRequest (SendResultRequest &request)

bool	onRemoveMethodRequest (RemoveMethodRequest &request)

bool	onFinishRequest (FinishRequest &request)

bool	onReturnValue (Peers::iterator &peerIt, const MessageID &messageID)

bool	onRemoteMethod (Peers::iterator &peerIt, const MethodID methodID, const MessageID &messageID, std::shared_ptr< MethodHandlers > methodCallbacks)

bool	onRemoteSignal (Peers::iterator &peerIt, const MethodID methodID, const MessageID &messageID, std::shared_ptr< SignalHandlers > signalCallbacks)

void	removePeerInternal (Peers::iterator peerIt, const std::exception_ptr &exceptionPtr)

void	removePeerSyncInternal (const PeerID &peerID, Lock &lock)

bool	onNewSignals (const PeerID &peerID, std::shared_ptr< RegisterSignalsProtocolMessage > &data)

bool	onErrorSignal (const PeerID &peerID, std::shared_ptr< ErrorProtocolMessage > &data)

Peers::iterator	getPeerInfoIterator (const FileDescriptor fd)

Peers::iterator	getPeerInfoIterator (const PeerID &peerID)

Private Attributes
epoll::EventPoll &	mEventPoll

std::string	mLogPrefix

RequestQueue< Event >	mRequestQueue

bool	mIsRunning

std::unordered_map< MethodID, std::shared_ptr < MethodHandlers > >	mMethodsCallbacks

std::unordered_map< MethodID, std::shared_ptr < SignalHandlers > >	mSignalsCallbacks

std::unordered_map< MethodID, std::list< PeerID > >	mSignalsPeers

Peers	mPeerInfo

std::unordered_map< MessageID, ReturnCallbacks >	mReturnCallbacks

std::mutex	mStateMutex

PeerCallback	mNewPeerCallback

PeerCallback	mRemovedPeerCallback

unsigned int	mMaxNumberOfPeers

Friends
std::ostream &	operator<< (std::ostream &os, const Processor::Event &event)

Detailed Description

This class wraps communication via UX sockets.

It's intended to be used both in Client and Service classes. It uses a serialization mechanism from Config. Library user will only have to pass the types that each call will send and receive

Message format:

MethodID - probably casted enum. MethodID == std::numeric_limits<MethodID>::max() is reserved for return messages
MessageID - unique id of a message exchange sent by this object instance. Used to identify reply messages.
Rest: The data written in a callback. One type per method.ReturnCallbacks

TODO:

synchronous call to many peers
implement HandlerStore class for storing both signals and methods
API for removing signals
implement CallbackStore - thread safe calling/setting callbacks
helper function for removing from unordered map
callbacks for serialization/parsing
store Sockets in a vector, maybe SocketStore?
waiting till the EventQueue is empty before leaving stop()
no new events added after stop() called

Member Typedef Documentation

typedef std::unique_lock<std::mutex> cargo::ipc::Processor::Lock

private

typedef std::vector<PeerInfo> cargo::ipc::Processor::Peers

private

typedef RequestQueue<Event>::Request cargo::ipc::Processor::Request

private

Member Enumeration Documentation

enum cargo::ipc::Processor::Event

strongprivate

Enumerator
FINISH
METHOD
SIGNAL
ADD_PEER
REMOVE_PEER
SEND_RESULT
REMOVE_METHOD

Constructor & Destructor Documentation

cargo::ipc::Processor::Processor	(	epoll::EventPoll &	eventPoll,
		const std::string &	logName = `""`,
		const PeerCallback &	newPeerCallback = `nullptr`,
		const PeerCallback &	removedPeerCallback = `nullptr`,
		const unsigned int	maxNumberOfPeers = `DEFAULT_MAX_NUMBER_OF_PEERS`
	)

Constructs the Processor, but doesn't start it.

The object is ready to add methods.

Parameters

eventPoll	event poll
logName	log name
newPeerCallback	called when a new peer arrives
removedPeerCallback	called when the Processor stops listening for this peer
maxNumberOfPeers	maximal number of peers

cargo::ipc::Processor::~Processor ( )

cargo::ipc::Processor::Processor ( const Processor & )

delete

cargo::ipc::Processor::Processor ( Processor && )

delete

Member Function Documentation

PeerID cargo::ipc::Processor::addPeer ( const std::shared_ptr< Socket > & socketPtr )

From now on socket is owned by the Processor object.

Calls the newPeerCallback.

Parameters

socketPtr pointer to the new socket

Returns: peerID of the new user

template<typename SentDataType , typename ReceivedDataType >

MessageID cargo::ipc::Processor::callAsync	(	const MethodID	methodID,
		const PeerID &	peerID,
		const std::shared_ptr< SentDataType > &	data,
		const typename ResultHandler< ReceivedDataType >::type &	process
	)

Asynchronous method call.

Parameters

methodID	API dependent id of the method
peerID	id of the peer
data	data to sent
process	callback processing the return data

Template Parameters

SentDataType	data type to send
ReceivedDataType	data type to receive

template<typename SentDataType , typename ReceivedDataType >

MessageID cargo::ipc::Processor::callAsyncNonBlock	(	const MethodID	methodID,
		const PeerID &	peerID,
		const std::shared_ptr< SentDataType > &	data,
		const typename ResultHandler< ReceivedDataType >::type &	process
	)

The same as callAsync, but not blocking on the state mutex.

Parameters

methodID	API dependent id of the method
peerID	id of the peer
data	data to sent
process	callback processing the return data

Template Parameters

SentDataType	data type to send
ReceivedDataType	data type to receive

template<typename SentDataType , typename ReceivedDataType >

std::shared_ptr< ReceivedDataType > cargo::ipc::Processor::callSync	(	const MethodID	methodID,
		const PeerID &	peerID,
		const std::shared_ptr< SentDataType > &	data,
		unsigned int	timeoutMS = `5000`
	)

Synchronous method call.

Parameters

methodID	API dependent id of the method
peerID	id of the peer
data	data to send
timeoutMS	optional, how long to wait for the return value before throw (milliseconds, default: 5000)

Template Parameters

SentDataType	data type to send
ReceivedDataType	data type to receive

Returns: call result data

FileDescriptor cargo::ipc::Processor::getEventFD ( )

Returns: file descriptor for the internal event's queue

Processor::Peers::iterator cargo::ipc::Processor::getPeerInfoIterator ( const FileDescriptor fd )

private

Processor::Peers::iterator cargo::ipc::Processor::getPeerInfoIterator ( const PeerID & peerID )

private

bool cargo::ipc::Processor::handleEvent ( )

Handle one event from the internal event's queue.

Returns: should the polling structure be rebuild

bool cargo::ipc::Processor::handleInput ( const FileDescriptor fd )

Handles input from one peer.

Handler used in external polling.

Parameters

fd	file description identifying the peer

Returns: should the polling structure be rebuild

bool cargo::ipc::Processor::handleLostConnection ( const FileDescriptor fd )

Removes one peer.

Handler used in external polling.

Parameters

fd	file description identifying the peer

Returns: should the polling structure be rebuild

bool cargo::ipc::Processor::isHandled ( const MethodID methodID )

Parameters

methodID MethodID defined in the user's API

Returns: is methodID handled by a signal or method

bool cargo::ipc::Processor::isStarted ( )

Returns: is processor running

bool cargo::ipc::Processor::onAddPeerRequest ( AddPeerRequest & request )

private

bool cargo::ipc::Processor::onErrorSignal	(	const PeerID &	peerID,
		std::shared_ptr< ErrorProtocolMessage > &	data
	)

private

bool cargo::ipc::Processor::onFinishRequest ( FinishRequest & request )

private

bool cargo::ipc::Processor::onMethodRequest ( MethodRequest & request )

private

bool cargo::ipc::Processor::onNewSignals	(	const PeerID &	peerID,
		std::shared_ptr< RegisterSignalsProtocolMessage > &	data
	)

private

bool cargo::ipc::Processor::onRemoteMethod	(	Peers::iterator &	peerIt,
		const MethodID	methodID,
		const MessageID &	messageID,
		std::shared_ptr< MethodHandlers >	methodCallbacks
	)

private

bool cargo::ipc::Processor::onRemoteSignal	(	Peers::iterator &	peerIt,
		const MethodID	methodID,
		const MessageID &	messageID,
		std::shared_ptr< SignalHandlers >	signalCallbacks
	)

private

bool cargo::ipc::Processor::onRemoveMethodRequest ( RemoveMethodRequest & request )

private

bool cargo::ipc::Processor::onRemovePeerRequest ( RemovePeerRequest & request )

private

bool cargo::ipc::Processor::onReturnValue	(	Peers::iterator &	peerIt,
		const MessageID &	messageID
	)

private

bool cargo::ipc::Processor::onSendResultRequest ( SendResultRequest & request )

private

bool cargo::ipc::Processor::onSignalRequest ( SignalRequest & request )

private

Processor& cargo::ipc::Processor::operator= ( const Processor & )

delete

void cargo::ipc::Processor::removeMethod ( const MethodID methodID )

Removes the callback associated with specific method id.

Parameters

methodID API dependent id of the method

See Also: setMethodHandler(); setSignalHandler()

void cargo::ipc::Processor::removePeerInternal	(	Peers::iterator	peerIt,
		const std::exception_ptr &	exceptionPtr
	)

private

void cargo::ipc::Processor::removePeerSyncInternal	(	const PeerID &	peerID,
		Lock &	lock
	)

private

void cargo::ipc::Processor::sendError	(	const PeerID &	peerID,
		const MessageID &	messageID,
		const int	errorCode,
		const std::string &	message
	)

Send error result of the method.

Parameters

peerID	id of the peer
messageID	id of the message to which it replies
errorCode	code of the error
message	description of the error

void cargo::ipc::Processor::sendResult	(	const MethodID	methodID,
		const PeerID &	peerID,
		const MessageID &	messageID,
		const std::shared_ptr< void > &	data
	)

Send result of the method.

Used for asynchronous communication, only internally.

Parameters

methodID	API dependent id of the method
peerID	id of the peer
messageID	id of the message to which it replies
data	data to send

void cargo::ipc::Processor::sendVoid	(	const MethodID	methodID,
		const PeerID &	peerID,
		const MessageID &	messageID
	)

Indicate that the method handler finished.

Parameters

methodID	API dependent id of the method
peerID	id of the peer
messageID	id of the message to which it replies

template<typename SentDataType , typename ReceivedDataType >

void cargo::ipc::Processor::setMethodHandler	(	const MethodID	methodID,
		const typename MethodHandler< SentDataType, ReceivedDataType >::type &	process
	)

Saves the callbacks connected to the method id.

When a message with the given method id is received, the data will be passed to the serialization callback through file descriptor.

Then the process callback will be called with the parsed data.

Parameters

methodID	API dependent id of the method
process	data processing callback

Template Parameters

SentDataType	data type to send
ReceivedDataType	data type to receive

template<typename SentDataType , typename ReceivedDataType >

void cargo::ipc::Processor::setMethodHandlerInternal	(	const MethodID	methodID,
		const typename MethodHandler< SentDataType, ReceivedDataType >::type &	process
	)

private

void cargo::ipc::Processor::setNewPeerCallback ( const PeerCallback & newPeerCallback )

Set the callback called for each new connection to a peer.

Parameters

newPeerCallback the callback

void cargo::ipc::Processor::setRemovedPeerCallback ( const PeerCallback & removedPeerCallback )

Set the callback called when connection to a peer is lost.

Parameters

removedPeerCallback the callback

template<typename ReceivedDataType >

void cargo::ipc::Processor::setSignalHandler	(	const MethodID	methodID,
		const typename SignalHandler< ReceivedDataType >::type &	process
	)

Saves the callbacks connected to the method id.

When a message with the given method id is received, the data will be passed to the serialization callback through file descriptor.

Then the process callback will be called with the parsed data. There is no return data to send back.

Adding signal sends a registering message to all peers

Parameters

methodID	API dependent id of the method
process	data processing callback

Template Parameters

ReceivedDataType data type to receive

template<typename ReceivedDataType >

void cargo::ipc::Processor::setSignalHandlerInternal	(	const MethodID	methodID,
		const typename SignalHandler< ReceivedDataType >::type &	handler
	)

private

template<typename SentDataType >

void cargo::ipc::Processor::signal	(	const MethodID	methodID,
		const std::shared_ptr< SentDataType > &	data
	)

Send a signal to the peer.

There is no return value from the peer Sends any data only if a peer registered this a signal

Parameters

methodID	API dependent id of the method
data	data to sent

Template Parameters

SentDataType data type to send

template<typename SentDataType >

void cargo::ipc::Processor::signalInternal	(	const MethodID	methodID,
		const PeerID &	peerID,
		const std::shared_ptr< SentDataType > &	data
	)

private

void cargo::ipc::Processor::start ( )

Start processing.

void cargo::ipc::Processor::stop ( bool wait )

Stops the processing thread.

No incoming data will be handled after.

Parameters

wait	does it block waiting for all internals to stop

Friends And Related Function Documentation

std::ostream& operator<<	(	std::ostream &	os,
		const Processor::Event &	event
	)

friend

Member Data Documentation

const MethodID cargo::ipc::Processor::ERROR_METHOD_ID = std::numeric_limits<MethodID>::max() - 2

static

Error return message.

epoll::EventPoll& cargo::ipc::Processor::mEventPoll

private

bool cargo::ipc::Processor::mIsRunning

private

std::string cargo::ipc::Processor::mLogPrefix

private

unsigned int cargo::ipc::Processor::mMaxNumberOfPeers

private

std::unordered_map<MethodID, std::shared_ptr<MethodHandlers> > cargo::ipc::Processor::mMethodsCallbacks

private

PeerCallback cargo::ipc::Processor::mNewPeerCallback

private

Peers cargo::ipc::Processor::mPeerInfo

private

PeerCallback cargo::ipc::Processor::mRemovedPeerCallback

private

RequestQueue<Event> cargo::ipc::Processor::mRequestQueue

private

std::unordered_map<MessageID, ReturnCallbacks> cargo::ipc::Processor::mReturnCallbacks

private

std::unordered_map<MethodID, std::shared_ptr<SignalHandlers> > cargo::ipc::Processor::mSignalsCallbacks

private

std::unordered_map<MethodID, std::list<PeerID> > cargo::ipc::Processor::mSignalsPeers

private

std::mutex cargo::ipc::Processor::mStateMutex

private

const MethodID cargo::ipc::Processor::REGISTER_SIGNAL_METHOD_ID = std::numeric_limits<MethodID>::max() - 1

static

Indicates an Processor's internal request/broadcast to register a Signal.

const MethodID cargo::ipc::Processor::RETURN_METHOD_ID = std::numeric_limits<MethodID>::max()

static

Used to indicate a message with the return value.

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

libs/cargo-ipc/internals/processor.hpp
libs/cargo-ipc/internals/processor.cpp

Classes

Public Member Functions

Static Public Attributes

Private Types

Private Member Functions

Private Attributes

Friends

Detailed Description

Member Typedef Documentation

Member Enumeration Documentation

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation

Member Data Documentation