Cluster.hpp

#pragma once
 
#include "data/cassandra/impl/ManagedObject.hpp"
#include "util/Assert.hpp"
#include "util/log/Logger.hpp"
 
#include <cassandra.h>
 
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <optional>
#include <string>
#include <string_view>
#include <thread>
#include <variant>
 
namespace data::cassandra::impl {
 
enum class Provider { Cassandra, Keyspace };
 
inline Provider
providerFromString(std::string const& provider)
{
    ASSERT(
        provider == "cassandra" || provider == "aws_keyspace",
        "Provider type must be one of 'cassandra' or 'aws_keyspace'"
    );
    return provider == "cassandra" ? Provider::Cassandra : Provider::Keyspace;
}
 
// TODO: move Settings to public interface, not impl

struct Settings {
    static constexpr std::size_t kDEFAULT_CONNECTION_TIMEOUT = 10000;
    static constexpr uint32_t kDEFAULT_MAX_WRITE_REQUESTS_OUTSTANDING = 10'000;
    static constexpr uint32_t kDEFAULT_MAX_READ_REQUESTS_OUTSTANDING = 100'000;
    static constexpr std::size_t kDEFAULT_BATCH_SIZE = 20;
    static constexpr Provider kDEFAULT_PROVIDER = Provider::Cassandra;

    struct ContactPoints {
        std::string contactPoints = "127.0.0.1";  // defaults to localhost
        std::optional<uint16_t> port;
    };

    struct SecureConnectionBundle {
        std::string bundle;  // no meaningful default
    };

    bool enableLog = false;

    std::chrono::milliseconds connectionTimeout =
        std::chrono::milliseconds{kDEFAULT_CONNECTION_TIMEOUT};

    std::chrono::milliseconds requestTimeout = std::chrono::milliseconds{0};  // no timeout at all

    std::variant<ContactPoints, SecureConnectionBundle> connectionInfo = ContactPoints{};

    uint32_t threads = std::thread::hardware_concurrency();

    uint32_t maxWriteRequestsOutstanding = kDEFAULT_MAX_WRITE_REQUESTS_OUTSTANDING;

    uint32_t maxReadRequestsOutstanding = kDEFAULT_MAX_READ_REQUESTS_OUTSTANDING;

    uint32_t coreConnectionsPerHost = 3u;

    std::size_t writeBatchSize = kDEFAULT_BATCH_SIZE;

    Provider provider = kDEFAULT_PROVIDER;

    std::optional<uint32_t> queueSizeIO =
        std::nullopt;  // NOLINT(readability-redundant-member-init)

    std::optional<std::string> certificate =
        std::nullopt;  // NOLINT(readability-redundant-member-init)

    std::optional<std::string> username =
        std::nullopt;  // NOLINT(readability-redundant-member-init)

    std::optional<std::string> password =
        std::nullopt;  // NOLINT(readability-redundant-member-init)

    Settings
    withContactPoints(std::string_view contactPoints)
    {
        auto tmp = *this;
        tmp.connectionInfo =
            ContactPoints{.contactPoints = std::string{contactPoints}, .port = std::nullopt};
        return tmp;
    }

    static Settings
    defaultSettings()
    {
        return Settings();
    }
};
 
class Cluster : public ManagedObject<CassCluster> {
    util::Logger log_{"Backend"};
 
public:
    Cluster(Settings const& settings);
 
private:
    void
    setupConnection(Settings const& settings);
 
    void
    setupContactPoints(Settings::ContactPoints const& points);
 
    void
    setupSecureBundle(Settings::SecureConnectionBundle const& bundle);
 
    void
    setupCertificate(Settings const& settings);
 
    void
    setupCredentials(Settings const& settings);
};
 
}  // namespace data::cassandra::impl