operations.cpp

Go to the documentation of this file.

#include "operations.h"
 
// Simox
#include <SimoxUtility/json.h>
 
#include <RobotAPI/libraries/armem/core/wm/aron_conversions.h>
#include <RobotAPI/libraries/aron/converter/json/NLohmannJSONConverter.h>
 
#include "../operations.h"
 
namespace armarx::armem::server::ltm::mongodb::util
{
    namespace bsoncxxbuilder = bsoncxx::builder::stream;
    namespace bsoncxxdoc = bsoncxx::document;
 
    namespace
    {
        void
        mongodbInsertForeignKey(mongocxx::collection& coll, const std::string& key)
        {
            auto q = bsoncxxbuilder::document{} << std::string(constantes::FOREIGN_KEY) << key
                                                << bsoncxxbuilder::finalize;
            coll.insert_one(q.view());
        }
 
        bool
        mongodbContainsForeignKey(mongocxx::collection& coll, const std::string& key)
        {
            // check mongodb
            auto q = bsoncxxbuilder::document{} << std::string(constantes::FOREIGN_KEY) << key
                                                << bsoncxxbuilder::finalize;
            auto res = coll.find_one(q.view());
            return (bool)res;
        }
 
        bool
        mongodbContainsTimestamp(mongocxx::collection& coll, const long ts)
        {
            // check mongodb
            auto q = bsoncxxbuilder::document{} << std::string(constantes::TIMESTAMP) << ts
                                                << bsoncxxbuilder::finalize;
            auto res = coll.find_one(q.view());
            return (bool)res;
        }
    } // namespace
 
    void
    load(const mongocxx::database& db, armem::wm::Memory& m)
    {
        if (!db.has_collection(m.id().str()))
        {
            return;
        }
        mongocxx::collection coll = db.collection(m.id().str());
        mongocxx::cursor cursor = coll.find({});
        for (const auto& doc : cursor)
        {
            auto el = doc[constantes::FOREIGN_KEY];
            auto foreignKey = el.get_utf8().value;
 
            MemoryID i((std::string)foreignKey);
            if (i.memoryName != m.id().memoryName)
            {
                throw error::InvalidMemoryID(
                    i,
                    "A MemoryID in mongodb was invalid. Found the wrong memory name: " + i.str());
            }
 
            std::string k = i.coreSegmentName;
            if (m.hasCoreSegment(k))
            {
                throw error::ArMemError("Somehow the container already contains the key k = " + k +
                                        ". Do you have double entries in mongodb?");
            }
            else
            {
                auto& cSeg = m.addCoreSegment(k);
                load(db, cSeg);
            }
        }
    }
 
    void
    load(const mongocxx::database& db, armem::wm::CoreSegment& c)
    {
        if (!db.has_collection(c.id().str()))
        {
            return;
        }
        mongocxx::collection coll = db.collection(c.id().str());
        mongocxx::cursor cursor = coll.find({});
        for (const auto& doc : cursor)
        {
            auto el = doc[constantes::FOREIGN_KEY];
            auto foreignKey = el.get_utf8().value;
 
            MemoryID i((std::string)foreignKey);
            if (i.coreSegmentName != c.id().coreSegmentName || i.memoryName != c.id().memoryName)
            {
                throw error::InvalidMemoryID(
                    i,
                    "A MemoryID in mongodb was invalid. Found the wrong memory name: " + i.str());
            }
 
            std::string k = i.providerSegmentName;
            if (c.hasProviderSegment(k))
            {
                throw error::ArMemError("Somehow the container already contains the key k = " + k +
                                        ". Do you have double entries in mongodb?");
            }
            else
            {
                auto& pSeg = c.addProviderSegment(k);
                load(db, pSeg);
            }
        }
    }
 
    void
    load(const mongocxx::database& db, armem::wm::ProviderSegment& p)
    {
        if (!db.has_collection(p.id().str()))
        {
            return;
        }
        mongocxx::collection coll = db.collection(p.id().str());
        mongocxx::cursor cursor = coll.find({});
        for (const auto& doc : cursor)
        {
            auto el = doc[constantes::FOREIGN_KEY];
            auto foreignKey = el.get_utf8().value;
 
            MemoryID i((std::string)foreignKey);
            if (i.providerSegmentName != p.id().providerSegmentName ||
                i.coreSegmentName != p.id().coreSegmentName || i.memoryName != p.id().memoryName)
            {
                throw error::InvalidMemoryID(
                    i,
                    "A MemoryID in mongodb was invalid. Found the wrong memory name: " + i.str());
            }
 
            std::string k = i.entityName;
            if (p.hasEntity(k))
            {
                throw error::ArMemError("Somehow the container already contains the key k = " + k +
                                        ". Do you have double entries in mongodb?");
            }
            else
            {
                auto& eSeg = p.addEntity(k);
                load(db, eSeg);
            }
        }
    }
 
    void
    load(const mongocxx::database& db, armem::wm::Entity& e)
    {
        if (!db.has_collection(e.id().str()))
        {
            return;
        }
        mongocxx::collection coll = db.collection(e.id().str());
        mongocxx::cursor cursor = coll.find({});
        for (const auto& doc : cursor)
        {
            auto ts = armem::Time::microSeconds(doc[constantes::TIMESTAMP].get_int64().value);
            auto& newSnapshot = e.addSnapshot(ts);
 
            auto i = doc[constantes::INSTANCES];
            unsigned long length =
                std::distance(i.get_array().value.begin(), i.get_array().value.end());
 
            for (unsigned long i = 0; i < length; ++i)
            {
                // add an empty instance as reference
                newSnapshot.addInstance({});
            }
 
            // check to update lastSnapshot map TODO
            //checkUpdateLatestSnapshot(newSnapshot);
        }
    }
 
    void
    convert(const mongocxx::database& db, armem::wm::Memory& m)
    {
        m.forEachCoreSegment([&db](armem::wm::CoreSegment& e) { convert(db, e); });
    }
 
    void
    convert(const mongocxx::database& db, armem::wm::CoreSegment& c)
    {
        c.forEachProviderSegment([&db](armem::wm::ProviderSegment& e) { convert(db, e); });
    }
 
    void
    convert(const mongocxx::database& db, armem::wm::ProviderSegment& p)
    {
        p.forEachEntity([&db](armem::wm::Entity& e) { convert(db, e); });
    }
 
    void
    convert(const mongocxx::database& db, armem::wm::Entity& e)
    {
        e.forEachSnapshot(
            [&db](armem::wm::EntitySnapshot& e)
            {
                if (!ltm::util::entityHasData(e))
                {
                    // Get data from mongodb
                    auto eColl = db.collection(e.id().getEntityID().str());
                    auto q = bsoncxxbuilder::document{} << std::string(constantes::TIMESTAMP)
                                                        << e.id().timestamp.toMicroSeconds()
                                                        << bsoncxxbuilder::finalize;
                    auto res = eColl.find_one(q.view());
 
                    if (!res)
                    {
                        // break if the data could not be found in ltm storage
                        // perhaps its a not-set maybe type from the cache (not yet consollidated to ltm)
                        return false;
                    }
 
                    // convert full json of this entry
                    nlohmann::json json = nlohmann::json::parse(bsoncxx::to_json(*res));
                    nlohmann::json instances = json[constantes::INSTANCES];
 
                    if (instances.size() != e.size())
                    {
                        throw error::ArMemError(
                            "The size of the mongodb entity entry at id " +
                            e.id().getEntitySnapshotID().str() +
                            " has wrong size. Expected: " + std::to_string(e.size()) +
                            " but got: " + std::to_string(instances.size()));
                    }
 
                    for (unsigned int i = 0; i < e.size(); ++i)
                    {
                        auto& ins = e.getInstance(e.id().withInstanceIndex(i));
 
                        // get ionstance json
                        nlohmann::json doc = instances[i];
                        auto aron = aron::converter::AronNlohmannJSONConverter::
                            ConvertFromNlohmannJSONObject(doc);
 
                        // remove metadata
                        wm::EntityInstance tmp(e.id().withInstanceIndex(i));
                        from_aron(aron, tmp);
 
                        // set data
                        ins.data() = tmp.data();
                    }
                }
                return true;
            });
    }
 
    void
    store(const mongocxx::database& db, const armem::wm::Memory& m)
    {
        auto coll = db.collection(m.id().str());
        m.forEachCoreSegment(
            [&db, &coll](armem::wm::CoreSegment& e)
            {
                if (!mongodbContainsForeignKey(coll, e.id().str()))
                {
                    mongodbInsertForeignKey(coll, e.id().str());
                }
 
                store(db, e);
            });
    }
 
    void
    store(const mongocxx::database& db, const armem::wm::CoreSegment& c)
    {
        auto coll = db.collection(c.id().str());
        c.forEachProviderSegment(
            [&db, &coll](armem::wm::ProviderSegment& e)
            {
                if (!mongodbContainsForeignKey(coll, e.id().str()))
                {
                    mongodbInsertForeignKey(coll, e.id().str());
                }
 
                store(db, e);
            });
    }
 
    void
    store(const mongocxx::database& db, const armem::wm::ProviderSegment& p)
    {
        auto coll = db.collection(p.id().str());
        p.forEachEntity(
            [&db, &coll](armem::wm::Entity& e)
            {
                if (!mongodbContainsForeignKey(coll, e.id().str()))
                {
                    mongodbInsertForeignKey(coll, e.id().str());
                }
 
                store(db, e);
            });
    }
 
    void
    store(const mongocxx::database& db, const armem::wm::Entity& e)
    {
        auto coll = db.collection(e.id().str());
        e.forEachSnapshot(
            [&coll](armem::wm::EntitySnapshot& e)
            {
                if (!mongodbContainsTimestamp(coll, e.id().timestamp.toMilliSeconds()))
                {
                    // timestamp not found in mongodb ==> new entry
                    bsoncxxbuilder::document builder{};
                    auto in_array = builder << std::string(constantes::ID) << e.id().str()
                                            << std::string(constantes::TIMESTAMP)
                                            << e.id().timestamp.toMicroSeconds()
                                            << std::string(constantes::INSTANCES);
                    auto array_builder = bsoncxx::builder::basic::array{};
 
                    e.forEachInstance(
                        [&array_builder](const wm::EntityInstance& e)
                        {
                            auto aron = std::make_shared<aron::data::Dict>();
                            to_aron(aron, e);
                            nlohmann::json j =
                                aron::converter::AronNlohmannJSONConverter::ConvertToNlohmannJSON(
                                    aron);
 
                            auto doc_value = bsoncxx::from_json(j.dump(2));
                            array_builder.append(doc_value);
                        });
 
                    auto after_array = in_array << array_builder;
                    bsoncxx::document::value doc = after_array
                                                   << bsoncxx::builder::stream::finalize;
                    coll.insert_one(doc.view());
 
                    // check to update lastSnapshot map TODO
                    //checkUpdateLatestSnapshot(e);
                }
            });
    }
 
} // namespace armarx::armem::server::ltm::mongodb::util