CommonStorage.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::CommonStorage
* @author     Alexey Kozlov ( kozlov at kit dot edu)
* @date       2012
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
 
#include "CommonStorage.h"
 
#include <filesystem>
#include <fstream>
#include <memory>
 
#include <Ice/ObjectAdapter.h>
#include <IceUtil/UUID.h>
 
#include <ArmarXCore/core/ArmarXObjectScheduler.h>
#include <ArmarXCore/interface/core/Log.h>
 
#include "Collection.h"
#include "Database.h"
#include "GridFileWrapper.h"
#include <mongo/client/dbclient.h>
#include <mongo/client/dbclientinterface.h>
 
namespace memoryx
{
    const char MONGO_ID_FIELD[] = "_id";
 
    std::string
    CommonStorage::getDefaultName() const
    {
        return "CommonStorage";
    }
 
    void
    CommonStorage::onInitComponent()
    {
        // http://api.mongodb.com/cxx/current/namespacemongo_1_1client.html#aad4213a224b92333d2395839e4c81498
        // initialize() MUST be called EXACTLY once after entering 'main' and before using the driver.
        // Do not call initialize() before entering 'main' (i.e. from a static initializer), as it
        // relies on all static initialization having been completed.
        mongo::client::initialize();
 
        accessGridFSFilesMutex.reset(new std::mutex());
        if (getProperty<std::string>("MongoHost").isSet())
        {
            hostAndPort = getProperty<std::string>("MongoHost").getValue();
        }
        else if (!getIceProperties()->getProperty("ArmarX.MongoHost").empty())
        {
            hostAndPort = getIceProperties()->getProperty("ArmarX.MongoHost");
        }
        else if (getenv("MONGODB_HOST"))
        {
            hostAndPort = getenv("MONGODB_HOST");
        }
        else
        {
            hostAndPort = getProperty<std::string>("MongoHost").getValue();
        }
 
        if (hostAndPort.find(':') == std::string::npos)
        {
            if (!getIceProperties()->getProperty("ArmarX.MongoPort").empty())
            {
                hostAndPort += ":" + getIceProperties()->getProperty("ArmarX.MongoPort");
            }
            else if (getenv("MONGODB_PORT"))
            {
                hostAndPort += ":" + std::string(getenv("MONGODB_PORT"));
            }
        }
 
        ARMARX_IMPORTANT << "Using MongoDB on " << hostAndPort;
        //        dbName = getProperty<std::string>("MongoDBName").getValue();
 
        useAuth = getProperty<bool>("MongoAuth").getValue();
        userName = getProperty<std::string>("MongoUser").getValue();
 
        connectionCheckerTask =
            new armarx::PeriodicTask<CommonStorage>(this, &CommonStorage::checkConnection, 1000);
    }
 
    void
    CommonStorage::onConnectComponent()
    {
        ARMARX_INFO << "Starting MemoryX::CommonStorage";
 
        while (!connect() && !this->getObjectScheduler()->isTerminationRequested())
        {
            sleep(5);
        }
 
        pwdDigest =
            createPasswordDigest(userName, getProperty<std::string>("MongoPassword").getValue());
        connectionCheckerTask->start();
        ARMARX_INFO << "MemoryX::CommonStorage started";
    }
 
    void
    CommonStorage::onExitComponent()
    {
        connectionCheckerTask->stop();
        {
            std::unique_lock l(openedDatabasesMutex);
            openedDatabases.clear();
        }
        {
            std::unique_lock l(openedCollectionsMutex);
            openedCollections.clear();
        }
        {
            std::unique_lock l(openedFilesMutex);
            openedFiles.clear();
        }
        {
            std::unique_lock l(openedGridFSMutex);
            openedGridFS.clear();
        }
 
        mongo::client::shutdown();
    }
 
    armarx::PropertyDefinitionsPtr
    CommonStorage::createPropertyDefinitions()
    {
        return armarx::PropertyDefinitionsPtr(
            new CommonStoragePropertyDefinitions(getConfigIdentifier()));
    }
 
    bool
    CommonStorage::connect()
    {
        try
        {
            openedGridFS.clear();
            conn.reset(new mongo::DBClientConnection);
            conn->connect(hostAndPort);
        }
        catch (const mongo::DBException& e)
        {
            ARMARX_ERROR << "Can't connect to MongoDB: " << e.what();
            return false;
        }
        catch (const std::exception& e)
        {
            ARMARX_ERROR << "Can't connect to MongoDB: " << e.what();
            return false;
        }
 
        ARMARX_INFO << "Connected to Mongo: host = " << hostAndPort;
 
        return true;
    }
 
    void
    CommonStorage::checkConnection()
    {
        try
        {
            mongo::DBClientConnection conn;
            conn.connect(hostAndPort);
            conn.getDatabaseNames();
        }
        catch (...)
        {
            ARMARX_ERROR << "Connection to MongoDB lost";
            this->getObjectScheduler()->disconnected(true);
        }
    }
 
    CommonStorage::ConnectionWrapper
    CommonStorage::getConnection()
    {
        std::shared_ptr<mongo::DBClientConnection> conn;
        {
            std::lock_guard<std::mutex> guard{serverSettingsMutex};
            if (pool.empty())
            {
                while (true)
                {
                    try
                    {
                        conn.reset(new mongo::DBClientConnection());
                        conn->connect(hostAndPort);
                    }
                    catch (std::exception& e)
                    {
                        ARMARX_ERROR << "Can't connect to MongoDB: " << e.what();
                        continue;
                    }
                    break;
                }
            }
            else
            {
                conn = std::move(pool.front());
                pool.pop_front();
            }
        }
        return {*this, std::move(conn)};
    }
 
    std::string
    CommonStorage::createPasswordDigest(const std::string& username, const std::string& password)
    {
        // No real connection is required here.
        // In later versions of the API this is a static function
        std::string digest = getConnection().conn().createPasswordDigest(username, password);
        return digest;
    }
 
    std::string
    CommonStorage::extractDBNameFromNS(const std::string& ns)
    {
        const size_t found = ns.find_first_of('.');
        return (found != std::string::npos) ? ns.substr(0, found) : "";
    }
 
    bool
    CommonStorage::forceAuthenticate(const std::string& dbName,
                                     const std::string& userName,
                                     const std::string& password)
    {
        std::string errmsg;
        bool result = false;
        result = getConnection().conn().auth(dbName, userName, pwdDigest, errmsg, false);
 
        if (result)
        {
            authDBs.insert(dbName);
        }
 
        return result;
    }
 
    bool
    CommonStorage::authenticateNS(const std::string& ns)
    {
        const std::string dbName = extractDBNameFromNS(ns);
 
        // check that database part is present in ns string
        if (dbName.empty())
        {
            throw DBNotSpecifiedException("Database name not specified for collection: " + ns +
                                          ". Please use <dbName>.<collectionName> format!");
        }
 
        return authenticateDB(dbName);
    }
 
    bool
    CommonStorage::authenticateDB(const std::string& dbName)
    {
        if (!useAuth)
        {
            return true;
        }
 
        ARMARX_INFO << "Try to Auth: db = " << dbName << ", user = " << userName
                    << ", pwd = " << pwdDigest << std::endl;
 
        if (authDBs.count(dbName))
        {
            return true;
        }
        else
        {
            return forceAuthenticate(dbName, userName, pwdDigest);
        }
    }
 
    bool
    CommonStorage::authDB(const std::string& dbName,
                          const std::string& userName,
                          const std::string& password,
                          const Ice::Current&)
    {
        pwdDigest = createPasswordDigest(userName, password);
        return forceAuthenticate(dbName, userName, this->pwdDigest);
    }
 
    std::string
    CommonStorage::getMongoHostAndPort(const ::Ice::Current& c)
    {
        return hostAndPort;
    }
 
    NameList
    CommonStorage::getDBNames(const ::Ice::Current&)
    {
        authenticateDB("admin");
 
        {
            std::list<std::string> result = getConnection().conn().getDatabaseNames();
            return NameList(result.begin(), result.end());
        }
    }
 
    NameList
    CommonStorage::getCollectionNames(const ::std::string& dbName, const ::Ice::Current&)
    {
        std::list<std::string> result = getConnection().conn().getCollectionNames(dbName);
        return NameList(result.begin(), result.end());
    }
 
    bool
    CommonStorage::isConnected(const ::Ice::Current& c)
    {
        return true;
    }
 
    bool
    CommonStorage::reconnect(const ::std::string& hostAndPort,
                             const ::std::string& userName,
                             const ::std::string& password,
                             const ::Ice::Current&)
    {
        this->hostAndPort = hostAndPort;
        this->userName = userName;
        this->pwdDigest = createPasswordDigest(userName, password);
        this->useAuth = !userName.empty();
        return connect();
    }
 
    DatabaseInterfacePrx
    CommonStorage::requestDatabase(const ::std::string& dbName, const ::Ice::Current& c)
    {
        // authenticate if needed
        if (authenticateDB(dbName))
        {
            DatabasePtr db = new Database(this, dbName);
            Ice::Identity dbId = db->getIceId();
            {
                std::unique_lock l(openedDatabasesMutex);
                openedDatabases[dbId] = db;
            }
            Ice::ObjectPrx node = c.adapter->add(db, dbId);
            return DatabaseInterfacePrx::uncheckedCast(node);
        }
        else
        {
            throw MongoAuthenticationException("Mongo authentication failed (user = " + userName +
                                               ", database = " + dbName + ")");
        }
    }
 
    void
    CommonStorage::releaseDatabase(const DatabaseInterfacePrx& db, const ::Ice::Current&)
    {
        std::unique_lock l(openedDatabasesMutex);
        openedDatabases.erase(db->ice_getIdentity());
    }
 
    CollectionInterfacePrx
    CommonStorage::requestCollection(const std::string& collectionNS, const ::Ice::Current& c)
    {
        // authenticate if needed
        if (authenticateNS(collectionNS))
        {
            CollectionPtr coll = new Collection(this, collectionNS);
            Ice::Identity collId = coll->getIceId();
            {
                std::unique_lock l(openedCollectionsMutex);
                openedCollections[collId] = coll;
            }
            Ice::ObjectPrx node = c.adapter->add(coll, collId);
            return CollectionInterfacePrx::uncheckedCast(node);
        }
        else
        {
            throw MongoAuthenticationException("Mongo authentication failed (user = " + userName +
                                               ", collection = " + collectionNS + ")");
        }
    }
 
    void
    CommonStorage::releaseCollection(const CollectionInterfacePrx& coll, const ::Ice::Current& c)
    {
        std::unique_lock l(openedCollectionsMutex);
        openedCollections.erase(coll->ice_getIdentity());
    }
 
    void
    CommonStorage::dropCollection(const std::string& collectionNS, const ::Ice::Current& c)
    {
        // authenticate if needed
        if (authenticateNS(collectionNS))
        {
            getConnection().conn().dropCollection(collectionNS);
        }
        else
        {
            throw MongoAuthenticationException("Mongo authentication failed (user = " + userName +
                                               ", collection = " + collectionNS + ")");
        }
    }
 
    std::string
    CommonStorage::getDocumentId(const mongo::BSONObj& doc)
    {
        return getDocumentField(doc, MONGO_ID_FIELD);
    }
 
    std::string
    CommonStorage::getDocumentField(const mongo::BSONObj& doc, const std::string& fieldName)
    {
        if (doc.hasField(fieldName.c_str()))
        {
            const mongo::BSONElement field = doc[fieldName.c_str()];
 
            switch (field.type())
            {
                case mongo::jstOID:
                    return field.OID().toString();
 
                case mongo::String:
                    return field.str();
 
                default:
                    return field.toString(false);
            }
        }
        else
        {
            return "";
        }
    }
 
    GridFSPtr
    CommonStorage::getGridFS(const std::string& dbName)
    {
        std::unique_lock l(openedGridFSMutex);
        std::map<std::string, GridFSPtr>::const_iterator it = openedGridFS.find(dbName);
 
        if (it != openedGridFS.end())
        {
            return it->second;
        }
        else
        {
            GridFSPtr gridFS(new mongo::GridFS(*conn, dbName));
            openedGridFS[dbName] = gridFS;
            return gridFS;
        }
    }
 
    std::string
    CommonStorage::storeFile(const std::string& dbName,
                             const std::string& fileName,
                             const ::std::string& gridFSName /* = "" */,
                             const Ice::Current& c)
    {
        GridFSPtr gridfs = getGridFS(dbName);
 
        {
            ARMARX_DEBUG << "Storing file: " << VAROUT(gridFSName) << VAROUT(fileName);
            if (!std::filesystem::exists(fileName))
            {
                throw FileNotFoundException("File could not be found: " + fileName, fileName);
            }
            std::unique_lock l(*accessGridFSFilesMutex);
            mongo::GridFile oldFile =
                gridfs->findFileByName((gridFSName.empty()) ? fileName : gridFSName);
            const mongo::BSONObj newFileDoc = gridfs->storeFile(fileName, gridFSName);
            mongo::GridFile newFile =
                gridfs->findFileByName((gridFSName.empty()) ? fileName : gridFSName);
 
            std::string oldId;
            if (keepOldFileIfEqual(oldFile, newFile, newFileDoc, dbName, oldId))
            {
                return oldId;
            }
 
            return getDocumentId(newFileDoc);
        }
    }
 
    std::string
    CommonStorage::storeTextFile(const std::string& dbName,
                                 const std::string& bufferToStore,
                                 const std::string& gridFSName,
                                 const Ice::Current& c)
    {
        GridFSPtr gridfs = getGridFS(dbName);
        {
            if (gridFSName.empty())
            {
                throw armarx::LocalException("gridFSName must not be empty");
            }
            std::unique_lock l(*accessGridFSFilesMutex);
            mongo::GridFile oldFile = gridfs->findFileByName(gridFSName);
            const mongo::BSONObj newFileDoc =
                gridfs->storeFile(bufferToStore.c_str(), bufferToStore.size(), gridFSName);
            mongo::GridFile newFile = gridfs->findFileByName(gridFSName);
 
            std::string oldId;
            if (keepOldFileIfEqual(oldFile, newFile, newFileDoc, dbName, oldId))
            {
                return oldId;
            }
            return getDocumentId(newFileDoc);
        }
    }
 
    std::string
    CommonStorage::storeBinaryFile(const std::string& dbName,
                                   const memoryx::Blob& bufferToStore,
                                   const std::string& gridFSName,
                                   const Ice::Current& c)
    {
        GridFSPtr gridfs = getGridFS(dbName);
        {
            if (gridFSName.empty())
            {
                throw armarx::LocalException("gridFSName must not be empty");
            }
            std::unique_lock l(*accessGridFSFilesMutex);
            mongo::GridFile oldFile = gridfs->findFileByName(gridFSName);
 
            const mongo::BSONObj newFileDoc = gridfs->storeFile(
                reinterpret_cast<const char*>(&bufferToStore[0]), bufferToStore.size(), gridFSName);
            mongo::GridFile newFile = gridfs->findFileByName(gridFSName);
 
            std::string oldId;
            if (keepOldFileIfEqual(oldFile, newFile, newFileDoc, dbName, oldId))
            {
                return oldId;
            }
 
            return getDocumentId(newFileDoc);
        }
    }
 
    bool
    CommonStorage::keepOldFileIfEqual(const mongo::GridFile& oldFile,
                                      const mongo::GridFile newFile,
                                      const mongo::BSONObj& newFileDoc,
                                      const std::string dbName,
                                      std::string& oldId)
    {
        if (oldFile.exists())
        {
            std::string docId = (oldFile.getFileField("_id").OID().toString());
 
            if (!docId.empty() && newFile.getMD5() == oldFile.getMD5())
            {
                // keep old file
                removeFileByQuery(dbName, newFileDoc);
                oldId = docId;
                return true;
            }
        }
        return false;
    }
 
    mongo::GridFile
    CommonStorage::getFileByQuery(const std::string& dbName, const mongo::BSONObj& query)
    {
        GridFSPtr gfs = getGridFS(dbName);
        {
            // this mutex was needed to avoid crashes when multiple clients access GridFS
            std::unique_lock l(*accessGridFSFilesMutex);
            auto find = [&]()
            {
                mongo::GridFile gf = gfs->findFile(query);
                if (!gf.exists())
                {
                    ARMARX_WARNING_S << "Could not query file." << VAROUT(dbName) << ";"
                                     << VAROUT(query);
                }
                return gf;
            };
 
            try
            {
                return find();
            }
            catch (...)
            {
                // dirty hack: try twice
                return find();
            }
        }
    }
 
    GridFileInterfacePrx
    CommonStorage::createFileProxy(mongo::GridFile gridFile, const Ice::Current& c)
    {
        if (gridFile.exists())
        {
            GridFileWrapperPtr fileWrapper = new GridFileWrapper(gridFile, accessGridFSFilesMutex);
            Ice::Identity fileIceId = fileWrapper->getIceId();
            {
                std::unique_lock l(openedFilesMutex);
                openedFiles[fileIceId] = fileWrapper;
            }
            Ice::ObjectPrx node = c.adapter->add(fileWrapper, fileIceId);
            return GridFileInterfacePrx::uncheckedCast(node);
        }
        else
        {
            ARMARX_WARNING << "Grid file does not exit " << gridFile.getFilename();
            return GridFileInterfacePrx();
        }
    }
 
    GridFileInterfacePrx
    CommonStorage::getFileProxyById(const std::string& dbName,
                                    const std::string& fileId,
                                    const Ice::Current& c)
    {
        const mongo::BSONObj query = BSON(MONGO_ID_FIELD << mongo::OID(fileId));
        mongo::GridFile gridFile = getFileByQuery(dbName, query);
        return createFileProxy(gridFile, c);
    }
 
    GridFileInterfacePrx
    CommonStorage::getFileProxyByName(const std::string& dbName,
                                      const std::string& gridFSName,
                                      const Ice::Current& c)
    {
        const mongo::BSONObj query = BSON("filename" << gridFSName);
        mongo::GridFile gridFile = getFileByQuery(dbName, query);
        return createFileProxy(gridFile, c);
    }
 
    void
    CommonStorage::releaseFileProxy(const GridFileInterfacePrx& fileProxy, const Ice::Current& c)
    {
        if (fileProxy)
        {
            std::unique_lock l(openedFilesMutex);
            c.adapter->remove(fileProxy->ice_getIdentity());
            openedFiles.erase(fileProxy->ice_getIdentity());
        }
    }
 
    bool
    CommonStorage::readTextFile(mongo::GridFile& gridFile, std::string& buffer)
    {
        if (gridFile.exists())
        {
            std::ostringstream ss;
            gridFile.write(ss);
            buffer = ss.str();
            return true;
        }
        else
        {
            return false;
        }
    }
 
    bool
    CommonStorage::readBinaryFile(mongo::GridFile& gridFile, memoryx::Blob& buffer)
    {
        if (gridFile.exists())
        {
            buffer.reserve(gridFile.getContentLength());
            std::filebuf sb;
            sb.pubsetbuf((char*)&buffer[0], buffer.capacity());
            std::iostream os(&sb);
            gridFile.write(os);
            os.flush();
            return true;
        }
        else
        {
            return false;
        }
    }
 
    bool
    CommonStorage::getTextFileById(const std::string& dbName,
                                   const std::string& fileId,
                                   std::string& buffer,
                                   const Ice::Current& c)
    {
        const mongo::BSONObj query = BSON(MONGO_ID_FIELD << mongo::OID(fileId));
        mongo::GridFile gridFile = getFileByQuery(dbName, query);
        return readTextFile(gridFile, buffer);
    }
 
    bool
    CommonStorage::getBinaryFileById(const std::string& dbName,
                                     const std::string& fileId,
                                     memoryx::Blob& buffer,
                                     const Ice::Current& c)
    {
        const mongo::BSONObj query = BSON(MONGO_ID_FIELD << mongo::OID(fileId));
        mongo::GridFile gridFile = getFileByQuery(dbName, query);
        return readBinaryFile(gridFile, buffer);
    }
 
    bool
    CommonStorage::getTextFileByName(const std::string& dbName,
                                     const std::string& gridFSName,
                                     std::string& buffer,
                                     const Ice::Current& c)
    {
        const mongo::BSONObj query = BSON("filename" << gridFSName);
        mongo::GridFile gridFile = getFileByQuery(dbName, query);
        return readTextFile(gridFile, buffer);
    }
 
    bool
    CommonStorage::getBinaryFileByName(const std::string& dbName,
                                       const std::string& gridFSName,
                                       memoryx::Blob& buffer,
                                       const Ice::Current& c)
    {
        const mongo::BSONObj query = BSON("filename" << gridFSName);
        mongo::GridFile gridFile = getFileByQuery(dbName, query);
        return readBinaryFile(gridFile, buffer);
    }
 
    void
    CommonStorage::removeFileByQuery(const std::string& dbName, const mongo::BSONObj& fileQuery)
    {
        std::string GridFsFilesNamespace = dbName + ".fs.files";
        std::string GridFsChunksNamespace = dbName + ".fs.chunks";
 
        try
        {
            auto conn = getConnection();
            std::unique_ptr<mongo::DBClientCursor> files =
                conn.conn().query(GridFsFilesNamespace, fileQuery);
 
            while (files->more())
            {
                mongo::BSONObj file = files->next();
                mongo::BSONElement id = file["_id"];
                //ARMARX_LOG << "Removing {(files)_id: " << id << "} from database [" << dbName << "]";
                conn.conn().remove(GridFsFilesNamespace, BSON("_id" << id));
                conn.conn().remove(GridFsChunksNamespace, BSON("files_id" << id));
            }
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error removing file by query " << fileQuery << ": " << e.what();
        }
    }
 
    NameList
    CommonStorage::getFileNameList(const std::string& dbName, const Ice::Current& c)
    {
        NameList result;
        auto gridfs = getGridFS(dbName);
        std::unique_ptr<mongo::DBClientCursor> list = gridfs->list();
 
        while (list->more())
        {
            mongo::BSONObj query = list->nextSafe();
            auto file = getFileByQuery(dbName, query);
            //        ARMARX_INFO << file.getFilename() << " - " << file.getContentType();
            result.push_back(file.getFilename());
        }
 
        return result;
    }
 
    NameList
    CommonStorage::getFileIdList(const std::string& dbName, const Ice::Current& c)
    {
        NameList result;
        auto gridfs = getGridFS(dbName);
        {
            std::unique_lock l(*accessGridFSFilesMutex);
 
            std::unique_ptr<mongo::DBClientCursor> list = gridfs->list();
 
            while (list->more())
            {
                mongo::BSONObj query = list->nextSafe();
                result.push_back(getDocumentId(query));
            }
        }
        return result;
    }
 
    bool
    CommonStorage::removeFileById(const std::string& dbName,
                                  const std::string& fileId,
                                  const Ice::Current&)
    {
        const mongo::BSONObj fileQuery = BSON(MONGO_ID_FIELD << mongo::OID(fileId));
        auto gridfs = getGridFS(dbName);
        {
            std::unique_lock l(*accessGridFSFilesMutex);
            mongo::GridFile gridFile = gridfs->findFile(fileQuery);
 
            if (!gridFile.exists())
            {
                return false;
            }
        }
 
        removeFileByQuery(dbName, fileQuery);
        return true;
    }
 
    bool
    CommonStorage::removeFileByName(const std::string& dbName,
                                    const std::string& gridFSName,
                                    const Ice::Current& c)
    {
        auto gridfs = getGridFS(dbName);
        {
            std::unique_lock l(*accessGridFSFilesMutex);
            gridfs->removeFile(gridFSName);
        }
        return true;
    }
 
    Ice::Int
    CommonStorage::count(const std::string& ns)
    {
        Ice::Int count = 0;
 
        try
        {
            count = getConnection().conn().count(ns);
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error on db.count(" << ns << "): " << e.what();
        }
 
        return count;
    }
 
    DBStorableData
    CommonStorage::findByMongoId(const std::string& ns, const std::string& id)
    {
        // try to init mongo id from provided string
        mongo::OID oid;
 
        try
        {
            oid.init(id);
        }
        catch (mongo::AssertionException& e)
        {
            ARMARX_ERROR << "findByMongoId failed for id " << id << ": " << e.what()
                         << "\nNS: " << ns;
            throw InvalidMongoIdException(e.what(), id);
        }
 
        mongo::Query query = mongo::Query(BSON(MONGO_ID_FIELD << oid));
        DBStorableDataList result = findByMongoQuery(ns, query, true);
 
        return result.size() > 0 ? result[0] : DBStorableData();
    }
 
    DBStorableDataList
    CommonStorage::findByFieldValue(const std::string& ns,
                                    const std::string& fieldName,
                                    const ::std::string& fieldValue)
    {
        const mongo::Query query = mongo::Query(BSON(fieldName << fieldValue));
        return findByMongoQuery(ns, query, false);
    }
 
    DBStorableDataList
    CommonStorage::findByFieldValueList(const std::string& ns,
                                        const std::string& fieldName,
                                        const NameList& fieldValueList)
    {
        mongo::BSONArrayBuilder b;
 
        for (const auto& it : fieldValueList)
        {
            b.append(it);
        }
 
        const mongo::Query query(BSON(fieldName << mongo::Query(BSON("$in" << b.arr()))));
        return findByMongoQuery(ns, query, false);
    }
 
    DBStorableData
    CommonStorage::findOneByFieldValue(const std::string& ns,
                                       const std::string& fieldName,
                                       const ::std::string& fieldValue)
    {
        const mongo::Query query = mongo::Query(BSON(fieldName << fieldValue));
        const DBStorableDataList result = findByMongoQuery(ns, query, false);
        return result.size() > 0 ? result[0] : DBStorableData();
    }
 
    DBStorableDataList
    CommonStorage::findByQuery(const std::string& ns,
                               const std::string& query,
                               const std::string& where)
    {
        mongo::Query q(query);
 
        if (!where.empty())
        {
            q.where(where);
        }
 
        return findByMongoQuery(ns, q, false);
    }
 
    DBStorableData
    CommonStorage::findOneByQuery(const std::string& ns, const std::string& query)
    {
        const DBStorableDataList result = findByMongoQuery(ns, mongo::Query(query), false);
        return result.size() > 0 ? result[0] : DBStorableData();
    }
 
    DBStorableDataList
    CommonStorage::findAll(const std::string& ns)
    {
        const mongo::Query query;
        return findByMongoQuery(ns, query, false);
    }
 
    DBStorableData
    CommonStorage::findAllUniqueByFieldName(const std::string& ns, const ::std::string& fieldName)
    {
        auto conn = getConnection();
        mongo::BSONObj fetch;
        std::size_t dotPosition = ns.find_first_of(".");
        std::string databaseName = ns.substr(0, dotPosition);
        std::string collectionName = ns.substr(dotPosition + 1, ns.size());
        conn.conn().runCommand(
            databaseName, BSON("distinct" << collectionName << "key" << fieldName), fetch);
        mongo::BSONObj fetchedValues = fetch.getObjectField("values");
        DBStorableData result;
        result.JSON = fetchedValues.jsonString();
        return result;
    }
 
    EntityIdList
    CommonStorage::findAllIds(const std::string& ns)
    {
        const mongo::Query query;
        return (EntityIdList)findFieldByMongoQuery(ns, query, MONGO_ID_FIELD);
    }
 
    NameList
    CommonStorage::findAllFieldValues(const std::string& ns, const std::string& fieldName)
    {
        const mongo::Query query;
        return findFieldByMongoQuery(ns, query, fieldName);
    }
 
    NameList
    CommonStorage::findFieldByMongoQuery(const std::string& ns,
                                         const mongo::Query& query,
                                         const std::string& fieldName)
    {
        NameList result;
        try
        {
            auto conn = getConnection();
            boost::scoped_ptr<mongo::DBClientCursor> cursor(conn.conn().query(ns, query));
 
            while (cursor->more())
            {
                result.push_back(getDocumentField(cursor->next(), fieldName));
            }
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error fetching field values by query: " << e.what();
        }
 
        return result;
    }
 
    DBStorableDataList
    CommonStorage::findByMongoQuery(const std::string& ns,
                                    const mongo::Query& query,
                                    bool justOne /* = false */)
    {
        DBStorableDataList result;
        try
        {
            auto conn = getConnection();
            boost::scoped_ptr<mongo::DBClientCursor> cursor(
                conn.conn().query(ns, query, justOne ? 1 : 0));
 
            while (cursor->more())
            {
                DBStorableData obj;
                obj.JSON = cursor->nextSafe().jsonString();
                result.push_back(obj);
            }
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error fetching objects by query: " << e.what();
        }
 
        return result;
    }
 
    std::string
    CommonStorage::insert(const std::string& ns,
                          const DBStorableData& obj,
                          bool upsert /* = false */)
    {
        std::string result = "";
 
        try
        {
            mongo::BSONObj bsonObj = mongo::fromjson(obj.JSON);
 
            // check if _id field already present; if not, it must be generated here
            if (bsonObj.hasField(MONGO_ID_FIELD))
            {
                result = getDocumentId(bsonObj);
            }
            else
            {
                // we need this trick to return generated mongoID to caller
                mongo::BSONObjBuilder builder;
                mongo::OID newID = mongo::OID::gen();
                builder.append(MONGO_ID_FIELD, newID);
                builder.appendElements(bsonObj);
                bsonObj = builder.obj();
                result = newID.toString();
            }
 
            if (upsert)
            {
                mongo::Query query = mongo::Query(BSON(MONGO_ID_FIELD << bsonObj[MONGO_ID_FIELD]));
                getConnection().conn().update(ns, query, bsonObj, true);
            }
            else
            {
                getConnection().conn().insert(ns, bsonObj);
            }
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error inserting object: " << e.what();
        }
 
        return result;
    }
 
    std::vector<std::string>
    CommonStorage::insertList(const std::string& ns, const DBStorableDataList& objectList)
    {
        std::vector<std::string> result(objectList.size(), "");
 
        try
        {
            std::vector<mongo::BSONObj> bsonObjects;
            bsonObjects.reserve(objectList.size());
 
            for (size_t i = 0; i < objectList.size(); i++)
            {
                bsonObjects.push_back(mongo::fromjson(objectList[i].JSON));
                result[i] = getDocumentId(bsonObjects[i]);
 
                // check if _id field already present; if not, it must be generated here
                if (!bsonObjects[i].hasField(MONGO_ID_FIELD))
                {
                    // we need this trick to return the generated mongoID to caller
                    mongo::OID newID = mongo::OID::gen();
                    result[i] = newID.toString();
 
                    mongo::BSONObjBuilder builder;
                    builder.append(MONGO_ID_FIELD, newID);
                    builder.appendElements(bsonObjects[i]);
                    bsonObjects[i] = builder.obj();
                }
            }
 
            getConnection().conn().insert(ns, bsonObjects);
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error inserting object: " << e.what();
        }
 
        return result;
    }
 
    bool
    CommonStorage::update(const std::string& ns,
                          const DBStorableData& obj,
                          const std::string& keyField,
                          bool upsert /*= false*/)
    {
        bool result = false;
 
        try
        {
            const mongo::BSONObj mongoObj = mongo::fromjson(obj.JSON);
 
            if (!mongoObj.hasField(keyField.c_str()))
            {
                throw FieldNotFoundException("field not found in supplied JSON object", keyField);
            }
 
            mongo::Query query(BSON(keyField << mongoObj[keyField]));
            {
                getConnection().conn().update(ns, query, mongoObj, upsert);
            }
            result = true;
        }
        catch (const mongo::DBException& e)
        {
            ARMARX_ERROR << "Error updating object: " << e.what();
        }
 
        return result;
    }
 
    bool
    CommonStorage::updateByQuery(const std::string& ns,
                                 const std::string& query,
                                 const mongo::BSONObj& obj)
    {
        bool result = false;
 
        try
        {
            conn->update(ns, query, obj);
            result = true;
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error updating object: " << e.what();
        }
 
        return result;
    }
 
    bool
    CommonStorage::removeByMongoId(const std::string& ns, const std::string& id)
    {
        // try to init mongo id from provided string
        mongo::OID oid;
 
        try
        {
            oid.init(id);
        }
        catch (mongo::AssertionException& e)
        {
            throw InvalidMongoIdException(e.what(), id);
        }
 
        mongo::Query query(BSON(MONGO_ID_FIELD << oid));
        return removeByMongoQuery(ns, query);
    }
 
    bool
    CommonStorage::removeByFieldValue(const std::string& ns,
                                      const std::string& fieldName,
                                      const std::string& fieldValue)
    {
        const mongo::Query query(BSON(fieldName << fieldValue));
        return removeByMongoQuery(ns, query);
    }
 
    bool
    CommonStorage::removeByQuery(const std::string& ns, const std::string& query)
    {
        return removeByMongoQuery(ns, mongo::Query(query));
    }
 
    bool
    CommonStorage::clearCollection(const std::string& ns)
    {
        return removeByMongoQuery(ns, mongo::Query());
    }
 
    bool
    CommonStorage::removeByMongoQuery(const std::string& ns, const mongo::Query& query)
    {
        int result = false;
 
        try
        {
            getConnection().conn().remove(ns, query);
            result = true;
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error deleting objects by query: " << e.what();
        }
 
        return result;
    }
 
    bool
    CommonStorage::ensureIndex(const std::string& ns, const std::string& fieldName, bool unique)
    {
        int result = false;
 
        try
        {
            const mongo::BSONObj keys = BSON(fieldName << 1);
            // the ubuntu 14.4 version of the lib does not has a define to querry the version
            // due to breaking changes in the interface, we have to call different functions here.
            // so we use the macro MONGOCLIENT_VERSION to decide which interface version we use
#ifdef MONGOCLIENT_VERSION
            getConnection().conn().createIndex(ns, keys);
#else
            getConnection().conn().ensureIndex(ns, keys, unique);
#endif
 
            result = true;
        }
        catch (mongo::DBException& e)
        {
            ARMARX_ERROR << "Error ensuring index: " << e.what();
        }
 
        return result;
    }
 
    CommonStorage::ConnectionWrapper::ConnectionWrapper(
        CommonStorage& storage,
        std::shared_ptr<mongo::DBClientConnection> connPtr) :
        connPtr{std::move(connPtr)}, hostAndPort{storage.hostAndPort}, storage{&storage}
    {
    }
 
    CommonStorage::ConnectionWrapper::~ConnectionWrapper()
    {
        std::lock_guard<std::mutex> guard{storage->serverSettingsMutex};
        if (hostAndPort == storage->hostAndPort)
        {
            storage->pool.emplace_back(std::move(connPtr));
        }
    }
 
    mongo::DBClientConnection&
    CommonStorage::ConnectionWrapper::conn()
    {
        return *connPtr;
    }
} // namespace memoryx