Skip to content
Merged
Show file tree
Hide file tree
Changes from 9 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions src/atomdb/AtomDB.h
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,38 @@ using namespace atoms;

namespace atomdb {

enum class AtomDBType { RedisMongoDB, MorkDB, InMemoryDB, RemoteAtomDB, AdapterDB };

class AtomDB : public HandleDecoder {
public:
AtomDB() = default;
virtual ~AtomDB() = default;

static AtomDBType string_to_type(const string& type) {
Comment thread
marcocapozzoli marked this conversation as resolved.
if (type == "redismongodb") return AtomDBType::RedisMongoDB;
if (type == "morkdb") return AtomDBType::MorkDB;
if (type == "inmemorydb") return AtomDBType::InMemoryDB;
if (type == "remotedb") return AtomDBType::RemoteAtomDB;
if (type == "adapterdb") return AtomDBType::AdapterDB;
RAISE_ERROR("Unsupported atomdb.type: " + type);
}

static string type_to_string(AtomDBType type) {
Comment thread
marcocapozzoli marked this conversation as resolved.
switch (type) {
case AtomDBType::RedisMongoDB:
return "redismongodb";
case AtomDBType::MorkDB:
return "morkdb";
case AtomDBType::InMemoryDB:
return "inmemorydb";
case AtomDBType::RemoteAtomDB:
return "remotedb";
case AtomDBType::AdapterDB:
return "adapterdb";
}
RAISE_ERROR("Unsupported AtomDBType");
}
Comment thread
marcocapozzoli marked this conversation as resolved.

virtual bool allow_nested_indexing() = 0;
virtual bool composite_type_enabled() const = 0;

Expand Down
116 changes: 116 additions & 0 deletions src/atomdb/AtomDBFactory.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
#include "AtomDBFactory.h"

#include "AdapterDB.h"
#include "InMemoryDB.h"
#include "MorkDB.h"
#include "RedisMongoDB.h"
#include "RemoteAtomDB.h"
#include "Utils.h"

using namespace atomdb;
using namespace commons;

// --------------------------------------------------------------------------------
// Public methods

shared_ptr<AtomDB> AtomDBFactory::create(const JsonConfig& config, const string& context) {
auto atomdb_type = config.at_path("type").get_or<string>("");

AtomDBType type = AtomDB::string_to_type(atomdb_type);

shared_ptr<AtomDB> atomdb;

if (type == AtomDBType::RedisMongoDB || type == AtomDBType::MorkDB ||
type == AtomDBType::InMemoryDB) {
atomdb = create_basic_atomdb(config, context);
} else if (type == AtomDBType::RemoteAtomDB || type == AtomDBType::AdapterDB) {
atomdb = create_composite_atomdb(config, context);
} else {
RAISE_ERROR("AtomDBFactory: unsupported AtomDB type: " + atomdb_type);
}

return wrap_if_protected(atomdb);
}

// --------------------------------------------------------------------------------
// Private methods

shared_ptr<AtomDB> AtomDBFactory::create_basic_atomdb(const JsonConfig& config, const string& context) {
auto atomdb_type = config.at_path("type").get_or<string>("");

AtomDBType type = AtomDB::string_to_type(atomdb_type);

shared_ptr<AtomDB> atomdb;

if (type == AtomDBType::RedisMongoDB) {
// make_shared cannot access RedisMongoDB's private ctor; friend can via new.
atomdb = shared_ptr<RedisMongoDB>(new RedisMongoDB(context, false, config));
Comment thread
andre-senna marked this conversation as resolved.
} else if (type == AtomDBType::MorkDB) {
atomdb = make_shared<MorkDB>(context, config);
} else if (type == AtomDBType::InMemoryDB) {
atomdb = make_shared<InMemoryDB>(context.empty() ? "inmemorydb_" : context);
} else {
RAISE_ERROR("AtomDBFactory: '" + atomdb_type + "' is not a basic AtomDB type");
}

return atomdb;
}

shared_ptr<AtomDB> AtomDBFactory::create_composite_atomdb(const JsonConfig& config,
const string& context) {
auto atomdb_type = config.at_path("type").get_or<string>("");

AtomDBType type = AtomDB::string_to_type(atomdb_type);

shared_ptr<AtomDB> atomdb;

if (type == AtomDBType::RemoteAtomDB) {
auto remote_peers_config = config.at_path("remote_peers").get_or<JsonConfig>(JsonConfig());

map<string, shared_ptr<RemoteAtomDBPeer>> remote_peers;

for (auto& entry : remote_peers_config) {
auto peer_config = JsonConfig(entry);
string uid = peer_config.at_path("uid").get_or<string>("");
if (uid.empty()) {
RAISE_ERROR("AtomDBFactory: remote peer is missing a non-empty uid");
}

string peer_context = peer_config.at_path("context").get_or<string>("");
if (peer_context.empty()) {
peer_context = "remotedb_" + uid;
}

shared_ptr<AtomDB> local_persistence = nullptr;
auto local_persistence_config =
peer_config.at_path("local_persistence").get_or<JsonConfig>(JsonConfig());
if (!local_persistence_config.empty()) {
string local_context =
local_persistence_config.at_path("context").get_or<string>(peer_context);
if (local_context.empty()) {
local_context = peer_context;
}
local_persistence = create_basic_atomdb(local_persistence_config, local_context);
}
remote_peers[uid] = make_shared<RemoteAtomDBPeer>(
create_basic_atomdb(peer_config, peer_context), local_persistence, uid);
}

return make_shared<RemoteAtomDB>(remote_peers);
}

if (type == AtomDBType::AdapterDB) {
// The backend AtomDB in AdapterDB could be RemoteAtomDB ?
auto atomdb_backend_config =
config.at_path("adapterdb.atomdb_backend").get_or<JsonConfig>(JsonConfig());
auto basic_atomdb = create_basic_atomdb(atomdb_backend_config, context);
return make_shared<AdapterDB>(config, basic_atomdb);
}

return atomdb;
Comment thread
marcocapozzoli marked this conversation as resolved.
}

shared_ptr<AtomDB> AtomDBFactory::wrap_if_protected(shared_ptr<AtomDB> atomdb) {
// AtomDBFactory::wrap_if_protected() is not implemented yet.
return atomdb;
}
48 changes: 48 additions & 0 deletions src/atomdb/AtomDBFactory.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
#pragma once

#include <memory>
#include <string>

#include "AtomDB.h"
#include "JsonConfig.h"

using namespace std;
using namespace commons;

namespace atomdb {

/**
* @brief Factory that builds AtomDB instances from a JsonConfig.
*
* This is the preferred way to obtain an AtomDB. Callers should not construct
* RedisMongoDB, MorkDB, InMemoryDB, RemoteAtomDB, or AdapterDB directly; instead
* pass a config whose "type" field selects the concrete implementation.
*
* Two kinds of AtomDB are supported:
* - Basic: RedisMongoDB, MorkDB, InMemoryDB — constructed from their own config.
* - Composite: RemoteAtomDB and AdapterDB — built by composing one or more basic
* AtomDBs (remote peers for RemoteAtomDB; a wrapped AtomDB for AdapterDB).
*
*/
class AtomDBFactory {
public:
/**
* @brief Creates a AtomDB and wraps it with ProtectedAtomDB when is applyable.
*/
static shared_ptr<AtomDB> create(const JsonConfig& config, const string& context = "");

private:
// Supported types: redismongodb, morkdb, inmemorydb.
static shared_ptr<AtomDB> create_basic_atomdb(const JsonConfig& config, const string& context = "");

// Supported types: remotedb, adapterdb.
static shared_ptr<AtomDB> create_composite_atomdb(const JsonConfig& config,
const string& context = "");

/**
* @brief Wraps an AtomDB with ProtectedAtomDB when protected and not already wrapped.
*/
static shared_ptr<AtomDB> wrap_if_protected(shared_ptr<AtomDB> atomdb);
};

} // namespace atomdb
24 changes: 3 additions & 21 deletions src/atomdb/AtomDBSingleton.cc
Original file line number Diff line number Diff line change
@@ -1,9 +1,6 @@
#include "AtomDBSingleton.h"

#include "AdapterDB.h"
#include "MorkDB.h"
#include "RedisMongoDB.h"
#include "RemoteAtomDB.h"
#include "AtomDBFactory.h"
#include "Utils.h"

using namespace atomdb;
Expand All @@ -19,24 +16,9 @@ void AtomDBSingleton::init(const JsonConfig& atomdb_config) {
if (AtomDBSingleton::initialized) {
RAISE_ERROR(
"AtomDBSingleton already initialized. AtomDBSingleton::init() should be called only once.");
} else {
auto atomdb_type = atomdb_config.at_path("type").get_or<string>("");
if (atomdb_type == "morkdb") {
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new MorkDB("", atomdb_config));
} else if (atomdb_type == "redismongodb") {
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new RedisMongoDB("", false, atomdb_config));
} else if (atomdb_type == "remotedb") {
auto remote_peers_config =
atomdb_config.at_path("remote_peers").get_or<JsonConfig>(JsonConfig());
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new RemoteAtomDB(remote_peers_config));
} else if (atomdb_type == "adapterdb") {
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new AdapterDB(atomdb_config));
} else {
RAISE_ERROR("Invalid AtomDB type: " + atomdb_type);
}

AtomDBSingleton::initialized = true;
}
AtomDBSingleton::atom_db = AtomDBFactory::create(atomdb_config);
AtomDBSingleton::initialized = true;
}

shared_ptr<AtomDB> AtomDBSingleton::get_instance() {
Expand Down
17 changes: 13 additions & 4 deletions src/atomdb/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,25 @@ cc_library(
deps = [
":atomdb",
":atomdb_api_types",
":atomdb_factory",
":atomdb_singleton",
":atomdbutils",
],
)

cc_library(
name = "atomdb_factory",
srcs = ["AtomDBFactory.cc"],
hdrs = ["AtomDBFactory.h"],
includes = ["."],
deps = [
":atomdb",
"//atomdb/adapterdb:adapterdb_lib",
"//atomdb/inmemorydb:inmemorydb_lib",
"//atomdb/morkdb:morkdb_lib",
"//atomdb/redis_mongodb:redis_mongodb_lib",
"//atomdb/remotedb:remotedb_lib",
"//commons:commons_lib",
],
)

Expand Down Expand Up @@ -56,11 +68,8 @@ cc_library(
hdrs = ["AtomDBSingleton.h"],
includes = ["."],
deps = [
":atomdb_factory",
"//atomdb:atomdb_api_types",
"//atomdb/adapterdb",
"//atomdb/morkdb",
"//atomdb/redis_mongodb",
"//atomdb/remotedb:remotedb_lib",
"//commons:commons_lib",
],
)
36 changes: 14 additions & 22 deletions src/atomdb/adapterdb/AdapterDB.cc
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,10 @@
#include "DatabaseOrchestrator.h"
#include "DedicatedThread.h"
#include "MongoInitializer.h"
#include "MorkDB.h"
#include "MorkMappingStrategy.h"
#include "PostgresMappingStrategy.h"
#include "PostgresWrapper.h"
#include "Processor.h"
#include "RedisMongoDB.h"
#include "RemoteAtomDB.h"
#include "Utils.h"
#include "expression_hasher.h"
#include "processor/ThreadPool.h"
Expand All @@ -33,11 +30,6 @@ string AdapterDB::MONGODB_ADAPTER_COLLECTION_NAME = "adapterdb";
// Construction / destruction
// ==============================

AdapterDB::AdapterDB(const JsonConfig& config) : config(config) {
this->atomdb_backend_setup();
this->initialize();
}

atomdb::AdapterDB::AdapterDB(const JsonConfig& config, std::shared_ptr<AtomDB> backend)
: config(config), atomdb_backend(backend) {
this->initialize(true);
Expand Down Expand Up @@ -310,20 +302,20 @@ void AdapterDB::persistence_setup() {
}
}

void AdapterDB::atomdb_backend_setup() {
auto atomdb_backend_config =
this->config.at_path("adapterdb.atomdb_backend").get_or<JsonConfig>(JsonConfig());
string atomdb_backend_type = atomdb_backend_config.at_path("type").get_or<string>("");
if (atomdb_backend_type == "morkdb") {
this->atomdb_backend = shared_ptr<AtomDB>(new MorkDB("", atomdb_backend_config));
} else if (atomdb_backend_type == "redismongodb") {
this->atomdb_backend = shared_ptr<AtomDB>(new RedisMongoDB("", false, atomdb_backend_config));
} else if (atomdb_backend_type == "remotedb") {
this->atomdb_backend = shared_ptr<AtomDB>(new RemoteAtomDB(atomdb_backend_config));
} else {
RAISE_ERROR("Invalid AtomDB type: " + atomdb_backend_type);
}
}
// void AdapterDB::atomdb_backend_setup() {
Comment thread
marcocapozzoli marked this conversation as resolved.
Outdated
// auto atomdb_backend_config =
// this->config.at_path("adapterdb.atomdb_backend").get_or<JsonConfig>(JsonConfig());
// string atomdb_backend_type = atomdb_backend_config.at_path("type").get_or<string>("");
// if (atomdb_backend_type == "morkdb") {
// this->atomdb_backend = shared_ptr<AtomDB>(new MorkDB("", atomdb_backend_config));
// } else if (atomdb_backend_type == "redismongodb") {
// this->atomdb_backend = shared_ptr<AtomDB>(new RedisMongoDB("", false, atomdb_backend_config));
// } else if (atomdb_backend_type == "remotedb") {
// this->atomdb_backend = shared_ptr<AtomDB>(new RemoteAtomDB(atomdb_backend_config));
// } else {
// RAISE_ERROR("Invalid AtomDB type: " + atomdb_backend_type);
// }
// }

bool AdapterDB::is_backend_ready() const { return this->backend_ready.load(); }

Expand Down
8 changes: 1 addition & 7 deletions src/atomdb/adapterdb/AdapterDB.h
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,7 @@ inline AdapterDbType parse_adapter_db_type(const string& value) {

class AdapterDB : public AtomDB {
public:
explicit AdapterDB(const JsonConfig& config);
AdapterDB(const JsonConfig& config, shared_ptr<AtomDB> backend); // for testing
AdapterDB(const JsonConfig& config, shared_ptr<AtomDB> backend);
~AdapterDB() override;

static string MONGODB_ADAPTER_COLLECTION_NAME;
Expand Down Expand Up @@ -128,11 +127,6 @@ class AdapterDB : public AtomDB {
*/
void persistence_setup();

/**
* @brief Initializes the AtomDB backend according to the configuration.
*/
void atomdb_backend_setup();

bool is_backend_ready() const;

void ensure_backend_ready() const;
Expand Down
3 changes: 0 additions & 3 deletions src/atomdb/adapterdb/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,6 @@ cc_library(
includes = ["."],
deps = [
"//atomdb",
"//atomdb/morkdb",
"//atomdb/redis_mongodb",
"//atomdb/remotedb:remotedb_lib",
"//commons:commons_lib",
"//commons/atoms:atoms_lib",
"//db_adapter:db_adapter_lib",
Expand Down
6 changes: 5 additions & 1 deletion src/atomdb/redis_mongodb/RedisMongoDB.h
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ enum MONGODB_FIELD { ID = 0, NAME, TARGETS, NAMED_TYPE, size };

class RedisMongoDB : public AtomDB {
public:
RedisMongoDB(const string& context, bool skip_redis, const JsonConfig& config);
~RedisMongoDB();

bool allow_nested_indexing() override;
Expand Down Expand Up @@ -146,6 +145,11 @@ class RedisMongoDB : public AtomDB {
map<string, vector<string>>& composite_type_entries_map);

private:
friend class AtomDBFactory;
friend class MorkDB;
Comment thread
marcocapozzoli marked this conversation as resolved.
Outdated

RedisMongoDB(const string& context, bool skip_redis, const JsonConfig& config);

string context;
bool skip_redis_;
bool composite_type_enabled_;
Expand Down
Loading
Loading