VFS_tcp_mount.cpp

Go to the documentation of this file.

 
#include "VFS_tcp_mount.h"
#include "VFS_base/VFS_icons.h"
#include "utilities/rutils.h"
#include "VFS_creator.h"
 
#include <QJsonDocument>
#include <QJsonObject>
 
#include "VFS.h"
 
QMutex VFS_tcp_mount::_mountsLock;
QList<VFS_tcp_mount *> VFS_tcp_mount::_mounts;
 
 
 VFS_tcp_mount::VFS_tcp_mount(QString name, QString path, QString address, quint16 port, quint16 interval, bool ssl, QString sslCertPath)
: VFS_tcp_client(address,port,ssl,sslCertPath)
, _name(name)
, _path(path)
, _attemptInterval(interval)
, _timerID(0)
, _mounted(false)
, _currentID(POST_ID_START)
, _codeDirectory(this)
{
        QMutexLocker l(&_mountsLock);
 
        _mounts << this;
}
 
 VFS_tcp_mount::~VFS_tcp_mount()
{
        QMutexLocker l(&_mountsLock);
 
        _mounts.removeAll(this);
}
 
 
void  VFS_tcp_mount::init()
{
        if (_initialized)
                return;
 
        connect(        this,           SIGNAL(connected()),
                        this,           SLOT(stopConnecting())          );
 
        connect(        this,           SIGNAL(connected()),
                        this,           SLOT(startPing())               );
 
        connect(        _socket,        SIGNAL(disconnected()),
                        this,           SLOT(startConnecting())         );
 
        connect(        _socket,        SIGNAL(disconnected()),
                        this,           SLOT(stopPing())                );
 
        connect(        this,           SIGNAL(readyMessage(QByteArray)),
                        this,           SLOT(receiveMessage(QByteArray)));
 
        connect(        &_ping,         SIGNAL(timeout()),
                        this,           SLOT(sendPing())                );
 
        attemptConnection();
        startConnecting();
 
        VFS_tcp_client::init();
}
 
 
void  VFS_tcp_mount::timerEvent(QTimerEvent *)
{
        attemptConnection();
 
        //more stuff here...?
}
 
 
void  VFS_tcp_mount::startConnecting()
{
        if (_mounted)
        {       VFS::WARN( QString("Lost conection to '%1:%2'... reconnecting.").arg(_address).arg(_port), 0, className() );
 
                QMutexLocker l(&_lock);
 
                VFS_request *r;
                QMap<postID, VFS_request *>::const_iterator it;
                for (it=_requests.begin();it!=_requests.end();it++)
                {       r = it.value();
                        r->_success = false;
                        issueResponse(r);
                        //it.value()->fail();
                }
 
                _requests.clear();
 
                _codeDirectory.unregisterDirectory();
 
                issueUnmounted();
        }
 
        if (!_timerID)
                _timerID = startTimer(_attemptInterval);
 
        _mounted = false;
}
 
 
void  VFS_tcp_mount::stopConnecting()
{
        if(_timerID)
        {
                killTimer(_timerID);
 
                QAbstractSocket::SocketState s = _socket->state();
 
                if (s==QAbstractSocket::ConnectedState)
                {
                        VFS::LOG( QString("Connected to '%1' @ '%2:%3' (%4)").arg(_name).arg(_address).arg(_port).arg(_ssl?"secure":"insecure"), 0, className() );
                        _mounted = true;
 
                        refreshSubscriptions();
                        sendUnmountedRequests();
                        //issueMounted();               // can't do this here... have to wait for subscription to complete
                }
        }
 
        _timerID = 0;
}
 
 
void  VFS_tcp_mount::attemptConnection()
{
        QAbstractSocket::SocketState s = _socket->state();
 
        if (s!=QAbstractSocket::ConnectingState && s!=QAbstractSocket::ConnectedState)
        //if (s == QAbstractSocket::UnconnectedState)
        {
                VFS::LOG( QString("Attempting to mount '%1 @ %2:%3' (%4)...").arg(_name).arg(_address).arg(_port).arg(_ssl?"secure":"insecure"), 9, className() );
 
                if (_ssl)
                {
                        QSslSocket *_sslSocket = dynamic_cast<QSslSocket *> (_socket);
                        _sslSocket->connectToHostEncrypted(_address,_port);
                }
                else
                {
                        _socket->connectToHost(_address,_port);
                }
        }
        else
        {
                VFS::WARN( QString("Already connected or connecting '%1:%2' (%3)").arg(_address).arg(_port).arg(s), 0, className() );
                if (s==QAbstractSocket::ConnectedState)
                        stopConnecting();
        }
}
 
 
void VFS_tcp_mount::startPing()
{
        _ping.start(VFS_TCP_MOUNT_PING_TIME_MS);
}
 
 
void VFS_tcp_mount::stopPing()
{
        _ping.stop();
}
 
 
void VFS_tcp_mount::sendPing()
{
        QJsonObject m;
 
        m["id"] = POST_ID_PING;
 
        //QString message = json::encode(m);
 
        QJsonDocument d( m );
        QByteArray message = d.toJson();
 
        sendMessage(message);
}
 
 
bool VFS_tcp_mount::isContainer()
{
        return true;
}
 
 
void VFS_tcp_mount::executeRequest(VFS_request *r)
{
        if (_mounted)
        {
                postID i = getNewRequestID();
 
                //remember the callback object for when the response comes
                _requests.insert(i,r);
                //printf("request list length: %d\n",_requests.size());
 
                switch(r->_requestType)
                {
                        case VFS_request::submit:       {
                                                                //QString subscriberSubpath = rutils::cleanPath( r->_originPath );
                                                        //      r->_notifyExceptions << notifyException(r->_origin,r->_initialPath);
                                                                //r->_notifyExceptions << notifyException(this,r->_initialPath);
                                                                //r->_notifyExceptions << notifyException(this,subscriberSubpath);
                                                                //r->_notifyExceptions << notifyException(this,r->_path);
                                                                //r->_notifyExceptions << notifyException(r->_origin,r->_path);
                                                                //printf("submit exception: %p %s\n",r->_origin,qUtf8Printable(r->_path));
                                                        }
 
                                                        break;
 
                        case VFS_request::subscribe:    {
                                                                //we want a modified version of subscribe here, where the entry is created but read() and metadata() are not called.
                                                                VFS_node *subscriber = r->_origin;
 
                                                                if (!__isNode(subscriber))
                                                                {
                                                                        _requests.remove(i);
                                                                        delete r;
                                                                        return;
                                                                }
 
                                                                QString subscriberSubpath = rutils::cleanPath( r->_originPath );
                                                                int count = r->_metadata.value("subscriptions").toInt(1);
 
                                                                if (r->_metadata.contains("resubscription") && r->_metadata["resubscription"].toBool())
                                                                {       count = 0;
                                                                        //r->_metadata.remove("resubscription");
                                                                }
 
                                                                _subscribers[r->_path][subscriberSubpath][subscriber]+=count;           //will create default/empty values as needed.
                                                                connect( subscriber, SIGNAL(unmounted(VFS_node *)), this, SLOT(unsubscribeAll(VFS_node *)), Qt::UniqueConnection );
 
                                                                //FIXME: should we just do a subscribe(r) with flags to skip metadata() and read()?
                                                                //this version will call read() and metadata(), which we don't want
                                                                //subscribe(r);
                                                        }
 
                                                        break;
 
                        case VFS_request::unsubscribe:  {
                                                                unsubscribe(r);
                                                        }
                                                        break;
 
                        case VFS_request::report:       {
                                                                VFS_tcp_client::executeRequest(r);
                                                                //report(r);
                                                                //issueResponse(r);
                                                                _requests.remove(i);
                                                                //delete r;
                                                                return;
                                                        }
                                                        //break;
 
                        default:                        break;
                }
 
                QByteArray message = r->toJson(i,true);//,initial);
 
                sendMessage(message);
 
                startPing();                            //restart the _ping timer
        }
        else
        {
                switch(r->_requestType)
                {
                /*
                        case VFS_request::none:
                //      case VFS_request::subscribe:
                                                        {
                                                                //we want a modified version of subscribe here, where the entry is created but read() and metadata() are not called.
                                                                VFS_node *subscriber = r->_origin;
                                                                QString subscriberSubpath = rutils::cleanPath( r->_originPath );
                                                                int count = r->_metadata.value("subscriptions").toInt(1);
                                                                _subscribers[r->_path][subscriberSubpath][subscriber]+=count;           //will create default/empty values as needed.
                                                                connect( subscriber, SIGNAL(unmounted(VFS_node *)), this, SLOT(unsubscribeAll(VFS_node *)), Qt::UniqueConnection );
 
                                                                r->_dontDelete = true;
                                                                issueResponse(r);
 
                                                                VFS_tcp_client::executeRequest(r);
                                                        }
 
                                                        break;
                */
 
                //      case VFS_request::none:
                //      case VFS_request::subscribe:
                //      case VFS_request::unsubscribe:
                //      case VFS_request::diff:
                        case VFS_request::report:
                //      case VFS_request::acl:
                                                        VFS_tcp_client::executeRequest(r);
                                                        break;
 
                        default:                        //don't call VFS_node::executeRequest()...
                                                        //and don't issueResponse()...
                                                        //let the stored _requests wait for the response from the other end
 
                                                        VFS::WARN( QString("%1 '%2' request on '%3' queued for when mounted.").arg(className()).arg(VFS_request::requestTypeStrings[r->_requestType]).arg(r->_path), 9, r->_user );
                                                        _unmountedRequests << r;
                                                        break;
                }
        }
}
 
 
void VFS_tcp_mount::receiveMessage(QByteArray message)
{
        #ifdef COMPRESS_TCP_MOUNT
        message = qUncompress(message);
        if (message.length() == 0)
                VFS::ERROR("(2) Uncompressed message size was 0.");
        #endif
 
        #ifdef SHOW_JSON
        printf("%s received: %s\n",qUtf8Printable(className()),qUtf8Printable(message));
        #endif
 
        startPing();                                                            //restart the _ping timer
 
        QJsonDocument d = QJsonDocument::fromJson(message);
        QJsonObject o = d.object();
 
        postID id = static_cast<postID>(o["id"].toInt());
 
        if (id == POST_ID_SESSION_DATA)
        {
                //printf("received session data request: %s\n",qUtf8Printable(d.toJson()));
 
                QString address = o["metadata"].toObject()["sessionid"].toString();
                VFS_session *s = VFS_session::fetchSession(address);
 
                if (s)
                {
                        //printf("found session!\n%s\n",qUtf8Printable(QJsonDocument(s->toJson()).toJson()));
 
                        QJsonObject m;
                        m["id"] = POST_ID_SESSION_DATA;
                        m["data"] = s->toJson();
                        m["metadata"] = o["metadata"];
 
                        sendMessage( QJsonDocument(m).toJson() );
                }
                else
                {
                        VFS::ERROR( QString("Session not found: '%1'.  Cannot resolve session request.").arg(address) );
 
                        QJsonObject m;
                        m["id"] = POST_ID_SESSION_DATA;
                        m["data"] = QJsonValue::Null;
                        m["metadata"] = o["metadata"];
 
                        //sendMessage( QJsonDocument(m).toJson() );
                }
        }
        else if (id == POST_ID_DIFF)
        {
                //printf("MOUNT RECEIVED DIFF\n");
                VFS_request r;
                r.fromJsonObject(o,true);
 
                // if the origin of this diff was a submit,
                // this field will contain the ID of that submit request,
                // which will refer to the id in _requests.
                // This will allow us to copy _notifyExceptions from the submit request on this new request
                if (o.contains("sourceID"))
                {
                        postID sourceID = static_cast<postID>(o["sourceID"].toInt());
                        //printf("sourceID: %d\n",sourceID);
 
                        if (_requests.contains(sourceID))
                        {
                                VFS_request *sr = _requests[sourceID];
                                r._notifyExceptions = sr->_notifyExceptions;
                                r._initialPath = sr->_initialPath;
 
                        //      printf("copied notifyExceptions\n");
                        //      for (int i=0;i<r._notifyExceptions.length();i++)
                        //              printf(" notify exception: %p %s\n",r._notifyExceptions[i].first,qUtf8Printable(r._notifyExceptions[i].second));
                        }
                        else
                                printf("requestID %d does not exist for notifyExceptions.\n",sourceID);
                }
 
                applyDiff(&r);
                //emit diff(this,&r);
 
                return;
        }
        else if (id == POST_ID_CODE_DIRECTORY)
        {
                VFS_request r;
                r.fromJsonObject(o,true);
 
                _codeDirectory.unregisterDirectory();
                _codeDirectory.registerDirectory(r._data.object());
 
                return;
        }
        else if (_requests.contains(id))
        {
                VFS_request *r = _requests.take(id);
                r->fromJsonObject(o);
 
                if (r->_requestType == VFS_request::codeDirectory)
                {
                        VFS::WARN("Deprecate code directory requests!");
                        _codeDirectory.registerDirectory((r->_data.object()));
                        delete r;
 
                        return;
                }
 
                bool willIssueMounted = false;
                QString issueMountedPath = "";
 
                // if this is a resubscription, we need to tell the client that the pane or resource has been mounted.
                if (r->_requestType == VFS_request::subscribe)
                {
                        if (r->_metadata.contains("resubscription") && r->_metadata["resubscription"].toBool())
                        {
                                //issueMounted(r->_initialPath);
                                willIssueMounted = true;
                                //issueMountedPath = r->_initialPath;
                                //issueMountedPath = r->_path;
                                issueMountedPath = r->_initialPath.mid(_path.length()+1);
 
                                //printf("RESUBSCRIBE: %s %p\n",qUtf8Printable(r->_initialPath),r->_origin);
                                //printf("RESUBSCRIBE: %p\n%s\n",r->_origin,qUtf8Printable(r->toJson(0,true,true)));
                        }
                }
 
                if (!r->_isCallback)
                        issueResponse(r);
 
                // need to submit this after the response, because otherwise it
                // will not have a _sourcePathMap entry yet.
                if (willIssueMounted)
                        issueMounted(issueMountedPath);
        }
        else
                VFS::WARN( QString("Received a message with an unknown id: %1.  Ignoring.").arg(id), 0, className() );
}
 
void VFS_tcp_mount::sendMessage( QByteArray message )
{
        #ifdef SHOW_JSON
        printf("%s sends: %s\n",qUtf8Printable(className()),qUtf8Printable(message));
        #endif
 
        #ifdef COMPRESS_TCP_MOUNT
        message = qCompress(message);
        #endif
 
        writeMessage(message);
}
 
void VFS_tcp_mount::unsubscribeAll(VFS_node *subscriber)
{
        //printf("UNSUBSCRIBE ALL\n");
 
        //send an unsubscribe request to the connected client for each matching subscription
        if (_mounted)
        {
                QMutexLocker l(&_lock);
 
                int c;
 
                QMutableMapIterator<QString, VFS_subscriptionOrigin > ito(_subscribers);
                while (ito.hasNext())
                {
                        ito.next();
 
                        QMutableMapIterator<QString, VFS_subscriptionCounter > it(ito.value());
                        while (it.hasNext())
                        {
                                it.next();
 
                                if (it.value().contains(subscriber))
                                        c=it.value()[subscriber];
                                else
                                        c = 0;
 
                                if (c)
                                {
                                        QJsonObject m;
                                        m["unsubscriptions"] = c;
                                        VFS_request *r = createRequest(VFS_request::unsubscribe,ito.key(),className(),QJsonDocument(),m);
                                        executeRequest(r);
                                        VFS::LOG( QString("unsubscribeAll removed '%1' %2 (%3 remote)").arg(ito.key()).arg(className()).arg(c), 9 );
                                        //printf( "%s [[%s]]\n", qUtf8Printable( QString("unsubscribeAll removed '%1' %2 (%3 remote)").arg(ito.key()).arg(className()).arg(c) ), qUtf8Printable(ito.key()) );
                                }
                        }
                }
        }
 
        VFS_tcp_client::unsubscribeAll(subscriber);
}
 
 
void VFS_tcp_mount::refreshSubscriptions()
{
        //return;
 
        QMutexLocker l(&_lock);
 
        if (_subscribers.count())
                VFS::LOG( QString("Refreshing subscriptions..."), 8, className() );
        else
                return;
 
        VFS_subscriptionType::iterator it = _subscribers.begin();
        VFS_subscriptionOrigin::iterator ito;
 
        QString path;
        VFS_subscriptionOrigin o;
        VFS_subscriptionCounter c;
 
        while (it!=_subscribers.end())
        {
                path = it.key();
                o = it.value();
 
                ito = o.begin();
                while (ito!=o.end())
                {
                        c = ito.value();
 
                        foreach(VFS_node *n,c.keys())
                        {
                                if (__isNode(n))
                                {
                                        int count = c[n];
 
                                        VFS_request *r = n->createRequest(VFS_request::subscribe,path,className());
                                        r->_origin = n;
                                        r->_originPath = ito.key();
                                        r->_initialPath = rutils::cleanPath(_path+"/"+path);
                                        r->_prefixPath = _path.split("/");
                                        r->_metadata["subscriptions"] = count;
                                        r->_metadata["resubscription"] = true;
 
                                        executeRequest(r);
                                        //issueRequest(r);
 
                                        VFS::LOG( QString("(re)subscribing to '%1' (%2)\n").arg(path).arg(count), 8, className() );
                                }
                                else
                                        VFS::WARN( "Skipping resubscribe on non-existent node\n" );
                        }
 
                        ito++;
                }
 
                it++;
        }
}
 
 
void VFS_tcp_mount::sendUnmountedRequests()
{
        //return;
 
        VFS_request *r;
 
        if (!_unmountedRequests.isEmpty())
        {
                VFS::LOG( QString("Sending unmounted requests..."), 9, className() );
                while (_unmountedRequests.length())
                {
                        r = _unmountedRequests.takeFirst();
 
                        if (__isNode(r->_origin))
                        {       //r->_dontDelete = false;
                                issueRequest(this,r);
                        }
                        else
                        {       VFS::WARN(QString("Skipping request for non-node 0x%1").arg((quintptr)r->_origin, QT_POINTER_SIZE * 2, 16, QChar('0')),9,className());
                                delete r;
                        }
                }
        }
}
 
 
void VFS_tcp_mount::issueUnmounted()
{
        //printf("ISSUED UNMOUNTED\n");
 
        QMutexLocker l(&_lock);
 
        QJsonObject d;
        d["mounted"] = false;
 
        VFS_subscriptionType::iterator it = _subscribers.begin();
        VFS_subscriptionOrigin::iterator ito;
 
        QString path;
        VFS_subscriptionCounter c;
 
        while (it!=_subscribers.end())
        {
                path = it.key();
 
                VFS_request *r = createRequest(VFS_request::diff,path,className(),QJsonDocument(d));
                emit diff(this,r);
                delete r;
 
                it++;
        }
}
 
 
 void VFS_tcp_mount::issueMounted(QString path)
 {
        //printf("ISSUED MOUNTED: %s\n",qUtf8Printable(path));
 
        QJsonObject d;
        d["mounted"] = true;
 
        VFS_request *r = createRequest(VFS_request::diff,path,className(),QJsonDocument(d));
        emit diff(this,r);
        delete r;
 }
 
 
postID VFS_tcp_mount::getNewRequestID()
{
        postID s = _currentID;
 
        //FIXME : this increment is only for debugging... it should not be here... it may be better to use UUIDs in general, as they may debug more easily
        //_currentID++;
 
        while (_requests.contains(_currentID))
        {
                //_currentID = rand();  //for debugging
                //continue;             //for debugging
 
                _currentID++;
                if (_currentID > POST_ID_MAX)
                        _currentID = POST_ID_START;
 
                if (_currentID == s)
                {
                        VFS::ERROR( QString("Could not find available id for request.  Aborting."), 0, className() );
                        exit(3);
                }
        }
 
        return _currentID;
}
 
 
VFS_node *VFS_tcp_mount::find(VFS_request *r)
{
        Q_UNUSED(r)
 
        return this;
}
 
 
QByteArray VFS_tcp_mount::icon()
{
        return VFS_icons::get("unknown");
}
 
 
bool VFS_tcp_mount::isMounted()
{
        QMutexLocker l(&_lock);
 
        return _mounted;
}
 
 
QString VFS_tcp_mount::path()
{
        QMutexLocker l(&_lock);
 
        return _path;
}