BriefServerRequestor.h

Go to the documentation of this file.

 
#pragma once
#include "NodeRequestResult.h"
#include "PeerConnectCode.h"
#include "ServerConnector.h"
#include "catapult/ionet/Node.h"
#include "catapult/ionet/PacketSocket.h"
#include "catapult/thread/Future.h"
#include "catapult/thread/IoThreadPool.h"
#include "catapult/thread/TimedCallback.h"
#include "catapult/utils/HexFormatter.h"
 
namespace catapult { namespace net {
 
    namespace detail {
        class AlwaysCompatibleResponseCompatibilityChecker {
        public:
            template<typename T>
            constexpr bool isResponseCompatible(const ionet::Node&, const T&) const {
                return true;
            }
        };
    }
 
    template<typename TRequestPolicy, typename TResponseCompatibilityChecker = detail::AlwaysCompatibleResponseCompatibilityChecker>
    class BriefServerRequestor : public std::enable_shared_from_this<BriefServerRequestor<TRequestPolicy, TResponseCompatibilityChecker>> {
    public:
        using ResponseType = typename TRequestPolicy::ResponseType;
 
        using CallbackType = consumer<NodeRequestResult, const ResponseType&>;
 
    private:
        // region NodeRequest
 
        class NodeRequest {
        public:
            NodeRequest(
                    const ionet::Node& requestNode,
                    const std::shared_ptr<thread::IoThreadPool>& pPool,
                    const TResponseCompatibilityChecker& compatibilityChecker,
                    const CallbackType& callback)
                    : m_requestNode(requestNode)
                    , m_pPool(pPool)
                    , m_compatibilityChecker(compatibilityChecker)
                    , m_callback(callback)
            {}
 
        public:
            void setTimeout(const utils::TimeSpan& timeout, const std::shared_ptr<ionet::PacketSocket>& pSocket) {
                auto pTimedCallback = thread::MakeTimedCallback(
                        m_pPool->ioContext(),
                        m_callback,
                        NodeRequestResult::Failure_Timeout,
                        ResponseType());
                pTimedCallback->setTimeout(timeout);
                pTimedCallback->setTimeoutHandler([pSocket, requestNode = m_requestNode]() {
                    pSocket->close();
                    CATAPULT_LOG(debug) << TRequestPolicy::Friendly_Name << " request connection to '" << requestNode << "' timed out";
                });
                m_callback = [pTimedCallback](auto result, const auto& response) {
                    pTimedCallback->callback(result, response);
                };
            }
 
        public:
            void complete(PeerConnectCode connectCode) {
                CATAPULT_LOG(debug)
                        << TRequestPolicy::Friendly_Name << " request connection to '" << m_requestNode
                        << "' failed: " << connectCode;
                complete(NodeRequestResult::Failure_Connection);
            }
 
            void complete(thread::future<ResponseType>&& responseFuture) {
                try {
                    const auto& response = responseFuture.get();
                    if (!m_compatibilityChecker.isResponseCompatible(m_requestNode, response))
                        return complete(NodeRequestResult::Failure_Incompatible);
 
                    complete(response);
                } catch (const catapult_runtime_error& e) {
                    CATAPULT_LOG(warning)
                            << "exception thrown during " << TRequestPolicy::Friendly_Name << " request to '"
                            << m_requestNode << "': " << e.what();
                    complete(NodeRequestResult::Failure_Interaction);
                }
            }
 
        private:
            void complete(NodeRequestResult result) {
                m_callback(result, ResponseType());
            }
 
            void complete(const ResponseType& response) {
                m_callback(NodeRequestResult::Success, response);
            }
 
        private:
            ionet::Node m_requestNode;
            std::shared_ptr<thread::IoThreadPool> m_pPool;
            TResponseCompatibilityChecker m_compatibilityChecker;
            CallbackType m_callback;
        };
 
        // endregion
 
    public:
        BriefServerRequestor(
                const std::shared_ptr<thread::IoThreadPool>& pPool,
                const crypto::KeyPair& keyPair,
                const ConnectionSettings& settings,
                const TResponseCompatibilityChecker& responseCompatibilityChecker)
                : m_pPool(pPool)
                , m_responseCompatibilityChecker(responseCompatibilityChecker)
                , m_requestTimeout(settings.Timeout)
                , m_pConnector(CreateServerConnector(pPool, keyPair, settings))
                , m_numTotalRequests(0)
                , m_numSuccessfulRequests(0)
        {}
 
    public:
        size_t numActiveConnections() const {
            return m_pConnector->numActiveConnections();
        }
 
        size_t numTotalRequests() const {
            return m_numTotalRequests;
        }
 
        size_t numSuccessfulRequests() const {
            return m_numSuccessfulRequests;
        }
 
    public:
        void beginRequest(const ionet::Node& node, const CallbackType& callback) {
            ++m_numTotalRequests;
            auto wrappedCallback = [pThis = this->shared_from_this(), callback](auto result, const auto& response) {
                if (NodeRequestResult::Success == result)
                    ++pThis->m_numSuccessfulRequests;
 
                callback(result, response);
            };
 
            auto pRequest = std::make_shared<NodeRequest>(node, m_pPool, m_responseCompatibilityChecker, wrappedCallback);
            m_pConnector->connect(node, [pRequest, requestTimeout = m_requestTimeout](auto connectCode, const auto& pSocket) {
                pRequest->setTimeout(requestTimeout, pSocket);
 
                if (PeerConnectCode::Accepted != connectCode)
                    return pRequest->complete(connectCode);
 
                TRequestPolicy::CreateFuture(*pSocket).then([pSocket, pRequest](auto&& responseFuture) {
                    pRequest->complete(std::move(responseFuture));
                });
            });
        }
 
        void shutdown() {
            m_pConnector->shutdown();
        }
 
    private:
        std::shared_ptr<thread::IoThreadPool> m_pPool;
        TResponseCompatibilityChecker m_responseCompatibilityChecker;
        utils::TimeSpan m_requestTimeout;
        std::shared_ptr<ServerConnector> m_pConnector;
 
        std::atomic<size_t> m_numTotalRequests;
        std::atomic<size_t> m_numSuccessfulRequests;
    };
 
    template<typename TRequestor, typename TResult = std::pair<NodeRequestResult, typename TRequestor::ResponseType>>
    thread::future<TResult> BeginRequestFuture(TRequestor& requestor, const ionet::Node& node) {
        auto pPromise = std::make_shared<thread::promise<TResult>>();
 
        requestor.beginRequest(node, [pPromise](auto result, const auto& response) {
            pPromise->set_value(std::make_pair(result, response));
        });
 
        return pPromise->get_future();
    }
}}