DbClient.h

Go to the documentation of this file.

 
#pragma once
 
#include <drogon/exports.h>
#include <drogon/orm/Exception.h>
#include <drogon/orm/Field.h>
#include <drogon/orm/Result.h>
#include <drogon/orm/ResultIterator.h>
#include <drogon/orm/Row.h>
#include <drogon/orm/RowIterator.h>
#include <drogon/orm/SqlBinder.h>
#include <exception>
#include <functional>
#include <future>
#include <string>
#include <trantor/utils/Logger.h>
#include <trantor/utils/NonCopyable.h>
 
#ifdef __cpp_impl_coroutine
#include <drogon/utils/coroutine.h>
#endif
 
namespace drogon
{
namespace orm
{
using ResultCallback = std::function<void(const Result &)>;
using ExceptionCallback = std::function<void(const DrogonDbException &)>;
 
class Transaction;
class DbClient;

enum class TransactionType
{
    Deferred,  
    Immediate,  
    Exclusive,  
};
 
namespace internal
{
#ifdef __cpp_impl_coroutine
struct [[nodiscard]] SqlAwaiter : public CallbackAwaiter<Result>
{
    explicit SqlAwaiter(internal::SqlBinder &&binder)
        : binder_(std::move(binder))
    {
    }
 
    void await_suspend(std::coroutine_handle<> handle)
    {
        binder_ >> [handle, this](const drogon::orm::Result &result) {
            setValue(result);
            handle.resume();
        };
        binder_ >> [handle, this](const std::exception_ptr &e) {
            setException(e);
            handle.resume();
        };
        binder_.exec();
    }
 
  private:
    internal::SqlBinder binder_;
};
 
struct [[nodiscard]] TransactionAwaiter
    : public CallbackAwaiter<std::shared_ptr<Transaction> >
{
    explicit TransactionAwaiter(
        DbClient *client,
        TransactionType transType = TransactionType::Deferred)
        : client_(client), transType_(transType)
    {
    }
 
    void await_suspend(std::coroutine_handle<> handle);
 
  private:
    DbClient *client_;
    TransactionType transType_;
};
 
#endif
 
}  // namespace internal

class DROGON_EXPORT DbClient : public trantor::NonCopyable
{
  public:
    virtual ~DbClient();

    static std::shared_ptr<DbClient> newPgClient(const std::string &connInfo,
                                                 size_t connNum,
                                                 bool autoBatch = false);
    static std::shared_ptr<DbClient> newMysqlClient(const std::string &connInfo,
                                                    size_t connNum);
    static std::shared_ptr<DbClient> newSqlite3Client(
        const std::string &connInfo,
        size_t connNum);


    template <typename FUNCTION1, typename FUNCTION2, typename... Arguments>
    void execSqlAsync(const std::string &sql,
                      FUNCTION1 &&rCallback,
                      FUNCTION2 &&exceptCallback,
                      Arguments &&...args) noexcept
    {
        auto binder = *this << sql;
        (void)std::initializer_list<int>{
            (binder << std::forward<Arguments>(args), 0)...};
        binder >> std::forward<FUNCTION1>(rCallback);
        binder >> std::forward<FUNCTION2>(exceptCallback);
    }

    template <typename... Arguments>
    std::future<Result> execSqlAsyncFuture(const std::string &sql,
                                           Arguments &&...args) noexcept
    {
        auto binder = *this << sql;
        (void)std::initializer_list<int>{
            (binder << std::forward<Arguments>(args), 0)...};
        std::shared_ptr<std::promise<Result> > prom =
            std::make_shared<std::promise<Result> >();
        binder >> [prom](const Result &r) { prom->set_value(r); };
        binder >>
            [prom](const std::exception_ptr &e) { prom->set_exception(e); };
        binder.exec();
        return prom->get_future();
    }
 
    // Sync and blocking method
    template <typename... Arguments>
    Result execSqlSync(const std::string &sql,
                       Arguments &&...args) noexcept(false)
    {
        Result r(nullptr);
        {
            auto binder = *this << sql;
            (void)std::initializer_list<int>{
                (binder << std::forward<Arguments>(args), 0)...};
            // Use blocking mode
            binder << Mode::Blocking;
 
            binder >> [&r](const Result &result) { r = result; };
            binder.exec();  // exec may be throw exception;
        }
        return r;
    }
 
#ifdef __cpp_impl_coroutine
    template <typename... Arguments>
    internal::SqlAwaiter execSqlCoro(const std::string &sql,
                                     Arguments &&...args) noexcept
    {
        auto binder = *this << sql;
        (void)std::initializer_list<int>{
            (binder << std::forward<Arguments>(args), 0)...};
        return internal::SqlAwaiter(std::move(binder));
    }

    template <typename T>
    internal::SqlAwaiter execSqlCoro(const std::string &sql,
                                     const std::vector<T> &args) noexcept
    {
        auto binder = *this << sql;
        for (const auto &arg : args)
        {
            binder << arg;
        }
        return internal::SqlAwaiter(std::move(binder));
    }
#endif

    internal::SqlBinder operator<<(const std::string &sql);
    internal::SqlBinder operator<<(std::string &&sql);
 
    template <int N>
    internal::SqlBinder operator<<(const char (&sql)[N])
    {
        return internal::SqlBinder(sql, N - 1, *this, type_);
    }
 
    internal::SqlBinder operator<<(const std::string_view &sql)
    {
        return internal::SqlBinder(sql.data(), sql.length(), *this, type_);
    }


    virtual std::shared_ptr<Transaction> newTransaction(
        const std::function<void(bool)> &commitCallback =
            std::function<void(bool)>(),
        TransactionType transType =
            TransactionType::Deferred) noexcept(false) = 0;

    std::shared_ptr<Transaction> newTransaction(
        TransactionType transType) noexcept(false)
    {
        return newTransaction(std::function<void(bool)>(), transType);
    }


    virtual void newTransactionAsync(
        const std::function<void(const std::shared_ptr<Transaction> &)>
            &callback,
        TransactionType transType = TransactionType::Deferred) = 0;

    void newTransactionAsync(
        TransactionType transType,
        const std::function<void(const std::shared_ptr<Transaction> &)>
            &callback)
    {
        newTransactionAsync(callback, transType);
    }
 
#ifdef __cpp_impl_coroutine
    orm::internal::TransactionAwaiter newTransactionCoro(
        TransactionType transType = TransactionType::Deferred)
    {
        return orm::internal::TransactionAwaiter(this, transType);
    }
#endif

    virtual bool hasAvailableConnections() const noexcept = 0;
 
    ClientType type() const
    {
        return type_;
    }
 
    const std::string &connectionInfo() const
    {
        return connectionInfo_;
    }

    virtual void setTimeout(double timeout) = 0;
    virtual void closeAll() = 0;

    // virtual void enableAutoBatch() = 0;
 
  private:
    friend internal::SqlBinder;
    virtual void execSql(
        const char *sql,
        size_t sqlLength,
        size_t paraNum,
        std::vector<const char *> &&parameters,
        std::vector<int> &&length,
        std::vector<int> &&format,
        ResultCallback &&rcb,
        std::function<void(const std::exception_ptr &)> &&exceptCallback) = 0;
 
  protected:
    ClientType type_;
    std::string connectionInfo_;
};
 
using DbClientPtr = std::shared_ptr<DbClient>;
 
class Transaction : public DbClient
{
  public:
    virtual void rollback() = 0;
    // virtual void commit() = 0;
    virtual void setCommitCallback(
        const std::function<void(bool)> &commitCallback) = 0;
 
    void closeAll() override
    {
    }
};
 
#ifdef __cpp_impl_coroutine
inline void internal::TransactionAwaiter::await_suspend(
    std::coroutine_handle<> handle)
{
    assert(client_ != nullptr);
    client_->newTransactionAsync(
        [this, handle](const std::shared_ptr<Transaction> &transaction) {
            if (transaction == nullptr)
                setException(std::make_exception_ptr(TimeoutError(
                    "Timeout, no connection available for transaction")));
            else
                setValue(transaction);
            handle.resume();
        },
        transType_);
}
#endif
 
}  // namespace orm
}  // namespace drogon