connection_pool.h

#pragma once
 
#include <ozo/connection_info.h>
#include <ozo/transaction_status.h>
#include <ozo/asio.h>
#include <ozo/connector.h>
#include <ozo/core/thread_safety.h>
#include <ozo/detail/connection_pool.h>
 
namespace ozo {
 
struct connection_pool_config {
    std::size_t capacity = 10; 
    std::size_t queue_capacity = 128; 
    time_traits::duration idle_timeout = std::chrono::seconds(60); 
    time_traits::duration lifespan = std::chrono::hours(24); 
};
 
struct /*[[deprecated]]*/ connection_pool_timeouts {
    time_traits::duration connect = std::chrono::seconds(10); 
    time_traits::duration queue = std::chrono::seconds(10); 
};
 
template <typename Rep>
struct connection_traits {
    using native_handle_type = typename Rep::native_handle_type; 
    using oid_map_type = typename Rep::oid_map_type; 
    using error_context_type = typename Rep::error_context_type; 
    using statistics_type = typename Rep::statistics_type; 
};
 
template <typename OidMap, typename Statistics = none_t>
class connection_rep {
public:
    using oid_map_type = OidMap;
    using native_handle_type = typename ozo::pg::conn::pointer;
    using statistics_type = Statistics;
    using error_context_type = std::string;
 
    const ozo::pg::conn& safe_native_handle() const & {return safe_handle_;}
    ozo::pg::conn& safe_native_handle() & {return safe_handle_;}
 
    const oid_map_type& oid_map() const & {return oid_map_;}
    oid_map_type& oid_map() & {return oid_map_;}
 
    const auto& statistics() const & {return statistics_;}
 
    template <typename Key, typename Value>
    void update_statistics(const Key&, Value&&) noexcept {
        static_assert(std::is_void_v<Key>, "update_statistics is not supperted");
    }
 
    const error_context_type& get_error_context() const noexcept {
        return error_context_;
    }
    void set_error_context(error_context_type v) {
        error_context_ = std::move(v);
    }
 
    connection_rep(
        ozo::pg::conn&& safe_handle,
        OidMap oid_map = OidMap{},
        error_context_type error_context = {},
        Statistics statistics = Statistics{})
    : safe_handle_(std::move(safe_handle)),
      oid_map_(std::move(oid_map)),
      error_context_(std::move(error_context)),
      statistics_(std::move(statistics)) {}
private:
    ozo::pg::conn safe_handle_;
    oid_map_type oid_map_;
    error_context_type error_context_;
    statistics_type statistics_;
};
 
template <typename Rep, typename Executor = asio::io_context::executor_type>
class pooled_connection {
public:
    using rep_type = Rep; 
    using native_handle_type = typename connection_traits<rep_type>::native_handle_type; 
    using oid_map_type = typename connection_traits<rep_type>::oid_map_type; 
    using error_context_type = typename connection_traits<rep_type>::error_context_type; 
    using statistics_type = typename connection_traits<rep_type>::statistics_type; 
    using executor_type = Executor; 
 
    pooled_connection(const Executor& ex, Rep&& rep);
 
    native_handle_type native_handle() const noexcept;
 
    const oid_map_type& oid_map() const noexcept { return ozo::unwrap(rep_).oid_map();}
 
    template <typename Key, typename Value>
    void update_statistics(const Key& key, Value&& v) noexcept {
        ozo::unwrap(rep_).update_statistics(key, std::forward<Value(v)>);
    }
    const statistics_type& statistics() const noexcept { return ozo::unwrap(rep_).statistics();}
 
    const error_context_type& get_error_context() const noexcept {
        return ozo::unwrap(rep_).get_error_context();
    }
 
    void set_error_context(error_context_type v = {}) {
        ozo::unwrap(rep_).set_error_context(std::move(v));
    }
 
    executor_type get_executor() const noexcept { return ex_; }
 
    template <typename WaitHandler>
    void async_wait_write(WaitHandler&& handler);
 
    template <typename WaitHandler>
    void async_wait_read(WaitHandler&& handler);
 
    error_code close() noexcept;
 
    void cancel() noexcept;
 
    bool is_bad() const noexcept;
 
    operator bool () const noexcept { return !is_bad();}
 
    bool is_open() const noexcept { return native_handle() != nullptr;}
 
    ~pooled_connection();
private:
    using stream_type = typename detail::connection_stream<executor_type>::type;
 
    rep_type rep_;
    executor_type ex_;
    stream_type stream_;
};
 
template <typename ...Ts>
struct is_connection<pooled_connection<Ts...>> : std::true_type {};
 
template <typename T>
struct connection_traits<yamail::resource_pool::handle<T>> :
    connection_traits<typename yamail::resource_pool::handle<T>::value_type> {};
 
template <typename Source, typename ThreadSafety = std::decay_t<decltype(thread_safe)>>
class connection_pool {
    static_assert(ConnectionSource<Source>, "should model ConnectionSource concept");
 
public:
    using connection_rep_type = ozo::connection_rep<typename ozo::unwrap_type<ozo::connection_type<Source>>::oid_map_type>;
 
    using impl_type = detail::get_connection_pool_impl_t<connection_rep_type, ThreadSafety>;
    connection_pool(Source source, const connection_pool_config& config, const ThreadSafety& /*thread_safety*/ = ThreadSafety{})
    : impl_(config.capacity, config.queue_capacity, config.idle_timeout, config.lifespan),
      source_(std::move(source)) {}
 
    using connection_type = std::shared_ptr<pooled_connection<yamail::resource_pool::handle<connection_rep_type>>>;
 
    template <typename TimeConstraint, typename Handler>
    void operator ()(io_context& io, TimeConstraint t, Handler&& handler);
 
    auto stats() const {
        return impl_.stats();
    }
 
    auto operator [](io_context& io) {
        return connection_provider(*this, io);
    }
 
private:
 
    auto queue_timeout(time_traits::time_point at) const {
        return time_left(at);
    }
 
    auto queue_timeout(time_traits::duration t) const {
        return t;
    }
 
    auto queue_timeout(none_t) const {
        return time_traits::duration(0);
    }
 
    impl_type impl_;
    Source source_;
};
 
//[[DEPRECATED]] for backward compatibility only
template <typename ...Ts>
[[deprecated]] auto make_connector(connection_pool<Ts...>& source, io_context& io, const connection_pool_timeouts& timeouts) {
    return bind_get_connection_timeout(source[io], timeouts.connect);
}
 
//[[DEPRECATED]] for backward compatibility only
template <typename ...Ts>
[[deprecated]] auto make_connector(connection_pool<Ts...>& source, io_context& io) {
    return source[io];
}
 
static_assert(ConnectionProvider<decltype(std::declval<connection_pool<connection_info<>>>()[std::declval<io_context&>()])>, "is not a ConnectionProvider");
 
template <typename T>
struct is_connection_pool : std::false_type {};
 
template <typename ...Args>
struct is_connection_pool<connection_pool<Args...>> : std::true_type {};
 
template <typename T>
constexpr auto ConnectionPool = is_connection_pool<std::decay_t<T>>::value;
 
template <typename ConnectionSource, typename ThreadSafety = decltype(thread_safe)>
auto make_connection_pool(ConnectionSource&& source, const connection_pool_config& config,
                          const ThreadSafety& thread_safety = ThreadSafety{}) {
    static_assert(ozo::ConnectionSource<ConnectionSource>, "source should model ConnectionSource concept");
    return connection_pool<std::decay_t<ConnectionSource>, std::decay_t<ThreadSafety>>{
        std::forward<ConnectionSource>(source), config, thread_safety};
}
 
} // namespace ozo
 
#include <ozo/impl/connection_pool.h>