operations.cpp

Go to the documentation of this file.

#include "operations.h"
 
// Simox
#include <SimoxUtility/json.h>
 
#include "../operations.h"
#include <RobotAPI/libraries/aron/converter/json/NLohmannJSONConverter.h>
#include <RobotAPI/libraries/armem/core/wm/aron_conversions.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;
        }
    }
 
    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