Line data    Source code

       1              : /**
       2              :  * @file DaphneV3Interface.cpp
       3              :  *
       4              :  * Implementations of DaphneV3Interface's functions
       5              :  *
       6              :  * This is part of the DUNE DAQ Software Suite, copyright 2020.
       7              :  * Licensing/copyright details are in the COPYING file that you should have
       8              :  * received with this code.
       9              :  *
      10              :  */
      11              : 
      12              : #include "DaphneV3Interface.hpp"
      13              : 
      14              : #include "logging/Logging.hpp"
      15              : #include <fmt/format.h>
      16              : 
      17              : #include <regex>
      18              : #include <string>
      19              : #include <utility>
      20              : 
      21              : using namespace dunedaq::daphnemodules;
      22              : using namespace daphne;
      23              : 
      24            4 : DaphneV3Interface::DaphneV3Interface(std::string address, std::string routing, std::chrono::milliseconds timeout)
      25            4 :   : m_context(1)
      26            4 :   , m_socket(m_context, zmq::socket_type::dealer)
      27            4 :   , m_timeout(timeout)
      28              : {
      29              : 
      30            4 :   m_socket.set(zmq::sockopt::routing_id, routing);
      31            4 :   auto value = (int)m_timeout.count(); // NOLINT
      32            8 :   TLOG() << routing << " timeout set to " << value << " ms";
      33            4 :   m_socket.set(zmq::sockopt::rcvtimeo, value);
      34            4 :   m_socket.set(zmq::sockopt::sndtimeo, value);
      35            4 :   m_socket.set(zmq::sockopt::immediate, 1); // Don't queue messages to incomplete connections
      36              : 
      37              :   // find out if the address has a port with a regex
      38            4 :   static const std::regex ip_with_port(R"(^([^\/\s:]+)(?::(\d{1,5}))?$)");
      39            4 :   std::smatch string_values;
      40            4 :   if (!std::regex_match(address, string_values, ip_with_port)) {
      41            2 :     throw InvalidIPAddress(ERS_HERE, address);
      42              :   }
      43              : 
      44            6 :   m_connection = string_values[2].matched ? fmt::format("tcp://{}", address)
      45            2 :                                           : fmt::format("tcp://{}:{}", address, s_default_control_port);
      46            4 :   TLOG() << "Connecting to: " << m_connection;
      47              : 
      48            2 :   m_socket.connect(m_connection);
      49              : 
      50            2 :   auto add = string_values[1];
      51            2 :   auto port = string_values[2].matched ? std::stoi(string_values[2]) : s_default_control_port;
      52              : 
      53            2 :   try {
      54            2 :     if (!validate_connection()) {
      55            0 :       auto add = string_values[1];
      56            0 :       auto port = string_values[2].matched ? std::stoi(string_values[2]) : s_default_control_port;
      57            0 :       throw FailedPing(ERS_HERE, add, port);
      58              :     }
      59            2 :   } catch (const ers::Issue& e) {
      60            2 :     throw FailedPing(ERS_HERE, add, port, e);
      61            2 :   }
      62           16 : }
      63              : 
      64              : void
      65            0 : DaphneV3Interface::close()
      66              : {
      67              : 
      68            0 :   const std::lock_guard<std::mutex> lock(m_access_mutex);
      69            0 :   m_socket.set(zmq::sockopt::linger, 0);
      70            0 :   m_socket.close();
      71            0 : }
      72              : 
      73              : ControlEnvelopeV2
      74            0 : DaphneV3Interface::send(std::string&& message, daphne::MessageTypeV2 type)
      75              : {
      76              : 
      77            0 :   const std::lock_guard<std::mutex> lock(m_access_mutex);
      78              : 
      79            0 :   _send(std::move(message), type);
      80              : 
      81            0 :   return _receive();
      82            0 : }
      83              : 
      84              : void
      85            2 : DaphneV3Interface::_send(std::string&& message, daphne::MessageTypeV2 type)
      86              : {
      87              : 
      88            2 :   ControlEnvelopeV2 env;
      89            2 :   env.set_version(2);
      90            2 :   env.set_dir(DIR_REQUEST);
      91              : 
      92            2 :   env.set_type(type);
      93            2 :   env.set_payload(message);
      94              : 
      95              :   // additional information
      96            2 :   env.set_msg_id(m_message_counter++);
      97            4 :   env.set_timestamp_ns(
      98            2 :     std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now().time_since_epoch()).count());
      99              : 
     100            2 :   std::string bytes = env.SerializeAsString();
     101              : 
     102            2 :   if (!m_socket.send(zmq::buffer(bytes), zmq::send_flags::none)) {
     103            2 :     throw FailedSend(ERS_HERE, MessageTypeV2_Name(type));
     104              :   }
     105            4 : }
     106              : 
     107              : ControlEnvelopeV2
     108            0 : DaphneV3Interface::_receive()
     109              : {
     110              : 
     111            0 :   zmq::message_t reply;
     112            0 :   if (!m_socket.recv(reply, zmq::recv_flags::none)) {
     113              :     // timeout or EAGAIN
     114            0 :     throw FailedReceive(ERS_HERE, m_connection);
     115              :   }
     116            0 :   if (reply.size() <= 0) {
     117            0 :     throw EmptyPayload(ERS_HERE, m_connection);
     118              :   }
     119              : 
     120            0 :   ControlEnvelopeV2 rep;
     121            0 :   if (!rep.ParseFromArray(reply.data(), static_cast<int>(reply.size()))) {
     122            0 :     throw FailedDecoding(ERS_HERE, rep.GetTypeName(), reply.to_string());
     123              :   }
     124              : 
     125            0 :   if (rep.version() != 2 || rep.dir() != DIR_RESPONSE) {
     126            0 :     ers::warning(UnexpectedDirection(ERS_HERE, rep.version(), Direction_Name(rep.dir()), reply.to_string()));
     127              :   }
     128              : 
     129            0 :   return rep;
     130            0 : }
     131              : 
     132              : bool
     133            2 : DaphneV3Interface::validate_connection()
     134              : {
     135            2 :   static const uint64_t good_value = 0xdeadbeef; // NOLINT
     136              : 
     137            2 :   TestRegRequest req; // empty
     138            2 :   auto reply = send<TestRegResponse>(
     139            2 :     req.SerializeAsString(), MessageTypeV2::MT2_READ_TEST_REG_REQ, MessageTypeV2::MT2_READ_TEST_REG_RESP);
     140              : 
     141            0 :   return reply.value() == good_value;
     142            2 : }