dunedaq::iomanager::ConfigClient Class Reference

#include <ConfigClient.hpp>

Public Member Functions
	ConfigClient (const std::string &server, const std::string &port, const std::string &session_name, std::chrono::milliseconds publish_interval)
	~ConfigClient ()
ConnectionResponse	resolve_connection (const ConnectionRequest &query, std::string session="")
void	publish (ConnectionRegistration const &connection)
void	publish (std::vector< ConnectionRegistration > const &connections)
void	retract (const ConnectionId &connectionId)
void	retract (const std::vector< ConnectionId > &connectionIds)
void	retract ()
bool	is_connected ()

Private Member Functions
void	publish ()

Private Attributes
std::string	m_session
net::io_context	m_io_context
net::ip::basic_resolver< net::ip::tcp >::results_type	m_addr
std::mutex	m_mutex
std::set< ConnectionRegistration >	m_registered_connections
std::thread	m_thread
bool	m_active
std::atomic< bool >	m_connected { false }

Detailed Description

Definition at line 34 of file ConfigClient.hpp.

Constructor & Destructor Documentation

ConfigClient()

ConfigClient::ConfigClient	(	const std::string &	server,
		const std::string &	port,
		const std::string &	session_name,
		std::chrono::milliseconds	publish_interval )

Constructor: Starts a thread that publishes all know connection information once a second

Parameters

server	Name/address of the connection server to publish to
port	Port on the connection server to connect to
session_name	Name of the current Session
publish_interval	Time to wait between connection republish (keep-alive)

Definition at line 37 of file ConfigClient.cpp.

{
  m_session = session_name;
 
  tcp::resolver resolver(m_io_context);
  m_addr = resolver.resolve(server, port);
  m_active = true;
  m_thread = std::thread([this, publish_interval]() {
    while (m_active) {
      try {
        publish();
        m_connected = true;
        TLOG_DEBUG(TLVL_PUBLISH) << "Automatic publish complete";
      } catch (ers::Issue& ex) {
        if (m_connected)
          ers::error(ex);
        else
          TLOG() << ex;
      } catch (std::exception& ex) {
        auto ers_ex = PublishException(ERS_HERE, ex.what());
        if (m_connected)
          ers::error(ers_ex);
        else
          TLOG() << ers_ex;
      }
      std::this_thread::sleep_for(publish_interval);
    }
    retract();
    if (!m_connected) {
      ers::error(PublishException(ERS_HERE, "Publish thread was unable to publish to Connectivity Service!"));
    }
  });
}

~ConfigClient()

ConfigClient::~ConfigClient

(

)

Destructor: stops the publishing hread and retracts all published information

Definition at line 74 of file ConfigClient.cpp.

{
  m_active = false;
  if (m_thread.joinable()) {
    m_thread.join();
  }
  m_connected = false;
  retract();
}

Member Function Documentation

is_connected()

bool dunedaq::iomanager::ConfigClient::is_connected ( )

inline

Definition at line 86 of file ConfigClient.hpp.

{ return m_connected.load(); }

publish() [1/3]

void ConfigClient::publish ( )

private

Definition at line 161 of file ConfigClient.cpp.

{
  json content{ { "partition", m_session } };
  json connections = json::array();
  {
    std::lock_guard<std::mutex> lock(m_mutex);
    for (auto& entry : m_registered_connections) {
      json item = entry;
      connections.push_back(item);
    }
    if (connections.size() == 0) {
      return;
    }
  }
  content["connections"] = connections;
  http::request<http::string_body> req{ http::verb::post, "/publish", HTTP_V1_1 };
  req.set(http::field::content_type, "application/json");
  req.body() = content.dump();
  req.prepare_payload();
 
  boost::beast::tcp_stream stream(m_io_context);
  beast::error_code ec;
  try {
    stream.connect(m_addr);
    http::write(stream, req);
 
    http::response<http::string_body> response;
    boost::beast::flat_buffer buffer;
    http::read(stream, buffer, response);
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    if (response.result_int() != 200) {
      throw(FailedPublish(ERS_HERE, std::string(response.reason())));
    }
  } catch (ers::Issue const&) {
    m_connected = false;
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    throw;
  } catch (std::exception const& ex) {
    m_connected = false;
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    throw(FailedPublish(ERS_HERE, ex.what(), ex));
  }
  m_connected = true;
}

publish() [2/3]

void ConfigClient::publish

(

ConnectionRegistration const &

connection

)

Definition at line 135 of file ConfigClient.cpp.

{
  {
    std::lock_guard<std::mutex> lock(m_mutex);
    TLOG_DEBUG(TLVL_PUBLISH) << "Adding connection with UID " << connection.uid << " and URI " << connection.uri
                             << " to publish list";
 
    m_registered_connections.insert(connection);
  }
}

publish() [3/3]

void ConfigClient::publish

(

std::vector< ConnectionRegistration > const &

connections

)

Definition at line 147 of file ConfigClient.cpp.

{
  {
    std::lock_guard<std::mutex> lock(m_mutex);
    for (auto& entry : connections) {
      TLOG_DEBUG(TLVL_PUBLISH) << "Adding connection with UID " << entry.uid << " and URI " << entry.uri
                               << " to publish list";
 
      m_registered_connections.insert(entry);
    }
  }
}

resolve_connection()

ConnectionResponse ConfigClient::resolve_connection	(	const ConnectionRequest &	query,
		std::string	session = "" )

Look up a connection in the connection server and return a list of uris that correspond to connection ids that match

Parameters

query	Query string to send to the server. Query is a regular expression that can match with multiple connection ids
session	The session that the requested connection is part of

Definition at line 85 of file ConfigClient.cpp.

{
  if (session == "") {
    session = m_session;
  }
  TLOG_DEBUG(TLVL_RESOLVE) << "Getting connections matching <" << query.uid_regex << "> in session " << session;
  std::string target = "/getconnection/" + session;
  http::request<http::string_body> req{ http::verb::post, target, HTTP_V1_1 };
  req.set(http::field::content_type, "application/json");
  nlohmann::json jquery = query;
  req.body() = jquery.dump();
  req.prepare_payload();
 
  http::response<http::string_body> response;
  boost::beast::tcp_stream stream(m_io_context);
  beast::error_code ec;
  try {
    stream.connect(m_addr);
    http::write(stream, req);
 
    boost::beast::flat_buffer buffer;
    http::read(stream, buffer, response);
 
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    TLOG_DEBUG(TLVL_RESOLVE) << "get " << target << " response: " << response;
 
    if (response.result_int() != 200) {
      throw(FailedLookup(ERS_HERE, query.uid_regex, target, std::string(response.reason())));
    }
  } catch (ers::Issue const&) {
    m_connected = false;
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    throw;
  } catch (std::exception const& ex) {
    m_connected = false;
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    ers::error(FailedLookup(ERS_HERE, query.uid_regex, target, ex.what()));
    return ConnectionResponse();
  }
  m_connected = true;
  json result = json::parse(response.body());
  TLOG_DEBUG(TLVL_RESOLVE) << result.dump();
  ConnectionResponse res;
  for (auto const& item : result) {
    res.connections.emplace_back(item.get<ConnectionInfo>());
  }
  return res;
}

retract() [1/3]

void ConfigClient::retract

(

)

Retract all connection information that ew have published

Definition at line 207 of file ConfigClient.cpp.

{
  TLOG_DEBUG(TLVL_RETRACT) << "retract() called, getting connection information";
  json connections = json::array();
  {
    std::lock_guard<std::mutex> lock(m_mutex);
    for (auto& con : m_registered_connections) {
      json item;
      item["connection_id"] = con.uid;
      item["data_type"] = con.data_type;
      connections.push_back(item);
    }
    m_registered_connections.clear();
  }
  if (connections.size() > 0) {
    TLOG_DEBUG(TLVL_RETRACT) << "retract(): Retracting " << connections.size() << " connections";
    http::request<http::string_body> req{ http::verb::post, "/retract", HTTP_V1_1 };
    req.set(http::field::content_type, "application/json");
    json body{ { "partition", m_session } };
    body["connections"] = connections;
    req.body() = body.dump();
    req.prepare_payload();
 
    boost::beast::tcp_stream stream(m_io_context);
    beast::error_code ec;
    try {
      stream.connect(m_addr);
      http::write(stream, req);
      http::response<http::string_body> response;
      boost::beast::flat_buffer buffer;
      http::read(stream, buffer, response);
      stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
      if (response.result_int() != 200) {
        throw(FailedRetract(ERS_HERE, "connection Id vector", std::string(response.reason())));
      }
    } catch (ers::Issue const&) {
      m_connected = false;
      stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
      throw;
    } catch (std::exception const& ex) {
      m_connected = false;
      stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
      ers::error(FailedRetract(ERS_HERE, "connection Id vector", ex.what()));
    }
    m_connected = true;
  }
}

retract() [2/3]

void ConfigClient::retract

(

const ConnectionId &

connectionId

)

Definition at line 256 of file ConfigClient.cpp.

{
  retract(std::vector<ConnectionId>{ connectionId });
}

retract() [3/3]

void ConfigClient::retract

(

const std::vector< ConnectionId > &

connectionIds

)

Retract multiple published connections

Parameters

connectionIds

A vector of previously published connection Ids to be retracted

Definition at line 262 of file ConfigClient.cpp.

{
  http::request<http::string_body> req{ http::verb::post, "/retract", HTTP_V1_1 };
  req.set(http::field::content_type, "application/json");
 
  json connections = json::array();
  {
    std::lock_guard<std::mutex> lock(m_mutex);
    for (auto& con : connectionIds) {
      auto reg_it = m_registered_connections.begin();
      for (; reg_it != m_registered_connections.end(); ++reg_it) {
        if (con.uid == reg_it->uid && con.data_type == reg_it->data_type)
          break;
      }
      if (reg_it == m_registered_connections.end()) {
        ers::error(
          FailedRetract(ERS_HERE, con.uid + " of type " + con.data_type, "not in registered connections list"));
      } else {
        json item;
        item["connection_id"] = con.uid;
        item["data_type"] = con.data_type;
        connections.push_back(item);
        m_registered_connections.erase(reg_it);
      }
    }
  }
  json body{ { "partition", m_session } };
  body["connections"] = connections;
  req.body() = body.dump();
  req.prepare_payload();
 
  boost::beast::tcp_stream stream(m_io_context);
  beast::error_code ec;
  try {
    stream.connect(m_addr);
    http::write(stream, req);
    http::response<http::string_body> response;
    boost::beast::flat_buffer buffer;
    http::read(stream, buffer, response);
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    if (response.result_int() != 200) {
      throw(FailedRetract(ERS_HERE, "connection Id vector", std::string(response.reason())));
    }
  } catch (ers::Issue const&) {
    m_connected = false;
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    throw;
  } catch (std::exception const& ex) {
    m_connected = false;
    stream.socket().shutdown(tcp::socket::shutdown_both, ec); // NOLINT
    ers::error(FailedRetract(ERS_HERE, "connection Id vector", ex.what()));
  }
  m_connected = true;
}

Member Data Documentation

m_active

bool dunedaq::iomanager::ConfigClient::m_active

private

Definition at line 97 of file ConfigClient.hpp.

m_addr

net::ip::basic_resolver<net::ip::tcp>::results_type dunedaq::iomanager::ConfigClient::m_addr

private

Definition at line 92 of file ConfigClient.hpp.

m_connected

std::atomic<bool> dunedaq::iomanager::ConfigClient::m_connected { false }

private

Definition at line 98 of file ConfigClient.hpp.

{ false };

m_io_context

net::io_context dunedaq::iomanager::ConfigClient::m_io_context

private

Definition at line 91 of file ConfigClient.hpp.

m_mutex

std::mutex dunedaq::iomanager::ConfigClient::m_mutex

private

Definition at line 94 of file ConfigClient.hpp.

m_registered_connections

std::set<ConnectionRegistration> dunedaq::iomanager::ConfigClient::m_registered_connections

private

Definition at line 95 of file ConfigClient.hpp.

m_session

std::string dunedaq::iomanager::ConfigClient::m_session

private

Definition at line 90 of file ConfigClient.hpp.

m_thread

std::thread dunedaq::iomanager::ConfigClient::m_thread

private

Definition at line 96 of file ConfigClient.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/iomanager/include/iomanager/network/ConfigClient.hpp
• /github/workspace/dunedaq/sourcecode/iomanager/src/network/ConfigClient.cpp