requestor.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2020-2025 European Southern Observatory (ESO)
//
// SPDX-License-Identifier: LGPL-3.0-only

#ifndef HLCC_CPPUTIL_REQUESTOR_HPP
#define HLCC_CPPUTIL_REQUESTOR_HPP
 
 
#include <atomic>
 
#include <mal/Cii.hpp>
#include <mal/utility/future.hpp>
 
#include <rad/logger.hpp>
 
 
 
namespace hlcc::cpputil {
 

template <typename INTERFACE_TYPE>
class Requestor {
public:
    explicit Requestor(log4cplus::Logger& logger)
      : m_mutex {},
        m_logger{logger},
        m_uri{},
        m_client{}
    {
    }

    explicit Requestor(log4cplus::Logger& logger, const elt::mal::Uri& uri,
              const std::optional<elt::mal::Mal::Properties> mal_properties = {})
        : Requestor(logger) {
        SetConnectionInfo(uri, mal_properties);
    }

    void SetConnectionInfo(const elt::mal::Uri& uri,
            const std::optional<elt::mal::Mal::Properties> mal_properties = {}) {
 
        std::lock_guard lck {m_mutex};
 
        if (uri != m_uri) {
            // Close previous instance. Probably would be done automatically after being dereferenced.
            if (m_client) {
                LOG4CPLUS_DEBUG(m_logger, "SetConnectionInfo will close old client");
                m_client->close();
            }
 
            try {
                // Create the MAL client.
                // Note that (at least with ZMQ MAL), client construction immediately triggers an async connection attempt.
                m_client = elt::mal::CiiFactory::getInstance().getClient<INTERFACE_TYPE>(
                    uri, elt::mal::rr::qos::QoS::DEFAULT,
                    mal_properties ? *mal_properties : elt::mal::Mal::Properties());
                m_uri = uri;
                LOG4CPLUS_DEBUG(m_logger, "Created rr client for " << uri);
 
                // Register connection listener that logs all later connections and disconnections.
                // This registration itself is async, which means that if immediately afterwards we call Connect
                // or connect implicitly by using the interface, then the new connection may not be logged yet.
                // We copy-capture the uri instead of using m_uri because these two may diverge when switching to a new URI.
                m_client->registerConnectionListener([this, uri](bool connected) {
                    m_connected.store(connected);
                    LOG4CPLUS_DEBUG(m_logger, "Connected with " << uri << ": " << connected);
                });
            } catch (const std::exception& ex) {
                // This happens, for example, when Nomad does not resolve the address and gives us a URI such as "zpb.rr:///TrackCmds".
                // We just log the error and leave m_client empty or closed. A following call to Connect will fail.
                LOG4CPLUS_WARN(m_logger, "Failed to create rr client for " << uri << ": " << ex.what());
            }
        }
    }

    std::shared_ptr<INTERFACE_TYPE>& GetInterface() {
        std::lock_guard lck {m_mutex};
        if (!m_client) {
            LOG4CPLUS_DEBUG(m_logger, "Requestor::GetInterface() called without having an interface client. The user must call SetConnectionInfo first.");
            // still we return the empty pointer.
        }
        return m_client;
    }

    bool Connect(std::chrono::seconds conn_timeout) {
        std::lock_guard lck {m_mutex};
        if (!m_client) {
            LOG4CPLUS_DEBUG(m_logger, "Requestor::Connect() failed because of missing MAL client.");
            return false;
        }
        if (m_connected) {
            LOG4CPLUS_DEBUG(m_logger, "Requestor::Connect() skipped because it is already connected.");
            return true;
        }
 
        elt::mal::future<void> conn_future = {};
        try {
            conn_future = m_client->asyncConnect();
        } catch (const elt::mal::MalException & ex) {
            // MAL header: if connect was called with already set promise
            // Not sure how this would happen. Perhaps through a delay in callback update of m_connected?
            LOG4CPLUS_WARN(m_logger, "Requestor::Connect() ClientAsync::asyncConnect call failed: " << ex.what());
            return false;
        }
 
        ::boost::chrono::seconds conn_timeout_cii{conn_timeout.count()};
        LOG4CPLUS_TRACE(m_logger, "Requestor::Connect() called asyncConnect() and will wait max " << conn_timeout_cii.count() << " s.");
        bool success = false;
        try {
            auto future_status = conn_future.wait_for(conn_timeout_cii);
            success = (future_status == boost::future_status::ready); // ready, timeout, deferred
            LOG4CPLUS_TRACE(m_logger, "Requestor::Connect() returned from waiting for connection, success=" << success);
        } catch (std::exception& ex) {
            LOG4CPLUS_INFO(m_logger, "Failed to wait for connection: " << ex.what());
        } catch (...) {
            LOG4CPLUS_INFO(m_logger, "Failed to wait for connection. Non-std ex.");
        }
        m_connected = success; // registerConnectionListener has a delay, thus we set our flag directly
        return success;
    }
 
    Requestor(const Requestor&) = delete;
    Requestor& operator=(const Requestor&) = delete;
 
private:
    mutable std::recursive_mutex m_mutex;
    log4cplus::Logger& m_logger;
    elt::mal::Uri m_uri;
    std::shared_ptr<INTERFACE_TYPE> m_client;  // Share pointer to MAL Client

    std::atomic<bool> m_connected{false};
};
 
 
 
}  // namespace hlcc::cpputil
 
#endif  // HLCC_CPPUTIL_REQUESTOR_HPP