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>
 
 
 
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) {
            // TODO should we skip this if m_connected == true?
            // Currently we also connect redundantly, just to be able to probe it independently of the timing of the connection callback.
 
            elt::mal::future<void> conn_future = m_client->asyncConnect();
            ::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.");
            auto future_status = conn_future.wait_for(conn_timeout_cii);
            bool success = future_status == (boost::future_status::ready); // ready, timeout, deferred
            LOG4CPLUS_TRACE(m_logger, "Requestor::Connect() returned from waiting for connection, success=" << success);
            return success;
        }
        else {
            LOG4CPLUS_DEBUG(m_logger, "Requestor::Connect() failed because of missing MAL client.");
            return false;
        }
    }
 
    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