lease_query_connection.h

Go to the documentation of this file.

// Copyright (C) 2022-2025 Internet Systems Consortium, Inc. ("ISC")
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
#ifndef LEASE_QUERY_CONNECTION_H
#define LEASE_QUERY_CONNECTION_H
 
#include <config.h>
#include <asiolink/asio_wrapper.h>
#include <asiolink/io_service.h>
#include <dhcp/pkt6.h>
#include <tcp/tcp_listener.h>
#include <tcp/tcp_stream_msg.h>
#include <bulk_lease_query.h>
 
#include <boost/multi_index/indexed_by.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/mem_fun.hpp>
#include <boost/multi_index/sequenced_index.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/tag.hpp>
 
namespace isc {
namespace lease_query {

template <class QueryType>
class XidQueue {
public:
    typedef boost::shared_ptr<QueryType> QueryPtrType;

    typedef boost::multi_index_container<
        // It stores pointers to BulkLeaseQuery.
        QueryPtrType,
        boost::multi_index::indexed_by<
            boost::multi_index::sequenced<>,
            boost::multi_index::ordered_unique<
                boost::multi_index::const_mem_fun<QueryType,
                                                  Xid,
                                                  &QueryType::getXid>
            >
        >
    > XidQueueContainer;

    void add(QueryPtrType query) {
        std::lock_guard<std::mutex> lck(mutex_);
        queries_.push_back(query);
    }

    QueryPtrType pop() {
        QueryPtrType query;
        {
            std::lock_guard<std::mutex> lck(mutex_);
            if (!queries_.empty()) {
                query = queries_.front();
                queries_.pop_front();
            }
        }
        return (query);
    }

    bool empty() const {
        std::lock_guard<std::mutex> lck(mutex_);
        return (queries_.empty());
    }

    size_t size() const {
        std::lock_guard<std::mutex> lck(mutex_);
        return (queries_.size());
    }

    QueryPtrType find(const Xid& xid) const {
        std::lock_guard<std::mutex> lck(mutex_);
        return (findInternal(xid));
    }

    void remove(const Xid& xid) {
        auto& xid_index = queries_.template get<1>();
        auto query = xid_index.find(xid);
        if (query != xid_index.end()) {
            xid_index.erase(query);
        }
    }

    void clear() {
        std::lock_guard<std::mutex> lck(mutex_);
        queries_.clear();
    }
 
private:
    QueryPtrType findInternal(const Xid& xid) const {
        auto const& xid_index = queries_.template get<1>();
        auto const query = xid_index.find(xid);
        if (query != xid_index.end()) {
            return (*query);
        }
        return (QueryPtrType());
    }

    XidQueueContainer queries_;

    mutable std::mutex mutex_;
};

class LeaseQueryConnection;

typedef boost::shared_ptr<LeaseQueryConnection> LeaseQueryConnectionPtr;

typedef boost::weak_ptr<LeaseQueryConnection> LeaseQueryConnectionWPtr;

class LeaseQueryConnection : public tcp::TcpConnection {
public:
    LeaseQueryConnection(const asiolink::IOServicePtr& io_service,
                         const tcp::TcpConnectionAcceptorPtr& acceptor,
                         const asiolink::TlsContextPtr& tls_context,
                         tcp::TcpConnectionPool& connection_pool,
                         const tcp::TcpConnectionAcceptorCallback& acceptor_callback,
                         const tcp::TcpConnectionFilterCallback& filter_callback,
                         const long idle_timeout,
                         const uint16_t family,
                         const size_t max_concurrent_queries,
                         const size_t read_max = 32768);

    virtual ~LeaseQueryConnection() {
    }

    virtual void shutdown();

    virtual void close();

    virtual void stopThisConnection();

    bool isStopping() const {
        return (stopping_);
    }

    bool canSend() const {
        return (can_send_);
    }

    virtual tcp::TcpRequestPtr createRequest();

    virtual void sendNextResponse();

    static void doSendNextResponse(LeaseQueryConnectionWPtr wptr) {
        LeaseQueryConnectionPtr ptr = wptr.lock();
        if (!ptr) {
            return;
        }
        ptr->sendNextResponse();
    }

    tcp::TcpResponsePtr makeTcpResponse(BlqResponsePtr blq_response) const;

    virtual bool responseSent(tcp::TcpResponsePtr response);

    virtual void requestReceived(tcp::TcpRequestPtr request);

    BlqQueryPtr unpackQuery4(const uint8_t* buffer, size_t length) const;

    BlqQueryPtr unpackQuery6(const uint8_t* buffer, size_t length) const;

    void startQuery(BlqQueryPtr query_msg);

    void addRunningQuery(BulkLeaseQueryPtr query) {
        running_queries_.add(query);
    }

    size_t getNumRunningQueries() const {
        return (running_queries_.size());
    }

    void removeRunningQuery(const Xid& xid) {
        running_queries_.remove(xid);
    }

    // query list or waiting query queue, False otherwise.
    bool findQuery(const Xid& xid) const {
        return (running_queries_.find(xid) || pending_queries_.find(xid));
    }

    void addPendingQuery(BlqQueryPtr query) {
        pending_queries_.add(query);
    }

    bool noPendingQuery() const {
        return (pending_queries_.empty());
    }

    BlqQueryPtr popPendingQuery();

    virtual void processNextQuery();

    virtual bool pushToSend(BlqResponsePtr response);

    static bool doPushToSend(LeaseQueryConnectionWPtr wptr,
                             BlqResponsePtr response) {
        LeaseQueryConnectionPtr ptr = wptr.lock();
        if (!ptr) {
            return (false);
        }
        return (ptr->pushToSend(response));
    }

    virtual void post(const BlqPostCbArg& callback);

    static void doPost(LeaseQueryConnectionWPtr wptr,
                       const BlqPostCbArg& callback) {
        LeaseQueryConnectionPtr ptr = wptr.lock();
        if (!ptr) {
            return;
        }
        ptr->post(callback);
    }

    virtual void queryComplete(const Xid& xid);

    static void doQueryComplete(LeaseQueryConnectionWPtr wptr, const Xid& xid) {
        LeaseQueryConnectionPtr ptr = wptr.lock();
        if (!ptr) {
            return;
        }
        ptr->queryComplete(xid);
    }

    size_t getNumResponses() const;

    asiolink::IOAddress getRequesterAddress() const;

    using tcp::TcpConnection::getRemoteEndpointAddressAsText;
 
protected:
    uint16_t family_;

    asiolink::IOServicePtr io_service_;

    bool stopping_;

    bool can_send_;

    XidQueue<BulkLeaseQuery> running_queries_;

    XidQueue<BlqMsg> pending_queries_;

    BlqResponseList responses_;

    mutable std::mutex responses_mutex_;

    BlqResponsePtr response_to_send_;

    size_t max_concurrent_queries_;
};
 
} // end of namespace isc::lease_query
} // end of namespace isc
 
#endif // LEASE_QUERY_CONNECTION_H