DaphneV3Interface.cpp

Go to the documentation of this file.

 
#include "DaphneV3Interface.hpp"
 
#include "logging/Logging.hpp"
#include <fmt/format.h>
 
#include <regex>
#include <string>
#include <utility>
 
using namespace dunedaq::daphnemodules;
using namespace daphne;
 
DaphneV3Interface::DaphneV3Interface( std::string address,
                      std::string routing, 
                      std::chrono::milliseconds timeout)
  : 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 );
  }
    
}
 
 
void DaphneV3Interface::close() {
 
  const std::lock_guard<std::mutex> lock(m_access_mutex);
  m_socket.set(zmq::sockopt::linger, 0);
  m_socket.close();
  
}
 
 
ControlEnvelopeV2 DaphneV3Interface::send( std::string && message, daphne::MessageTypeV2 type) {
 
  const std::lock_guard<std::mutex> lock(m_access_mutex);
  
  _send(std::move(message), type);
 
  return _receive();
}
 
 
void DaphneV3Interface::_send( std::string && message, daphne::MessageTypeV2 type) {
 
  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) );
  }
}
 
ControlEnvelopeV2 DaphneV3Interface::_receive() {
 
  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;
}
 
 
bool DaphneV3Interface::validate_connection()
{
  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;
}