client.hh

/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  You may not use this file except in compliance with the License.
 *
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
/*
 * Copyright (C) 2022 Scylladb, Ltd.
 */
 
#pragma once
 
#ifndef SEASTAR_MODULE
#include <boost/intrusive/list.hpp>
#endif
#include <seastar/net/api.hh>
#include <seastar/http/connection_factory.hh>
#include <seastar/http/reply.hh>
#include <seastar/core/condition-variable.hh>
#include <seastar/core/iostream.hh>
#include <seastar/util/modules.hh>
 
namespace bi = boost::intrusive;
 
namespace seastar {
 
SEASTAR_MODULE_EXPORT_BEGIN
 
namespace tls { class certificate_credentials; }
 
namespace http {
 
namespace experimental { class client; }
struct request;
struct reply;
 
namespace internal {
 
class client_ref {
    http::experimental::client* _c;
public:
    client_ref(http::experimental::client* c) noexcept;
    ~client_ref();
    client_ref(client_ref&& o) noexcept : _c(std::exchange(o._c, nullptr)) {}
    client_ref(const client_ref&) = delete;
};
 
}
 
namespace experimental {
 
class connection : public enable_shared_from_this<connection> {
    friend class client;
    using hook_t = bi::list_member_hook<bi::link_mode<bi::auto_unlink>>;
    using reply_ptr = std::unique_ptr<reply>;
 
    connected_socket _fd;
    input_stream<char> _read_buf;
    output_stream<char> _write_buf;
    hook_t _hook;
    future<> _closed;
    internal::client_ref _ref;
    // Client sends HTTP-1.1 version and assumes the server is 1.1-compatible
    // too and thus the connection will be persistent by default. If the server
    // responds with older version, this flag will be dropped (see recv_reply())
    bool _persistent = true;
 
public:
    connection(connected_socket&& fd, internal::client_ref cr);
 
    future<reply> make_request(request rq);
 
    input_stream<char> in(reply& rep);
 
    future<> close();
 
private:
    future<reply_ptr> do_make_request(request& rq);
    void setup_request(request& rq);
    future<> send_request_head(const request& rq);
    future<reply_ptr> maybe_wait_for_continue(const request& req);
    future<> write_body(const request& rq);
    future<reply_ptr> recv_reply();
 
    void shutdown() noexcept;
};
 
class client {
public:
    using reply_handler = noncopyable_function<future<>(const reply&, input_stream<char>&& body)>;
    using retry_requests = bool_class<struct retry_requests_tag>;
 
private:
    friend class http::internal::client_ref;
    using connections_list_t = bi::list<connection, bi::member_hook<connection, typename connection::hook_t, &connection::_hook>, bi::constant_time_size<false>>;
    static constexpr unsigned default_max_connections = 100;
 
    std::unique_ptr<connection_factory> _new_connections;
    unsigned _nr_connections = 0;
    unsigned _max_connections;
    unsigned long _total_new_connections = 0;
    const retry_requests _retry;
    condition_variable _wait_con;
    connections_list_t _pool;
 
    using connection_ptr = seastar::shared_ptr<connection>;
 
    future<connection_ptr> get_connection(abort_source* as);
    future<connection_ptr> make_connection(abort_source* as);
    future<> put_connection(connection_ptr con);
    future<> shrink_connections();
 
    template <std::invocable<connection&> Fn>
    auto with_connection(Fn&& fn, abort_source*);
 
    template <typename Fn>
    requires std::invocable<Fn, connection&>
    auto with_new_connection(Fn&& fn, abort_source*);
 
    future<> do_make_request(connection& con, request& req, reply_handler& handle, abort_source*, std::optional<reply::status_type> expected);
 
public:
    explicit client(socket_address addr);
 
    client(socket_address addr, shared_ptr<tls::certificate_credentials> creds, sstring host = {});
 
    explicit client(std::unique_ptr<connection_factory> f, unsigned max_connections = default_max_connections, retry_requests retry = retry_requests::no);
 
    future<> make_request(request&& req, reply_handler&& handle, std::optional<reply::status_type>&& expected = std::nullopt, abort_source* as = nullptr);
 
    future<> make_request(request& req, reply_handler& handle, std::optional<reply::status_type> expected = std::nullopt, abort_source* as = nullptr);
 
    future<> set_maximum_connections(unsigned nr);
 
    future<> close();
 
    unsigned connections_nr() const noexcept {
        return _nr_connections;
    }
 
    unsigned idle_connections_nr() const noexcept {
        return _pool.size();
    }
 
    unsigned long total_new_connections_nr() const noexcept {
        return _total_new_connections;
    }
};
 
} // experimental namespace
 
} // http namespace
 
SEASTAR_MODULE_EXPORT_END
} // seastar namespace