ZMQClient Class Reference

A helper class for communicating over ZMQ. Includes a multicast and (if needed) also a data socket. More...

#include <ZMQClient.h>

Collaboration diagram for ZMQClient:

Public Member Functions
template<int AZMQType>
void	initialize (const std::string &pubSocketAddress, const std::string &subSocketAddress, const std::string &socketAddress, bool bind)
	Initialize the multicast and a data socket of the given type.
void	initialize (const std::string &pubSocketAddress, const std::string &subSocketAddress)
	Initialize only the multicast.
void	terminate (bool sendGoodbye=true)
	Terminate the sockets properly.
void	reset ()
	Reset the sockets. ATTENTION: this does not close the sockets! Use only after forks to not clean up to times.
void	subscribe (EMessageTypes messageType)
	Subscribe to the given multicast message type.
template<class AZMQMessage>
void	send (AZMQMessage message) const
	Send a message over the data socket.
void	send (zmq::message_t &message) const
	Send a zmq message over the data socket. ATTENTION: we are taking ownership here!
template<class AZMQMessage>
void	publish (AZMQMessage message) const
	Publish the message to the multicast.
bool	isOnline () const
	Check if the client was initialized and not terminated.
template<class AMulticastAnswer, class ASocketAnswer>
int	poll (unsigned int timeout, AMulticastAnswer multicastAnswer, ASocketAnswer socketAnswer) const
	Poll both the multicast and the data socket until, either:
template<class ASocketAnswer>
int	pollSocket (unsigned int timeout, ASocketAnswer socketAnswer) const
	Poll method to only the data socket.
template<class AMulticastAnswer>
int	pollMulticast (unsigned int timeout, AMulticastAnswer multicastAnswer) const
	Poll method to only the multicast socket.

Static Private Member Functions
static int	pollSocketVector (const std::vector< zmq::socket_t * > &socketList, int timeout)
	Internal poll function.

Private Attributes
std::unique_ptr< zmq::context_t >	m_context
	ZMQ context.
std::vector< zmq::socket_t * >	m_pollSocketPtrList
	Will use this vector for polling.
std::unique_ptr< zmq::socket_t >	m_pubSocket
	ZMQ Pub socket.
std::unique_ptr< zmq::socket_t >	m_subSocket
	ZMQ sub socket.
std::unique_ptr< zmq::socket_t >	m_socket
	ZMQ socket.

Detailed Description

A helper class for communicating over ZMQ. Includes a multicast and (if needed) also a data socket.

Definition at line 22 of file ZMQClient.h.

Member Function Documentation

initialize() [1/2]

void initialize	(	const std::string &	pubSocketAddress,
		const std::string &	subSocketAddress )

Initialize only the multicast.

Definition at line 87 of file ZMQClient.cc.

{
  m_context = std::make_unique<zmq::context_t>(1);
  m_pubSocket = std::make_unique<zmq::socket_t>(*m_context, ZMQ_PUB);
  m_subSocket = std::make_unique<zmq::socket_t>(*m_context, ZMQ_SUB);
 
  m_pubSocket->connect(pubSocketAddress);
  m_pubSocket->set(zmq::sockopt::linger, 0);
 
  m_subSocket->connect(subSocketAddress);
  m_subSocket->set(zmq::sockopt::linger, 0);
 
  B2DEBUG(200, "Having initialized multicast with sub on " << subSocketAddress << " and pub on " << pubSocketAddress);
 
  std::this_thread::sleep_for(std::chrono::milliseconds(10));
 
  m_pollSocketPtrList.clear();
  m_pollSocketPtrList.push_back(m_subSocket.get());
}

initialize() [2/2]

template<int AZMQType>

template void initialize< ZMQ_PUB >	(	const std::string &	pubSocketAddress,
		const std::string &	subSocketAddress,
		const std::string &	socketAddress,
		bool	bind )

Initialize the multicast and a data socket of the given type.

Definition at line 61 of file ZMQClient.cc.

{
  initialize(pubSocketAddress, subSocketAddress);
  m_socket = std::make_unique<zmq::socket_t>(*m_context, AZMQType);
 
  if (AZMQType == ZMQ_DEALER) {
    const std::string uniqueID = std::to_string(getpid());
    m_socket->set(zmq::sockopt::routing_id, uniqueID);
  }
 
  m_socket->set(zmq::sockopt::linger, 0);
  if (bind) {
    m_socket->bind(socketAddress.c_str());
  } else {
    m_socket->connect(socketAddress.c_str());
  }
 
  // Give the sockets some time to start
  std::this_thread::sleep_for(std::chrono::milliseconds(10));
 
  B2DEBUG(100, "Created socket: " << socketAddress);
 
  m_pollSocketPtrList.push_back(m_socket.get());
}

isOnline()

bool isOnline ( ) const

inline

Check if the client was initialized and not terminated.

Definition at line 59 of file ZMQClient.h.

    {
      return m_context.get();
    }

pollSocketVector()

int pollSocketVector ( const std::vector< zmq::socket_t * > & socketList, int timeout )

staticprivate

Internal poll function.

Definition at line 126 of file ZMQClient.cc.

{
  auto start = std::chrono::system_clock::now();
  int return_bitmask = 0;
  assert(socketList.size() <= 2);
  std::vector<zmq::pollitem_t> items(socketList.size());
 
  for (unsigned int i = 0; i < socketList.size(); i++) {
    items[i].socket = static_cast<void*>(*socketList[i]);
    items[i].events = ZMQ_POLLIN;
    items[i].revents = 0;
  }
 
  while (timeout >= 0) {
    try {
      zmq::poll(items.data(), socketList.size(), timeout);
 
      for (unsigned int i = 0; i < socketList.size(); i++) {
        if (static_cast<bool>(items[i].revents & ZMQ_POLLIN)) {
          return_bitmask = return_bitmask | 1 << i;
        }
      }
      return return_bitmask;
    } catch (zmq::error_t& error) {
      if (error.num() == EINTR) {
        auto now = std::chrono::system_clock::now();
        timeout -= std::chrono::duration_cast<std::chrono::milliseconds>(now - start).count();
      } else {
        // cannot handle, rethrow exception
        throw;
      }
    }
  }
  return 0;
}

publish()

template<class AZMQMessage>

void publish ( AZMQMessage message ) const

inline

Publish the message to the multicast.

Definition at line 53 of file ZMQClient.h.

    {
      AZMQMessage::element_type::toSocket(std::move(message), m_pubSocket);
    }

reset()

void reset

(

)

Reset the sockets. ATTENTION: this does not close the sockets! Use only after forks to not clean up to times.

Definition at line 48 of file ZMQClient.cc.

{
  // Deliberately give up ownership without destructing: see the comment in terminate().
  // cppcheck-suppress-begin ignoredReturnValue
  m_context.release();
  m_subSocket.release();
  m_pubSocket.release();
  m_socket.release();
  // cppcheck-suppress-end ignoredReturnValue
}

send() [1/2]

template<class AZMQMessage>

void send ( AZMQMessage message ) const

inline

Send a message over the data socket.

Definition at line 43 of file ZMQClient.h.

    {
      AZMQMessage::element_type::toSocket(std::move(message), m_socket);
    }

send() [2/2]

void send

(

zmq::message_t &

message

)

const

Send a zmq message over the data socket. ATTENTION: we are taking ownership here!

Definition at line 116 of file ZMQClient.cc.

{
  B2ASSERT("Can only run this on started clients", m_socket);
  m_socket->send(message, zmq::send_flags::none);
}

subscribe()

void subscribe

(

EMessageTypes

messageType

)

Subscribe to the given multicast message type.

Definition at line 107 of file ZMQClient.cc.

{
  B2ASSERT("Can only run this on started clients", m_subSocket);
  char char_filter[2];
  char_filter[0] = static_cast<char>(filter);
  char_filter[1] = 0;
  m_subSocket->set(zmq::sockopt::subscribe, char_filter);
}

terminate()

void terminate

(

bool

sendGoodbye = true

)

Terminate the sockets properly.

Definition at line 18 of file ZMQClient.cc.

{
  if (m_pubSocket and sendGoodbye) {
    auto multicastMessage = ZMQMessageFactory::createMessage(EMessageTypes::c_terminateMessage, getpid());
    publish(std::move(multicastMessage));
  }
 
  // The sockets and the context are deliberately released and not destructed:
  // running the ZMQ destructors in a forked process would also tear down the
  // file descriptors still used by the parent process.
  // cppcheck-suppress-begin ignoredReturnValue
  if (m_socket) {
    m_socket->close();
    m_socket.release();
  }
  if (m_pubSocket) {
    m_pubSocket->close();
    m_pubSocket.release();
  }
  if (m_subSocket) {
    m_subSocket->close();
    m_subSocket.release();
  }
  if (m_context) {
    m_context->close();
    m_context.release();
  }
  // cppcheck-suppress-end ignoredReturnValue
}

Member Data Documentation

m_context

std::unique_ptr<zmq::context_t> m_context

private

ZMQ context.

Definition at line 99 of file ZMQClient.h.

m_pollSocketPtrList

std::vector<zmq::socket_t*> m_pollSocketPtrList

private

Will use this vector for polling.

Definition at line 102 of file ZMQClient.h.

m_pubSocket

std::unique_ptr<zmq::socket_t> m_pubSocket

private

ZMQ Pub socket.

Definition at line 105 of file ZMQClient.h.

m_socket

std::unique_ptr<zmq::socket_t> m_socket

private

ZMQ socket.

Definition at line 109 of file ZMQClient.h.

m_subSocket

std::unique_ptr<zmq::socket_t> m_subSocket

private

ZMQ sub socket.

Definition at line 107 of file ZMQClient.h.

---

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

• framework/pcore/zmq/sockets/include/ZMQClient.h
• framework/pcore/zmq/sockets/src/ZMQClient.cc