X-Git-Url: https://git.quassel-irc.org/?a=blobdiff_plain;f=src%2Fcommon%2Fsignalproxy.cpp;h=76fccb73eb7013865015f9a5b0d5d3b6e2f2468a;hb=9d02d714ff6ddfdd4a81fbe33d829c845d1fd4b8;hp=9078f24a36c602aeda94141319c05162040ef938;hpb=57d44eafac4d353d1523e93cc7613debc1826e94;p=quassel.git diff --git a/src/common/signalproxy.cpp b/src/common/signalproxy.cpp index 9078f24a..76fccb73 100644 --- a/src/common/signalproxy.cpp +++ b/src/common/signalproxy.cpp @@ -179,7 +179,7 @@ void SignalRelay::attachSignal(int methodId, const QByteArray &func) { sigNames.insert(methodId, fn); } // ==================== -// END SIGNALRELAY +// /SIGNALRELAY // ==================== @@ -211,14 +211,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 +251,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,51 +270,53 @@ 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(); } void SignalProxy::removePeerBySender() { // OK we're brutal here... but since it's a private slot we know what we've got connected to it... + // this Slot is not triggered by destroyed, so the object is still alive and can be used! QIODevice *ioDev = (QIODevice *)(sender()); removePeer(ioDev); - qDebug() << "Client disconnected."; } void SignalProxy::objectRenamed(const QString &newname, const QString &oldname) { @@ -604,23 +606,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 +678,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 +690,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 +902,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"); 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); @@ -924,8 +951,21 @@ bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVarian 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 +1016,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)); }