X-Git-Url: https://git.quassel-irc.org/?p=quassel.git;a=blobdiff_plain;f=src%2Fcommon%2Fsignalproxy.cpp;h=20392081ba27ad066a603480b5c944f6e459767d;hp=22e71f135323e12e4e8d92d0d5e2a022f708d32f;hb=dbd52200c4ce9bdf30f5aea707b54a28ea900003;hpb=444e91f948b435e652205c4d0f1148906e9e86dc diff --git a/src/common/signalproxy.cpp b/src/common/signalproxy.cpp index 22e71f13..20392081 100644 --- a/src/common/signalproxy.cpp +++ b/src/common/signalproxy.cpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -179,7 +180,7 @@ void SignalRelay::attachSignal(int methodId, const QByteArray &func) { sigNames.insert(methodId, fn); } // ==================== -// END SIGNALRELAY +// /SIGNALRELAY // ==================== @@ -211,14 +212,14 @@ SignalProxy::~SignalProxy() { detachObject(sender); // close peer connections - foreach(QIODevice *device, _peerByteCount.keys()) { + foreach(QIODevice *device, _peers.keys()) { device->close(); delete device; } } void SignalProxy::setProxyMode(ProxyMode mode) { - foreach(QIODevice* peer, _peerByteCount.keys()) { + foreach(QIODevice* peer, _peers.keys()) { if(peer->isOpen()) { qWarning() << "SignalProxy: Cannot change proxy mode while connected"; return; @@ -251,10 +252,10 @@ bool SignalProxy::addPeer(QIODevice* iodev) { if(!iodev) return false; - if(_peerByteCount.contains(iodev)) + if(_peers.contains(iodev)) return true; - if(proxyMode() == Client && !_peerByteCount.isEmpty()) { + if(proxyMode() == Client && !_peers.isEmpty()) { qWarning("SignalProxy: only one peer allowed in client mode!"); return false; } @@ -270,43 +271,45 @@ bool SignalProxy::addPeer(QIODevice* iodev) { connect(sock, SIGNAL(disconnected()), this, SLOT(removePeerBySender())); } - _peerByteCount[iodev] = 0; + _peers[iodev] = peerInfo(); + if(iodev->property("UseCompression").toBool()) + _peers[iodev].usesCompression = true; - if(_peerByteCount.count() == 1) + if(_peers.count() == 1) emit connected(); return true; } void SignalProxy::removePeer(QIODevice* iodev) { - if(_peerByteCount.isEmpty()) { + if(_peers.isEmpty()) { qWarning() << "SignalProxy::removePeer(): No peers in use!"; return; } if(proxyMode() == Server && !iodev) { // disconnect all - QList peers = _peerByteCount.keys(); + QList peers = _peers.keys(); foreach(QIODevice *peer, peers) removePeer(peer); } if(proxyMode() != Server && !iodev) - iodev = _peerByteCount.keys().first(); + iodev = _peers.keys().first(); Q_ASSERT(iodev); - if(!_peerByteCount.contains(iodev)) { + if(!_peers.contains(iodev)) { qWarning() << "SignalProxy: unknown QIODevice" << iodev; return; } - _peerByteCount.remove(iodev); + _peers.remove(iodev); disconnect(iodev, 0, this, 0); emit peerRemoved(iodev); - if(_peerByteCount.isEmpty()) + if(_peers.isEmpty()) emit disconnected(); } @@ -604,23 +607,27 @@ void SignalProxy::synchronize(SyncableObject *obj) { if(proxyMode() == Server) { connect(obj, SIGNAL(objectRenamed(QString, QString)), this, SLOT(objectRenamed(QString, QString))); - setInitialized(obj); + obj->setInitialized(); + emit objectInitialized(obj); } else { - requestInit(obj); + if(obj->isInitialized()) + emit objectInitialized(obj); + else + requestInit(obj); } } -void SignalProxy::setInitialized(SyncableObject *obj) { - obj->setInitialized(); - emit objectInitialized(obj); -} +// void SignalProxy::setInitialized(SyncableObject *obj) { +// obj->setInitialized(); +// emit objectInitialized(obj); +// } -bool SignalProxy::isInitialized(SyncableObject *obj) const { - return obj->isInitialized(); -} +// bool SignalProxy::isInitialized(SyncableObject *obj) const { +// return obj->isInitialized(); +// } void SignalProxy::requestInit(SyncableObject *obj) { - if(proxyMode() == Server || isInitialized(obj)) + if(proxyMode() == Server || obj->isInitialized()) return; QVariantList params; @@ -672,10 +679,11 @@ void SignalProxy::stopSync(SyncableObject* obj) { } void SignalProxy::dispatchSignal(QIODevice *receiver, const RequestType &requestType, const QVariantList ¶ms) { + Q_ASSERT(_peers.contains(receiver)); QVariantList packedFunc; packedFunc << (qint16)requestType; packedFunc << params; - writeDataToDevice(receiver, QVariant(packedFunc)); + writeDataToDevice(receiver, QVariant(packedFunc), _peers[receiver].usesCompression); } void SignalProxy::dispatchSignal(const RequestType &requestType, const QVariantList ¶ms) { @@ -683,8 +691,10 @@ void SignalProxy::dispatchSignal(const RequestType &requestType, const QVariantL QVariantList packedFunc; packedFunc << (qint16)requestType; packedFunc << params; - foreach(QIODevice* dev, _peerByteCount.keys()) - writeDataToDevice(dev, QVariant(packedFunc)); + foreach(QIODevice* dev, _peers.keys()) { + Q_ASSERT(_peers.contains(dev)); + writeDataToDevice(dev, QVariant(packedFunc), _peers[dev].usesCompression); + } } void SignalProxy::receivePeerSignal(QIODevice *sender, const QVariant &packedFunc) { @@ -893,27 +903,45 @@ bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList void SignalProxy::dataAvailable() { // yet again. it's a private slot. no need for checks. QIODevice* ioDev = qobject_cast(sender()); + Q_ASSERT(_peers.contains(ioDev)); QVariant var; - while(readDataFromDevice(ioDev, _peerByteCount[ioDev], var)) + while(readDataFromDevice(ioDev, _peers[ioDev].byteCount, var, _peers[ioDev].usesCompression)) receivePeerSignal(ioDev, var); } -void SignalProxy::writeDataToDevice(QIODevice *dev, const QVariant &item) { +void SignalProxy::writeDataToDevice(QIODevice *dev, const QVariant &item, bool compressed) { QAbstractSocket* sock = qobject_cast(dev); if(!dev->isOpen() || (sock && sock->state()!=QAbstractSocket::ConnectedState)) { - qWarning("SignalProxy: Can't call on a closed device"); + qWarning("SignalProxy: Can't call write on a closed device"); return; } + QByteArray block; QDataStream out(&block, QIODevice::WriteOnly); out.setVersion(QDataStream::Qt_4_2); - out << (quint32)0 << item; + out << (quint32)0; + + if(compressed) { + QByteArray rawItem; + QDataStream itemStream(&rawItem, QIODevice::WriteOnly); + + itemStream.setVersion(QDataStream::Qt_4_2); + itemStream << item; + + rawItem = qCompress(rawItem); + + out << rawItem; + } else { + out << item; + } + out.device()->seek(0); out << (quint32)(block.size() - sizeof(quint32)); + dev->write(block); } -bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVariant &item) { +bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVariant &item, bool compressed) { QDataStream in(dev); in.setVersion(QDataStream::Qt_4_2); @@ -922,10 +950,31 @@ bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVarian in >> blockSize; } + if(blockSize > 1 << 22) { + qWarning() << qPrintable(tr("Client tried to send package larger than max package size!")); + QAbstractSocket* sock = qobject_cast(dev); + qWarning() << qPrintable(tr("Disconnecting")) << (sock ? qPrintable(sock->peerAddress().toString()) : qPrintable(tr("local client"))); + dev->close(); + return false; + } + if(dev->bytesAvailable() < blockSize) return false; - in >> item; + + if(compressed) { + QByteArray rawItem; + in >> rawItem; + rawItem = qUncompress(rawItem); + + QDataStream itemStream(&rawItem, QIODevice::ReadOnly); + itemStream.setVersion(QDataStream::Qt_4_2); + itemStream >> item; + } else { + in >> item; + } + blockSize = 0; + return true; } @@ -976,10 +1025,11 @@ QVariantMap SignalProxy::initData(SyncableObject *obj) const { } void SignalProxy::setInitData(SyncableObject *obj, const QVariantMap &properties) { - if(isInitialized(obj)) + if(obj->isInitialized()) return; obj->fromVariantMap(properties); - setInitialized(obj); + obj->setInitialized(); + emit objectInitialized(obj); invokeSlot(obj, updatedRemotelyId(obj)); }