PersistentEntitySegment.cpp

Go to the documentation of this file.

/*
* This file is part of ArmarX.
*
* ArmarX is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation.
*
* ArmarX is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* @package    MemoryX::Core
* @author     Alexey Kozlov ( kozlov at kit dot edu)
* @date       2013
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
 
#include "PersistentEntitySegment.h"
 
#include <iostream>
 
#include <Ice/ObjectAdapter.h>
#include <IceUtil/UUID.h>
 
#include <ArmarXCore/core/exceptions/local/ExpressionException.h>
#include <ArmarXCore/core/logging/Logging.h>
#include <ArmarXCore/interface/serialization/JSONSerialization.h>
#include <ArmarXCore/util/json/JSONObject.h>
 
#include "SegmentedMemory.h"
#include <MemoryX/core/MongoSerializer.h>
#include <MemoryX/core/entity/Entity.h>
#include <MemoryX/core/entity/EntityRef.h>
 
namespace memoryx
{
    PersistentEntitySegment::PersistentEntitySegment(CollectionInterfacePrx entityCollection,
                                                     Ice::CommunicatorPtr ic,
                                                     bool useMongoIds /* = true */) :
        useMongoIds(useMongoIds)
    {
        dbSerializer = new MongoSerializer(ic, useMongoIds);
        setSingleRWCollection(entityCollection);
    }
 
    PersistentEntitySegment::~PersistentEntitySegment()
    {
    }
 
    NameList
    PersistentEntitySegment::getReadCollectionsNS(const ::Ice::Current& c) const
    {
        memoryx::NameList result;
        auto lock = getReadLock(c);
 
        for (CollectionPrxList::const_iterator it = readCollections.begin();
             it != readCollections.end();
             ++it)
        {
            if (!*it)
            {
                continue;
            }
 
            result.push_back((*it)->getNS());
        }
 
        return result;
    }
 
    void
    PersistentEntitySegment::clearReadCollections(const ::Ice::Current& c)
    {
        ARMARX_INFO_S << "Clearing read collections";
        auto lock = getWriteLock(c);
        readCollections.clear();
    }
 
    void
    PersistentEntitySegment::addReadCollection(const CollectionInterfacePrx& coll,
                                               const ::Ice::Current& c)
    {
        if (coll)
        {
            ScopedSharedLockPtr lock(getReadLock(c));
 
            bool found = false;
            const std::string newName = coll->getNS();
 
            for (CollectionPrxList::iterator it = readCollections.begin();
                 it != readCollections.end();
                 it++)
            {
                ARMARX_CHECK_EXPRESSION((*it));
 
                if (newName == (*it)->getNS())
                {
                    found = true;
                    break;
                }
            }
 
            if (!found)
            {
                readCollections.push_back(coll);
                ARMARX_INFO_S << coll->getNS() << " collection added to list";
            }
            else
            {
                ARMARX_INFO_S << coll->getNS() << " collection already in list";
            }
        }
    }
 
    std::string
    PersistentEntitySegment::getWriteCollectionNS(const ::Ice::Current& c) const
    {
        return writeCollection->getNS();
    }
 
    void
    PersistentEntitySegment::setWriteCollection(const CollectionInterfacePrx& coll,
                                                const ::Ice::Current& c)
    {
        if (coll)
        {
            writeCollection = coll;
        }
    }
 
    void
    PersistentEntitySegment::setSingleRWCollection(const CollectionInterfacePrx& coll,
                                                   const ::Ice::Current& c)
    {
        setWriteCollection(coll);
        clearReadCollections();
        addReadCollection(coll);
    }
 
    std::string
    PersistentEntitySegment::getObjectTypeId(const ::Ice::Current& c) const
    {
        return objectTypeId;
    }
 
    std::string
    PersistentEntitySegment::addEntity(const EntityBasePtr& entity, const ::Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
        return addEntityThreadUnsafe(entity);
    }
 
    std::string
    PersistentEntitySegment::addEntityThreadUnsafe(const EntityBasePtr& entity)
    {
        DBStorableData dbEntity;
        {
            std::unique_lock lock(dbSerializerMutex);
            dbEntity = dbSerializer->serializeIceObject(entity);
        }
        return writeCollection->insert(dbEntity);
    }
 
    std::string
    PersistentEntitySegment::upsertEntity(const std::string& entityId,
                                          const EntityBasePtr& entity,
                                          const Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
        return upsertEntityThreadUnsafe(entityId, entity);
    }
 
    std::string
    PersistentEntitySegment::upsertEntityThreadUnsafe(const std::string& entityId,
                                                      const EntityBasePtr& entity)
    {
        if (hasEntityByIdThreadUnsafe(entityId))
        {
            updateEntityThreadUnsafe(entityId, entity);
            return entityId;
        }
        else
        {
            return addEntityThreadUnsafe(entity);
        }
    }
 
    EntityIdList
    PersistentEntitySegment::upsertEntityList(const EntityBaseList& entityList,
                                              const Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
        return upsertEntityListThreadUnsafe(entityList);
    }
 
    EntityIdList
    PersistentEntitySegment::upsertEntityListThreadUnsafe(const EntityBaseList& entityList)
    {
        EntityIdList entityIds(entityList.size(), "");
 
        for (size_t i = 0; i < entityList.size(); i++)
        {
            try
            {
                entityIds[i] = upsertEntityThreadUnsafe(entityList[i]->getId(), entityList[i]);
            }
            catch (const memoryx::InvalidEntityException& e)
            {
                // don't handle this exception but return an empty string as ID for this specific entity
            }
        }
 
        return entityIds;
    }
 
    std::string
    PersistentEntitySegment::upsertEntityByName(const std::string& entityName,
                                                const EntityBasePtr& entity,
                                                const Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
        return upsertEntityByNameThreadUnsafe(entityName, entity);
    }
 
    std::string
    PersistentEntitySegment::upsertEntityByNameThreadUnsafe(const std::string& entityName,
                                                            const EntityBasePtr& entity)
    {
        auto oldEntity = getEntityByNameThreadUnsafe(entityName);
        if (oldEntity)
        {
            updateEntityThreadUnsafe(oldEntity->getId(), entity);
            return oldEntity->getId();
        }
        else
        {
            return addEntityThreadUnsafe(entity);
        }
    }
 
    EntityIdList
    PersistentEntitySegment::addEntityList(const EntityBaseList& entityList, const Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
        return addEntityListThreadUnsafe(entityList);
    }
 
    EntityIdList
    PersistentEntitySegment::addEntityListThreadUnsafe(const EntityBaseList& entityList)
    {
        DBStorableDataList dbEntities;
        dbEntities.reserve(entityList.size());
 
        for (size_t i = 0; i < entityList.size(); i++)
        {
            std::unique_lock lock(dbSerializerMutex);
 
            dbEntities.push_back(dbSerializer->serializeIceObject(entityList[i]));
        }
 
        return writeCollection->insertList(dbEntities);
    }
 
    void
    PersistentEntitySegment::updateEntity(const ::std::string& entityId,
                                          const EntityBasePtr& update,
                                          const ::Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
        updateEntityThreadUnsafe(entityId, update);
    }
 
    void
    PersistentEntitySegment::updateEntityThreadUnsafe(const std::string& entityId,
                                                      const EntityBasePtr& update)
    {
        update->setId(entityId);
        DBStorableData dbEntity;
        {
            std::unique_lock lock(dbSerializerMutex);
            dbEntity = dbSerializer->serializeIceObject(update);
        }
 
        writeCollection->update(dbEntity);
    }
 
    void
    PersistentEntitySegment::removeEntity(const ::std::string& entityId, const ::Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
        removeEntityThreadUnsafe(entityId);
    }
 
    void
    PersistentEntitySegment::removeEntityThreadUnsafe(const std::string& entityId)
    {
        if (useMongoIds)
        {
            writeCollection->removeByMongoId(entityId);
        }
        else
        {
            writeCollection->removeByFieldValue("_id", entityId);
        }
    }
 
    void
    PersistentEntitySegment::removeAllEntities(const Ice::Current& c)
    {
        ScopedUniqueLockPtr lock(getWriteLock(c));
 
        auto ids = getAllEntityIdsThreadUnsafe();
 
        for (auto& id : ids)
        {
            removeEntityThreadUnsafe(id);
        }
    }
 
    bool
    PersistentEntitySegment::hasEntityById(const std::string& entityId, const Ice::Current& c) const
    {
        ScopedSharedLockPtr lock(getReadLock(c));
        return hasEntityByIdThreadUnsafe(entityId);
    }
 
    bool
    PersistentEntitySegment::hasEntityByIdThreadUnsafe(const std::string& entityId) const
    {
        if (entityId.empty() || entityId.size() != 24)
        {
            return false;
        }
        try
        {
            for (CollectionPrxList::const_iterator it = readCollections.begin();
                 it != readCollections.end();
                 ++it)
            {
                if (!*it)
                {
                    continue;
                }
 
                const DBStorableData dbEntity = useMongoIds
                                                    ? (*it)->findByMongoId(entityId)
                                                    : (*it)->findOneByFieldValue("_id", entityId);
 
                if (!dbEntity.JSON.empty())
                {
                    return true;
                }
            }
        }
        catch (...)
        {
        }
 
        return false;
    }
 
    bool
    PersistentEntitySegment::hasEntityByName(const std::string& entityName,
                                             const Ice::Current& c) const
    {
        ScopedSharedLockPtr lock(getReadLock(c));
        return hasEntityByNameThreadUnsafe(entityName);
    }
 
    bool
    PersistentEntitySegment::hasEntityByNameThreadUnsafe(const std::string& entityName) const
    {
        for (CollectionPrxList::const_iterator it = readCollections.begin();
             it != readCollections.end();
             ++it)
        {
            CollectionInterfacePrx coll = *it;
 
            if (!coll)
            {
                continue;
            }
 
            const DBStorableData dbEntity = coll->findOneByFieldValue("name", entityName);
 
            if (!dbEntity.JSON.empty())
            {
                return true;
            }
        }
 
        return false;
    }
 
    EntityBasePtr
    PersistentEntitySegment::getEntityById(const ::std::string& entityId,
                                           const ::Ice::Current& c) const
    {
        ScopedSharedLockPtr lock(getReadLock(c));
 
        return getEntityByIdThreadUnsafe(entityId);
    }
 
    EntityBasePtr
    PersistentEntitySegment::getEntityByIdThreadUnsafe(const std::string& entityId) const
    {
        if (entityId.empty())
        {
            ARMARX_ERROR_S << "Entity id must not be empty!";
            return EntityBasePtr();
        }
 
 
        for (CollectionPrxList::const_iterator it = readCollections.begin();
             it != readCollections.end();
             ++it)
        {
            if (!*it)
            {
                continue;
            }
 
            const DBStorableData dbEntity = useMongoIds
                                                ? (*it)->findByMongoId(entityId)
                                                : (*it)->findOneByFieldValue("_id", entityId);
 
            if (!dbEntity.JSON.empty())
            {
                return deserializeEntity(dbEntity);
            }
        }
 
        return EntityBasePtr();
    }
 
    EntityBasePtr
    PersistentEntitySegment::getEntityByName(const ::std::string& name,
                                             const ::Ice::Current& c) const
    {
        ScopedSharedLockPtr lock(getReadLock(c));
        return getEntityByNameThreadUnsafe(name);
    }
 
    EntityBasePtr
    PersistentEntitySegment::getEntityByNameThreadUnsafe(const std::string& name) const
    {
        for (CollectionPrxList::const_iterator it = readCollections.begin();
             it != readCollections.end();
             ++it)
        {
            if (!*it)
            {
                continue;
            }
 
            const DBStorableData dbEntity = (*it)->findOneByFieldValue("name", name);
 
            if (!dbEntity.JSON.empty())
            {
                return deserializeEntity(dbEntity);
            }
        }
 
        return EntityBasePtr();
    }
 
    EntityBaseList
    PersistentEntitySegment::getEntitiesByAttrValue(const ::std::string& attrName,
                                                    const ::std::string& attrValue,
                                                    const ::Ice::Current& c) const
    {
        EntityBaseList result;
        const std::string fieldName = "attrs." + attrName + ".value.value";
        ScopedSharedLockPtr lock(getReadLock(c));
 
        for (CollectionPrxList::const_iterator itColl = readCollections.begin();
             itColl != readCollections.end();
             ++itColl)
        {
            if (!*itColl)
            {
                continue;
            }
 
            const DBStorableDataList dbEntities = (*itColl)->findByFieldValue(fieldName, attrValue);
 
            for (DBStorableDataList::const_iterator itEntity = dbEntities.begin();
                 itEntity != dbEntities.end();
                 ++itEntity)
            {
                EntityBasePtr entity = deserializeEntity(*itEntity);
 
                if (entity)
                {
                    result.push_back(entity);
                }
            }
        }
 
        return result;
    }
 
    EntityBaseList
    PersistentEntitySegment::getEntitiesByAttrValueList(const ::std::string& attrName,
                                                        const NameList& attrValueList,
                                                        const ::Ice::Current& c) const
    {
        EntityBaseList result;
        const std::string fieldName = "attrs." + attrName + ".value.value";
        ScopedSharedLockPtr lock(getReadLock(c));
 
        for (CollectionPrxList::const_iterator itColl = readCollections.begin();
             itColl != readCollections.end();
             ++itColl)
        {
            if (!*itColl)
            {
                continue;
            }
 
            const DBStorableDataList dbEntities =
                (*itColl)->findByFieldValueList(fieldName, attrValueList);
 
            for (DBStorableDataList::const_iterator itEntity = dbEntities.begin();
                 itEntity != dbEntities.end();
                 ++itEntity)
            {
                EntityBasePtr entity = deserializeEntity(*itEntity);
 
                if (entity)
                {
                    result.push_back(entity);
                }
            }
        }
 
        return result;
    }
 
    Ice::Int
    PersistentEntitySegment::size(const ::Ice::Current& c) const
    {
        Ice::Int result = 0;
        ScopedSharedLockPtr lock(getReadLock(c));
 
        for (CollectionPrxList::const_iterator it = readCollections.begin();
             it != readCollections.end();
             ++it)
        {
            if (!*it)
            {
                continue;
            }
 
            result += (*it)->count();
        }
 
        return result;
    }
 
    void
    PersistentEntitySegment::print(const ::Ice::Current& c) const
    {
        //TODO implement
    }
 
    void
    PersistentEntitySegment::clear(const ::Ice::Current& c)
    {
        writeCollection->clear();
    }
 
    Ice::Identity
    PersistentEntitySegment::getIceId(const Ice::Current&) const
    {
        Ice::Identity id;
        id.name = (parentMemory ? parentMemory->getMemoryName() : "") + "_" + segmentName;
        return id;
    }
 
    std::string
    PersistentEntitySegment::getSegmentName(const Ice::Current&) const
    {
        return segmentName;
    }
 
    EntityIdList
    PersistentEntitySegment::getAllEntityIds(const ::Ice::Current& c) const
    {
        ScopedSharedLockPtr lock(getReadLock(c));
        return getAllEntityIdsThreadUnsafe();
    }
 
    EntityIdList
    PersistentEntitySegment::getAllEntityIdsThreadUnsafe() const
    {
        EntityIdList result;
 
        for (CollectionPrxList::const_iterator it = readCollections.begin();
             it != readCollections.end();
             ++it)
        {
            if (!*it)
            {
                continue;
            }
 
            const EntityIdList collIds = (*it)->findAllIds();
            result.insert(result.end(), collIds.begin(), collIds.end());
        }
        return result;
    }
 
    EntityBaseList
    PersistentEntitySegment::getAllEntities(const ::Ice::Current& c) const
    {
        EntityBaseList result;
        ScopedSharedLockPtr lock(getReadLock(c));
 
        for (const auto& col : readCollections)
        {
            if (!col)
            {
                continue;
            }
 
            for (const auto& dbEntity : col->findAll())
            {
                auto entity = deserializeEntity(dbEntity);
 
                if (entity)
                {
                    result.push_back(std::move(entity));
                }
            }
        }
 
        return result;
    }
 
    IdEntityMap
    PersistentEntitySegment::getIdEntityMap(const ::Ice::Current& c) const
    {
        IdEntityMap result;
 
        for (auto&& entity : getAllEntities())
        {
            result.insert({entity->getId(), std::move(entity)});
        }
 
        return result;
    }
 
    EntityBasePtr
    PersistentEntitySegment::deserializeEntity(const DBStorableData& dbEntity) const
    {
        std::unique_lock lock(dbSerializerMutex);
        EntityBasePtr entity;
 
        try
        {
            entity = EntityBasePtr::dynamicCast(dbSerializer->deserializeIceObject(dbEntity));
        }
        catch (const armarx::JSONException& ex)
        {
            entity = EntityBasePtr();
        }
 
        return entity;
    }
 
    EntityRefBasePtr
    memoryx::PersistentEntitySegment::getEntityRefById(const std::string& id,
                                                       const Ice::Current& c) const
    {
        auto entity = getEntityById(id);
 
        if (!entity)
        {
            return NULL;
        }
 
        auto proxy =
            EntityMemorySegmentInterfacePrx::uncheckedCast(c.adapter->createProxy(getIceId()));
        SegmentedMemoryPtr mem = SegmentedMemoryPtr::dynamicCast(parentMemory);
        std::string memName = mem->getMemoryName();
        auto memoryProxy =
            MemoryInterfacePrx::checkedCast(c.adapter->getCommunicator()->stringToProxy(memName));
        EntityRefPtr ref = new EntityRef(entity, segmentName, memName, memoryProxy, proxy);
        return ref;
    }
 
    EntityRefBasePtr
    memoryx::PersistentEntitySegment::getEntityRefByName(const std::string& name,
                                                         const Ice::Current& c) const
    {
        auto entity = getEntityByName(name);
 
        if (!entity)
        {
            return NULL;
        }
 
        auto proxy =
            EntityMemorySegmentInterfacePrx::uncheckedCast(c.adapter->createProxy(getIceId()));
        SegmentedMemoryPtr mem = SegmentedMemoryPtr::dynamicCast(parentMemory);
        std::string memName = mem->getMemoryName();
        auto memoryProxy =
            MemoryInterfacePrx::checkedCast(c.adapter->getCommunicator()->stringToProxy(memName));
        EntityRefPtr ref = new EntityRef(entity, segmentName, memName, memoryProxy, proxy);
        return ref;
    }
 
    void
    PersistentEntitySegment::setEntityAttribute(const std::string& entityId,
                                                const EntityAttributeBasePtr& attribute,
                                                const Ice::Current& c)
    {
        EntityBasePtr entity = getEntityById(entityId);
        if (!entity)
        {
            throw EntityNotFoundException();
        }
        entity->putAttribute(attribute);
        DBStorableData dbEntity;
        {
            std::unique_lock lock(dbSerializerMutex);
            dbEntity = dbSerializer->serializeIceObject(entity);
        }
        ScopedUniqueLockPtr lock(getWriteLock(c));
 
        writeCollection->update(dbEntity);
    }
 
    void
    PersistentEntitySegment::setEntityAttributes(const std::string& entityId,
                                                 const EntityAttributeList& attributeMap,
                                                 const Ice::Current& c)
    {
        EntityBasePtr entity = getEntityById(entityId);
        if (!entity)
        {
            throw EntityNotFoundException();
        }
        for (auto attribute : attributeMap)
        {
            entity->putAttribute(attribute);
        }
        {
            DBStorableData dbEntity;
            {
                std::unique_lock lock(dbSerializerMutex);
                dbEntity = dbSerializer->serializeIceObject(entity);
            }
            ScopedUniqueLockPtr lock(getWriteLock(c));
            writeCollection->update(dbEntity);
        }
    }
 
    void
    memoryx::PersistentEntitySegment::setParentMemory(const MemoryInterfacePtr& memory,
                                                      const Ice::Current&)
    {
        this->parentMemory = memory;
    }
 
    EntityRefList
    PersistentEntitySegment::findRefsByQuery(const std::string& query, const Ice::Current& c)
    {
        EntityRefList result;
        ScopedSharedLockPtr lock(getReadLock(c));
 
        for (CollectionPrxList::const_iterator it = readCollections.begin();
             it != readCollections.end();
             ++it)
        {
            if (!*it)
            {
                continue;
            }
 
            CollectionInterfacePrx prx = *it;
            const DBStorableDataList data = prx->findByQuery(query);
 
            for (const DBStorableData& e : data)
            {
                EntityBasePtr entity = deserializeEntity(e);
 
                if (entity)
                {
                    auto proxy = EntityMemorySegmentInterfacePrx::uncheckedCast(
                        c.adapter->createProxy(getIceId()));
                    SegmentedMemoryPtr mem = SegmentedMemoryPtr::dynamicCast(parentMemory);
                    std::string memName = mem->getMemoryName();
                    auto memoryProxy = MemoryInterfacePrx::checkedCast(
                        c.adapter->getCommunicator()->stringToProxy(memName));
                    EntityRefPtr ref =
                        new EntityRef(entity, segmentName, memName, memoryProxy, proxy);
                    result.push_back(ref);
                }
            }
        }
 
        return result;
    }
 
    void
    PersistentEntitySegment::setSegmentName(const std::string& segmentName, const Ice::Current&)
    {
        this->segmentName = segmentName;
    }
} // namespace memoryx