dunedaq::daphnemodules::DaphneV3Interface Class Reference

#include <DaphneV3Interface.hpp>

Public Member Functions
	DaphneV3Interface (std::string address, std::string routing, std::chrono::milliseconds timeout=std::chrono::milliseconds(500))
	~DaphneV3Interface ()
	DaphneV3Interface (const DaphneV3Interface &)=delete
DaphneV3Interface &	operator= (const DaphneV3Interface &)=delete
	DaphneV3Interface (DaphneV3Interface &&)=delete
DaphneV3Interface &	operator= (DaphneV3Interface &&)=delete
daphne::ControlEnvelopeV2	send (std::string &&message, daphne::MessageTypeV2)
template<class T>
T	send (std::string &&message, daphne::MessageTypeV2 sent_type, daphne::MessageTypeV2 received_type)
bool	validate_connection ()

Protected Member Functions
void	_send (std::string &&message, daphne::MessageTypeV2)
daphne::ControlEnvelopeV2	_receive ()
void	close ()

Private Attributes
zmq::context_t	m_context
zmq::socket_t	m_socket
std::mutex	m_access_mutex
std::string	m_connection
std::chrono::milliseconds	m_timeout {1000}
uint64_t	m_message_counter = 0

Static Private Attributes
static const size_t	s_default_control_port = 40001

Detailed Description

Definition at line 88 of file DaphneV3Interface.hpp.

Constructor & Destructor Documentation

DaphneV3Interface() [1/3]

DaphneV3Interface::DaphneV3Interface	(	std::string	address,
		std::string	routing,
		std::chrono::milliseconds	timeout = std::chrono::milliseconds(500) )

Definition at line 24 of file DaphneV3Interface.cpp.

  : m_context(1)
  , m_socket(m_context, zmq::socket_type::dealer)
  , m_timeout(timeout) {
 
 
  m_socket.set(zmq::sockopt::routing_id, routing);
  auto value = (int) m_timeout.count();  // NOLINT
  TLOG() << routing << " timeout set to " << value << " ms";
  m_socket.set(zmq::sockopt::rcvtimeo, value);
  m_socket.set(zmq::sockopt::sndtimeo, value);
  m_socket.set(zmq::sockopt::immediate, 1); // Don't queue messages to incomplete connections
  
  // find out if the address has a port with a regex
  static const std::regex ip_with_port(R"(^([^\/\s:]+)(?::(\d{1,5}))?$)");
  std::smatch string_values; 
  if (! std::regex_match( address, string_values, ip_with_port ) ) {
    throw InvalidIPAddress(ERS_HERE, address);
  }
 
  m_connection = string_values[2].matched ?
    fmt::format("tcp://{}", address) :
    fmt::format("tcp://{}:{}", address, s_default_control_port) ;
  TLOG() << "Connecting to: " << m_connection;
  
  m_socket.connect(m_connection);
 
  auto add = string_values[1];
  auto port = string_values[2].matched ? std::stoi(string_values[2]) : s_default_control_port;
      
  try {
    if ( ! validate_connection() ) {
      auto add = string_values[1];
      auto port = string_values[2].matched ? std::stoi(string_values[2]) : s_default_control_port;
      throw FailedPing(ERS_HERE, add, port );
    }
  } catch ( const ers::Issue & e ) {
    throw FailedPing(ERS_HERE, add, port, e );
  }
    
}

~DaphneV3Interface()

dunedaq::daphnemodules::DaphneV3Interface::~DaphneV3Interface ( )

inline

Definition at line 96 of file DaphneV3Interface.hpp.

{ close(); }

DaphneV3Interface() [2/3]

dunedaq::daphnemodules::DaphneV3Interface::DaphneV3Interface ( const DaphneV3Interface & )

delete

DaphneV3Interface() [3/3]

dunedaq::daphnemodules::DaphneV3Interface::DaphneV3Interface ( DaphneV3Interface && )

delete

Member Function Documentation

_receive()

ControlEnvelopeV2 DaphneV3Interface::_receive ( )

protected

Definition at line 109 of file DaphneV3Interface.cpp.

                                              {
 
  zmq::message_t reply;
  if (! m_socket.recv(reply, zmq::recv_flags::none)) {
    // timeout or EAGAIN
    throw FailedReceive(ERS_HERE, m_connection);
  }
  if (reply.size() <= 0) {
    throw EmptyPayload(ERS_HERE, m_connection);
  }
  
  ControlEnvelopeV2 rep;
  if (!rep.ParseFromArray(reply.data(), static_cast<int>(reply.size()) )) {
    throw FailedDecoding(ERS_HERE, rep.GetTypeName(), reply.to_string());
  }
  
  if (rep.version() != 2 || rep.dir() != DIR_RESPONSE) {
    ers::warning(UnexpectedDirection(ERS_HERE, rep.version(), Direction_Name(rep.dir()), reply.to_string() )); 
  }
 
  return rep;
}

_send()

void DaphneV3Interface::_send ( std::string && message, daphne::MessageTypeV2 type )

protected

Definition at line 88 of file DaphneV3Interface.cpp.

                                                                             {
 
  ControlEnvelopeV2 env;
  env.set_version(2);
  env.set_dir(DIR_REQUEST);
 
  env.set_type(type);
  env.set_payload(message);
 
  // additional information
  env.set_msg_id( m_message_counter++ );
  env.set_timestamp_ns(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now().time_since_epoch()).count());
 
  
  std::string bytes = env.SerializeAsString();
 
  if (! m_socket.send(zmq::buffer(bytes),  zmq::send_flags::none)) {
    throw FailedSend(ERS_HERE, MessageTypeV2_Name(type) );
  }
}

close()

void DaphneV3Interface::close ( )

protected

Definition at line 69 of file DaphneV3Interface.cpp.

                              {
 
  const std::lock_guard<std::mutex> lock(m_access_mutex);
  m_socket.set(zmq::sockopt::linger, 0);
  m_socket.close();
  
}

operator=() [1/2]

DaphneV3Interface & dunedaq::daphnemodules::DaphneV3Interface::operator= ( const DaphneV3Interface & )

delete

operator=() [2/2]

DaphneV3Interface & dunedaq::daphnemodules::DaphneV3Interface::operator= ( DaphneV3Interface && )

delete

send() [1/2]

template<class T>

T dunedaq::daphnemodules::DaphneV3Interface::send	(	std::string &&	message,
		daphne::MessageTypeV2	sent_type,
		daphne::MessageTypeV2	received_type )

Implementations of DaphneV3Interface's functions

This is part of the DUNE DAQ Software Suite, copyright 2020.
Licensing/copyright details are in the COPYING file that you should have

Definition at line 11 of file DaphneV3Interface.hxx.

                                                                                                                                      {
 
  std::unique_lock<std::mutex> lock(m_access_mutex);
 
  _send(std::move(message), sent_type);
 
  auto ret = _receive();
 
  lock.unlock();
 
  const auto ty = ret.type();
  T out;
  if ( ty != received_type ) {
    throw FailedDecoding(ERS_HERE, out.GetTypeName(), ret.payload(),
             TypeMismatch(ERS_HERE, MessageTypeV2_Name(ty), MessageTypeV2_Name(received_type)) );
  }
 
  if (!out.ParseFromString(ret.payload())) {
    throw FailedDecoding(ERS_HERE, out.GetTypeName(), ret.payload());
  }
 
  return out;
}

send() [2/2]

ControlEnvelopeV2 DaphneV3Interface::send	(	std::string &&	message,
		daphne::MessageTypeV2	type )

Definition at line 78 of file DaphneV3Interface.cpp.

                                                                                         {
 
  const std::lock_guard<std::mutex> lock(m_access_mutex);
  
  _send(std::move(message), type);
 
  return _receive();
}

validate_connection()

bool DaphneV3Interface::validate_connection

(

)

Definition at line 133 of file DaphneV3Interface.cpp.

{
  static const uint64_t good_value = 0xdeadbeef;  // NOLINT
 
  TestRegRequest req; // empty
  auto reply = send<TestRegResponse>( req.SerializeAsString(),
                      MessageTypeV2::MT2_READ_TEST_REG_REQ,
                      MessageTypeV2::MT2_READ_TEST_REG_RESP );
  
  return reply.value() == good_value;
}

Member Data Documentation

m_access_mutex

std::mutex dunedaq::daphnemodules::DaphneV3Interface::m_access_mutex

mutableprivate

Definition at line 125 of file DaphneV3Interface.hpp.

m_connection

std::string dunedaq::daphnemodules::DaphneV3Interface::m_connection

private

Definition at line 127 of file DaphneV3Interface.hpp.

m_context

zmq::context_t dunedaq::daphnemodules::DaphneV3Interface::m_context

private

Definition at line 122 of file DaphneV3Interface.hpp.

m_message_counter

uint64_t dunedaq::daphnemodules::DaphneV3Interface::m_message_counter = 0

private

Definition at line 131 of file DaphneV3Interface.hpp.

m_socket

zmq::socket_t dunedaq::daphnemodules::DaphneV3Interface::m_socket

private

Definition at line 124 of file DaphneV3Interface.hpp.

m_timeout

std::chrono::milliseconds dunedaq::daphnemodules::DaphneV3Interface::m_timeout {1000}

private

Definition at line 129 of file DaphneV3Interface.hpp.

{1000};

s_default_control_port

const size_t dunedaq::daphnemodules::DaphneV3Interface::s_default_control_port = 40001

inlinestaticprivate

Definition at line 133 of file DaphneV3Interface.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/daphnemodules/src/DaphneV3Interface.hpp
• /github/workspace/dunedaq/sourcecode/daphnemodules/src/DaphneV3Interface.cpp
• /github/workspace/dunedaq/sourcecode/daphnemodules/src/DaphneV3Interface.hxx