VFS_mongo.cpp

Go to the documentation of this file.

 
#include "VFS.h"
#include "utilities/rutils.h"
 
#include "VFS_base/VFS_node_type.h"
#include "VFS_base/VFS_icons.h"
 
#include "VFS_mongo.h"
 
#include <QFileInfo>
 
#include <iostream>
 
#ifndef Q_MOC_RUN
#include <bsoncxx/json.hpp>
#include <bsoncxx/exception/exception.hpp>
#include <mongocxx/exception/exception.hpp>
#include <mongocxx/exception/query_exception.hpp>
#include <mongocxx/exception/logic_error.hpp>
#endif
 
//this is bullshit... their driver should manage this
mongocxx::instance VFS_mongo::_instance = std::move(mongocxx::instance::current());
 
VFS_mongo_cache_entry::VFS_mongo_cache_entry(VFS_mongo *mongo, QString path, bool debug, int flushInterval, int expireInterval, bool create, bool container, QJsonObject createData)
: VFS_datastore_cache_entry(debug,flushInterval,expireInterval)
, _mongo(mongo)
, _path(path)
, _container(container)
{
        if (_debug)
                VFS::WARN( QString("Creating mongo cache entry: '%1'").arg(_path) );
 
        try //try to read the document
        {
                QJsonObject o = _mongo->readDocument(_path,_size,_container);
 
                //printf("CACHE DATA: %s\n%s\n%s\n",qUtf8Printable(_path),qUtf8Printable(b),qUtf8Printable( QJsonDocument(o).toJson() ));
 
                setData(o);
 
                _dirty = false;
                _valid = true;
        }
        catch (VFS_mongo_collectionException &e)        //exception meaning that the collection doesn't exist
        {
                VFS::ERROR( _reason = QString("Collection was not found, and createCollection was not set: %1").arg(e.what()));
                _valid = false;
        }
        catch (VFS_mongo_entryException &e)
        {
                if (create)
                {
                        try
                        {
                                if (_mongo->createDocument(_path,createData,_size,container))
                                {       setData(createData);
                                        _valid = true;
                                        _dirty = false;
                                        return;
                                }
                                else
                                {
                                        _valid = false;
                                }
                        }
                        catch(VFS_mongo_exception &e)
                        {
                                _reason = QString("VFS_mongo cache exception: %1").arg(e.what());
                                _valid = false;
                        }
                        catch(...)
                        {
                                _reason = "VFS_mongo cache exception, and no message was available. (2a) "+_path;
                                _valid = false;
                        }
                }
                else
                {
                        //VFS::ERROR( _reason = "Entry was not found and create was not set: "+_path);
                        _valid = false;
                }
 
                _reason = QString("VFS_Mongo cache entryException: %1").arg(e.what());//+_path;
                _valid = false;
        }
        catch (VFS_mongo_directoryException &e)         //exception meaning that the path was too short
        {
                _reason = QString("VFS_Mongo cache directoryException: %1").arg(e.what());//+_path;
                _valid = false;
        }
        catch (bsoncxx::exception &e)
        {
                _reason = e.what();
                _dirty = true;
                _valid = false;
                return;
        }
        catch (std::exception &e)
        {
                _reason = QString("VFS_mongo cache exception: %1").arg(e.what());
                _valid = false;
        }
        catch (...)
        {
                _reason = "VFS_mongo cache exception, and no message was available. (1) "+_path;
                _valid = false;
        }
}
 
 
VFS_mongo_cache_entry::~VFS_mongo_cache_entry()
{
        if (_debug)
                VFS::WARN( QString("Cache deleted: '%1'").arg(_path) );
 
        if (_dirty && _valid)
                flush();
}
 
 
bool VFS_mongo_cache_entry::valid(QString &reason)
{
        if (_valid)
                return _valid;
 
        return VFS_datastore_cache_entry::valid(reason);
}
 
bool VFS_mongo_cache_entry::isContainer()
{
        return _container;
}
 
 
void VFS_mongo_cache_entry::flush()
{
        if (_debug)
                VFS::WARN( QString("Flushing '%1'").arg(_path) );
 
        if (!_dirty)
                return;
 
        try
        {
                _size = _mongo->writeDocument(_path, _data.object() );
                _dirty = false;
        }
        catch (bsoncxx::exception &e)
        {
                _reason = e.what();
                _dirty = true;
                _valid = false;
        }
        catch (std::exception& e)
        {
                _reason = e.what();
                _dirty = true;
                _valid = false;
        }
        catch (...)
        {
                VFS::ERROR( "Caught unspecified mongo exception while flushing "+_path );
                //issue a VFS::CRITICAL message if disk is full or other deep failure happens
                VFS::CRITICAL( _reason = QString("Unable to write to %1... Is the disk full or a permission problem?").arg(_path) );
                _dirty = true;
                _valid = false;
        }
}
 
 
VFS_mongo::VFS_mongo(QString url, QString db, QString collection, quint64 size, bool debug, int flushInterval, int expireInterval, bool createCollection)
: VFS_datastore(size, debug)
, _flushInterval(flushInterval)
, _expireInterval(expireInterval)
, _url(url)
, _databaseName(db)
, _collection(collection)
, _createCollection(createCollection)
{
        try
        {
                _client = client( uri(qUtf8Printable(url)) );
                _database = _client[ qUtf8Printable(db) ];
                VFS::LOG( QString("Created %1 node pointing to '%2/%3@%4'").arg(className()).arg(url).arg(db).arg(collection) );
        }
        catch (mongocxx::exception& e)
        {
                VFS::ERROR( e.what() );
        }
        catch (...)
        {
                VFS::ERROR( "Caught unspecified mongo error." );
        }
}
 
VFS_mongo::~VFS_mongo()
{
 
}
 
bool VFS_mongo::createDocument(QString path, QJsonObject o, quint64 &size, bool container)
{
        if (!_database.has_collection(_collection.toStdString()))
                if (!_createCollection || !createCollection(_collection))
                        throw VFS_mongo_collectionException("Collection '"+_collection+"' doesn't exist.");
 
        if (path == "")
                throw VFS_mongo_directoryException("Path was empty.");
 
        try                                             //query to see if path already exists
        {
                bool c;
                readDocument(path,size,c);              //if read is successful
 
                if (c!=container)
                        VFS::WARN("tried to create a container when a file already existed.");
 
                return false;                           //don't create the document, as it already exists
        }
        catch (VFS_mongo_entryException)                //if the path does not exist, create it, and check/create all parent directories as well
        {
                QStringList pp = path.split("/",Qt::SkipEmptyParts);
 
                if (pp.length())
                        pp.takeLast();                  //remove the actual entry we are creating
 
                QString p;
                quint64 s;
                bool c;
                while (pp.length())                     //ensure all parent directories are created
                {
                        p = rutils::cleanPath( p + "/" + pp.takeFirst() );
 
                        try
                        {
                                readDocument(p,s,c);
                        }
                        catch (VFS_mongo_entryException)
                        {
                                //printf("Creating parent entry: %s\n",qUtf8Printable(p));
                                writeDocument(p,QJsonObject(),true);
                        }
                }
 
                size = writeDocument(path,o,container);
        }
 
        return true;
}
 
 
quint64 VFS_mongo::writeDocument(QString path, QJsonObject o, bool container)
{
        if (!_database.has_collection(_collection.toStdString()))
                if (!_createCollection || !createCollection(_collection))
                        throw VFS_mongo_collectionException("Collection '"+_collection+"' doesn't exist.");
 
        if (path == "")
                throw VFS_mongo_directoryException("Path was empty.");
 
        o.remove("_id");                //never ever ever update or create the _id field!
        o["_path"] = path;
 
        if (container)
                o["_container"] = true;
 
        QByteArray j = QJsonDocument(o).toJson();
 
        bsoncxx::document::value u = bsoncxx::from_json( j.toStdString() );
        bsoncxx::document::value f = bsoncxx::builder::stream::document{} << "_path" << path.toStdString() << bsoncxx::builder::stream::finalize;
 
        options::update op;
        op.upsert(true);
 
        //bsoncxx::stdx::optional<mongocxx::result::update> updated = _database[_collection.toStdString()].update_one( f.view(), u.view(), op );
        bsoncxx::stdx::optional<mongocxx::result::replace_one> updated = _database[_collection.toStdString()].replace_one( f.view(), u.view(), op );
 
        if (!updated)
                throw VFS_mongo_entryException("Unable to write to '"+path+"'.");
 
        return (quint64) j.size();
}
 
 
bool VFS_mongo::removeDocument(QString path)
{
        if (!_database.has_collection(_collection.toStdString()))
                if (!_createCollection || !createCollection(_collection))
                        throw VFS_mongo_collectionException("Collection '"+_collection+"' doesn't exist.");
 
        bsoncxx::document::view_or_value f = bsoncxx::builder::stream::document{} << "_path" << path.toStdString() << bsoncxx::builder::stream::finalize;
 
        // FIXME: should not allow deleting a "_container" if it has children entries
        // this will require a query first and then a decision to delete or not
 
        bsoncxx::stdx::optional<mongocxx::result::delete_result> deleted = _database[_collection.toStdString()].delete_one( f.view() );
 
        if (!deleted)
                throw VFS_mongo_entryException("Unable to remove '"+path+"'.");
 
        return true;
}
 
 
QJsonObject VFS_mongo::readDocument(QString path, quint64 &size, bool &container)
{
        if (!_database.has_collection(_collection.toStdString()))
                if (!_createCollection || !createCollection(_collection))
                        throw VFS_mongo_collectionException("Collection '"+_collection+"' doesn't exist.");
 
        if (path == "")                 //if the path is empty, the collection is a container
        {       size = 0;
                container = true;
                return QJsonObject();
        }
 
        bsoncxx::document::value f = bsoncxx::builder::stream::document{} << "_path" << path.toStdString() << bsoncxx::builder::stream::finalize;
 
        bsoncxx::stdx::optional<bsoncxx::document::value> maybe_result = _database[_collection.toStdString()].find_one(std::move(f));
        if (maybe_result)
        {
                bsoncxx::document::value v = maybe_result.value();
 
                QByteArray b = QByteArray::fromStdString(bsoncxx::to_json(v));  //I wish there was a better way here!
                QJsonDocument d = QJsonDocument::fromJson(b);
                QJsonObject o = d.object();
 
                size = (quint64) b.size();
                container = o.value("_container").toBool(false);
 
                o.remove("_id");
                o.remove("_path");
                o.remove("_container");
 
                return o;
        }
        else
                throw VFS_mongo_entryException("Unable to read '"+path+"'.");
}
 
 
bool VFS_mongo::createCollection( QString name )
{
        bool success = false;
 
        try
        {
                _database.create_collection(name.toStdString());
 
                auto index_specification = bsoncxx::builder::stream::document{};
                index_specification << "_path" << 1;
                bsoncxx::document::value ii = index_specification << bsoncxx::builder::stream::finalize;
 
                try {
                        bsoncxx::stdx::optional<bsoncxx::document::value> maybe_result = _database[name.toStdString()].create_index(std::move(ii));
 
                        if (!maybe_result)
                        {       VFS::ERROR("(1) Could not index collection "+name);
                                return false;
                        }
                }
                catch (mongocxx::logic_error &e)                //if the options are invalid.
                {
                        VFS::ERROR("(2) Could not index collection "+name);
                        VFS::ERROR(e.what());
 
                        return false;
                }
                catch (mongocxx::operation_exception &e)        //if index creation fails.
                {
                        VFS::ERROR("(3) Could not index collection "+name);
                        VFS::ERROR(e.what());
 
                        return false;
                }
                catch (...)
                {
                        VFS::ERROR("Unknown error when creating indexes on '"+name+"'.  Behavior will be undefined.");
 
                        return false;
                }
 
                VFS::WARN("Created and indexed collection '"+name+"'", 0, className());
                success = true;
        }
        catch (mongocxx::operation_exception &e)
        {
                VFS::ERROR("Could not create collection "+name);
                VFS::ERROR(e.what());
        }
        catch (...)
        {
                VFS::ERROR("Could not create collection "+name);
        }
 
        return success;
}
 
 
QByteArray VFS_mongo::icon()
{
        return VFS_icons::get("disk");
}
 
QString VFS_mongo::reportDetails()
{
        QString s = VFS_datastore::reportDetails();
 
        return s;
        //return s+"\nFIXME: calculate cache usage and file count here!";
}
 
 
void VFS_mongo::ls(VFS_request *r)
{
        QMutexLocker _l(&_lock);
 
        QString path = r->_path;
 
        QJsonObject out;
        bool container(false);
 
        try
        {
        //check collection
                if (!_database.has_collection(_collection.toStdString()))
                        if (!_createCollection || !createCollection(_collection))
                        {       r->_success = false;
                                r->_reason = QString("Unable to create collection '%1'").arg(_collection);
                                return;
                        }
 
        //ignore the "closest" flag for now
 
        //query
                QString query;
                if (path=="")
                {       query = QString("                                       \
                        {       \"_path\": {    \"$regex\": \"^[^/]*$\",        \
                                                \"$options\": \"\"      }       \
                        }").arg(path);
                }
                else
                {       query = QString("                                       \
                        {       \"_path\": {    \"$regex\": \"^%1/[^/]*$\",     \
                                                \"$options\": \"\"      }       \
                        }").arg(path);
                }
 
                bsoncxx::document::view_or_value q = {};
                q = bsoncxx::from_json( query.toStdString() );
 
                //printf("path: %s\nquery: %s\n",qUtf8Printable(path),qUtf8Printable(query));
 
        //projection
                QString projection("{ \"_path\": 1, \"_container\":1 }");       //we only care about the path and container-ness
 
                bsoncxx::document::view_or_value p = bsoncxx::from_json( projection.toStdString() );
 
                options::find o;
                o.projection(p);
 
        //perform query
                //this should use an aggregator
 
                cursor c = _database[_collection.toStdString()].find(q,o);
                //throw VFS_mongo_entryException("Unable to read '"+path+"'.");
 
                QString m;
                QStringList pp;
                bsoncxx::document::view v;
                bsoncxx::document::element e;
                cursor::iterator i = c.begin();
                while (i != c.end())
                {
                        //std::cout << bsoncxx::to_json(*i) << std::endl;
 
                        v = *i;
 
                        e = v["_container"];
 
                        if (e && e.type() == bsoncxx::type::k_bool)
                                container = e.get_bool();
                        else
                                container = false;
 
                        e = v["_path"];
                        m = "";
 
                        if (e)  //if element is valid
                        {
                                if (e.type() == bsoncxx::type::k_utf8)
                                        m = QString::fromStdString(e.get_utf8().value.to_string());
 
                                if (m != "")
                                {
                                        //remove the path p from the beginning of the string
                                        m.remove(0,path.length());
                                        if (m.startsWith("/"))
                                                m.remove(0,1);
 
                                        //find the directory or file name, which is now the beginning of the string
                                        pp = m.split("/",Qt::SkipEmptyParts);
                                        if (pp.length() == 1)
                                                out[m] = container;
                                }
                        }
 
                        i++;
                }
 
                //printf("OUT: %s\n",qUtf8Printable(QJsonDocument(out).toJson()));
        }
        catch (mongocxx::logic_error &e)
        {
                VFS::ERROR( r->_reason = r->_path+": "+e.what() );
                r->_success = false;
                return;
        }
        catch (...)
        {
                VFS::ERROR( r->_reason = r->_path+": Caught unspecified mongo error." );
                r->_success = false;
                return;
        }
 
        if (r->_metadata.value("sequence").toBool(false))
                out = rutils::sequenceListing(out);
 
        r->_data.setObject(out);
        r->_success = true;
}
 
 
void VFS_mongo::read(VFS_request *r)
{
        QMutexLocker l(&_lock);
 
        //printf("VFS_MONGO read %s\n",qUtf8Printable(r->toJson(0,true,true)));
 
        QString p = r->_path;
 
        bool creating = r->_metadata.value("createIfMissing").toBool(false);
        bool container = r->_metadata.value("container").toBool(false);
        bool cache = r->_metadata.value("cache").toBool(true);
 
        VFS_mongo_cache_entry *en = dynamic_cast<VFS_mongo_cache_entry *>( _cache[r->_path] );
        if (!en)
        {
                en = new VFS_mongo_cache_entry(this,p,_debug,_flushInterval,_expireInterval,creating,container,r->_data.object());
                //printf("MONGO DATASTORE READ: %d %p '%s'\n%s",creating,this,qUtf8Printable(r->_path),qUtf8Printable(QJsonDocument(en->getData()).toJson()));
 
                if (!en->valid(r->_reason))
                {
                        //printf("MONGO DATASTORE INVALID\n");
 
                        delete en;
 
                        if (creating)
                        {
                                //printf("CREATE FROM READ: %s\n%s\n",qUtf8Printable(r->_path),qUtf8Printable(r->_data.toJson()));
 
                                VFS_request::requestType t = r->_requestType;
                                r->_requestType = VFS_request::create;
                                write(r);
                                r->_requestType = t;
                                return;
                        }
                        else
                        {
                                //ls(r);
                                //printf("MONGO READ LS:%s\n",qUtf8Printable(r->toJson(0,true,true)));
                                r->_success = false;
                                return;
                        }
                }
 
                if (en->isContainer())
                {
                        ls(r);
                        //en->setData(r->_data.object());
                        //printf("MONGO READ LS:%s\n",qUtf8Printable(r->toJson(0,true,true)));
                        return;         //we have the data... don't cache it, just return.
                }
 
                if (cache)
                if (!_cache.insert(r->_path,en))
                        VFS::WARN( QString("%1::read(): '%2' was too large to cache!").arg(className()).arg(r->_path) );
        }
        else
        {
                //printf("MONGO CACHE was valid\n");
        }
 
        r->_data = en->getData();
        r->_success = true;
}
 
 
void VFS_mongo::write(VFS_request *r)
{
        QMutexLocker l(&_lock);
 
        QString p = r->_path;
        QJsonObject m = r->_metadata;
        bool creating = r->_requestType == VFS_request::create;
        bool container = r->_metadata.value("container").toBool(false);
 
        VFS_mongo_cache_entry *en = dynamic_cast<VFS_mongo_cache_entry *>( _cache[r->_path] );
        if (!en)
        {
                en = new VFS_mongo_cache_entry(this,p,_debug,_flushInterval,_expireInterval,creating,container,r->_data.object());
 
                if (!en->valid(r->_reason))
                {       r->_success = false;
                        if (creating)
                        {       //VFS::ERROR("report some bad create error here. "+r->_reason);
                        }
                        delete en;
                        return;
                }
 
                if (creating)
                {
                        if (_debug)
                                VFS::WARN( QString("Creating: '%1'").arg(p) );
 
                        r->_data.setObject( en->getData().object() );
 
                        //create a diff for anything that is subscribed to the containing directory
 
                        QStringList dirs = r->_path.split("/",Qt::SkipEmptyParts);
                        QString newFile = dirs.takeLast();
                        QString dirPath = dirs.join("/");
 
                        QJsonObject o;
                        o[newFile] = true;
                        //FIXME: may need to write metadata here
                        VFS_request *c = createRequest(r->_requestType,dirPath,r->_user,QJsonDocument(o));
                        emit diff(this,c);
                        delete c;
 
                        if (!_cache.remove(dirPath))                                            //it will be re-created if needed
                        {       //printf("Unable to remove cache entry: %s\n",qUtf8Printable(dirPath));
                        }
                }
 
                if (!_cache.insert(en->_path,en))
                        VFS::WARN( QString("%1::write(): '%2' was too large to cache!").arg(className()).arg(r->_path) );
        }
 
        en->setData(r->_data.object());
        //en->setMetadata(r->_metadata);
 
        r->_success = true;
 
        emit diff(this,r);
}
 
 
void VFS_mongo::metadata(VFS_request *r)
{
        QMutexLocker l(&_lock);
 
        //printf("VFS_MONGO metadata %s\n",qUtf8Printable(r->_path));
 
        QString p = r->_path;
 
        QJsonObject mt = r->_metadata;
 
        VFS_datastore::metadata(r);
 
        r->_metadata["type"] = VFS_node_type::getType(p);
        r->_success = true;
}
 
 
void VFS_mongo::submit(VFS_request *r)
{
        QMutexLocker l(&_lock);
 
        QString p = r->_path;
        bool creating = r->_metadata.value("createIfMissing").toBool(false);
        bool container = r->_metadata.value("container").toBool(false);
 
        VFS_datastore_cache_entry *en = _cache[r->_path];
        if (!en)
        {
                en = new VFS_mongo_cache_entry(this, p,_debug,_flushInterval,_expireInterval,creating,container,r->_data.object());
 
                if (!en->valid(r->_reason))
                {
                        delete en;
 
                        if (creating)
                        {
                                //printf("CREATE FROM SUBMIT: %s\n%s\n",qUtf8Printable(r->_path),qUtf8Printable(r->_data.toJson()));
 
                                VFS_request::requestType t = r->_requestType;
                                r->_requestType = VFS_request::create;
                                write(r);
                                r->_requestType = t;
                                return;
                        }
                        else
                        {
                                r->_success = false;
                                return;
                        }
                }
 
                if (!_cache.insert(r->_path,en))
                        VFS::WARN( QString("%1::submit(): '%2' was too large to cache!").arg(className()).arg(r->_path) );
        }
 
        QJsonDocument fd = en->getData();
        QJsonObject o;
 
        o = applyJsonDiff( fd.object(), r->_data.object(), "", r->_user );
 
        //printf("Diffed: %s\n",qUtf8Printable( QJsonDocument(o).toJson() ));
 
        en->setData(o);
 
        r->_success = true;
 
        emit diff(this,r);
}
 
 
void VFS_mongo::rm(VFS_request *r)
{
        //check if path is empty... that means that this node is to be removed
        if (r->_path=="")
        {
                VFS_node::rm(r);
                return;
        }
 
        bool s = false;
 
        try
        {
                _cache.remove(r->_path);                //hate to flush and then remove, but there doesn't seem a better way.
                s = removeDocument(r->_path);
        }
        catch (VFS_mongo_entryException)
        {
                s = false;
        }
        catch (...)
        {
                s = false;
        }
 
        if (s)
        {
                QStringList dirs = r->_path.split("/",Qt::SkipEmptyParts);
                QString rmFile = dirs.takeLast();
                QString dirPath = dirs.join("/");
 
                _cache.remove(dirPath);                                                                         //it will be re-created if needed.
 
                QJsonDocument diffData;
                QJsonObject o;
                o[ rmFile ] = QJsonValue::Null;         //null means remove
                diffData.setObject(o);
 
                unsubscribePath(r->_path);                                                                      //unsubscribePath on this path... it no longer exists!
 
                r->_data = diffData;
                r->_path = dirPath;
                r->_success = true;
 
                emit diff(this,r);
        }
        else
        {
                r->_reason = QString("Can't delete non-existent file: '%1'").arg(r->_path);
                r->_success = false;
        }
}
 
 
void VFS_mongo::subscribe(VFS_request *r)
{
        VFS_node::subscribe(r);
}
 
 
VFS_mongo_exception::VFS_mongo_exception(QString m) : _message(m)
{
        if (!_message.endsWith('\n'))
                _message += "\n";
}
 
VFS_mongo_exception::~VFS_mongo_exception() throw (){}
 
const char *VFS_mongo_exception::what() const throw()
{
        return qUtf8Printable(_message);
}
 
VFS_mongo_entryException::VFS_mongo_entryException(QString m) : VFS_mongo_exception(m) {}
 
VFS_mongo_entryException::~VFS_mongo_entryException() throw (){}
 
VFS_mongo_directoryException::VFS_mongo_directoryException(QString m) : VFS_mongo_exception(m) {}
 
VFS_mongo_directoryException::~VFS_mongo_directoryException() throw (){}
 
VFS_mongo_collectionException::VFS_mongo_collectionException(QString m) : VFS_mongo_exception(m) {}
 
VFS_mongo_collectionException::~VFS_mongo_collectionException() throw (){}