X-Git-Url: https://git.quassel-irc.org/?p=quassel.git;a=blobdiff_plain;f=src%2Fcommon%2Fsignalproxy.cpp;h=7808ad1087ca13fff4d98d6b2725fd49667c7d2c;hp=b3ea4ffbc9e0b30b7180c23708ac283a32e0fd35;hb=aa7841638c6537034102da959cad8c36853bf280;hpb=efe20e20080a0c22e1a7b8b84ef622130dbf116e diff --git a/src/common/signalproxy.cpp b/src/common/signalproxy.cpp index b3ea4ffb..7808ad10 100644 --- a/src/common/signalproxy.cpp +++ b/src/common/signalproxy.cpp @@ -45,12 +45,18 @@ // ================================================== class PeerSignalEvent : public QEvent { public: - PeerSignalEvent(SignalProxy *sender, SignalProxy::RequestType requestType, const QVariantList ¶ms) : QEvent(QEvent::User), sender(sender), requestType(requestType), params(params) {} + PeerSignalEvent(SignalProxy *sender, SignalProxy::RequestType requestType, const QVariantList ¶ms) : QEvent(QEvent::Type(SignalProxy::PeerSignal)), sender(sender), requestType(requestType), params(params) {} SignalProxy *sender; SignalProxy::RequestType requestType; QVariantList params; }; +class RemovePeerEvent : public QEvent { +public: + RemovePeerEvent(QObject *peer) : QEvent(QEvent::Type(SignalProxy::RemovePeer)), peer(peer) {} + QObject *peer; +}; + // ================================================== // SIGNALRELAY // ================================================== @@ -67,12 +73,12 @@ public: void setSynchronize(bool); bool synchronize() const; - + int sigCount() const; - + private: bool isSyncMethod(int i); - + SignalProxy* proxy; QObject* caller; QMultiHash sigNames; @@ -110,7 +116,7 @@ int SignalRelay::qt_metacall(QMetaObject::Call _c, int _id, void **_a) { proxy->dispatchSignal(SignalProxy::RpcCall, QVariantList() << funcIter.value() << params); funcIter++; } - + // dispatch Sync Signal if necessary QByteArray signature(caller->metaObject()->method(_id).signature()); SyncableObject *syncObject = qobject_cast(caller); @@ -162,7 +168,7 @@ bool SignalRelay::isSyncMethod(int i) { QByteArray signature = syncObject->syncMetaObject()->method(i).signature(); if(!proxy->syncMap(syncObject).contains(signature)) return false; - + if(proxy->proxyMode() == SignalProxy::Server && !signature.contains("Requested")) return true; @@ -206,6 +212,14 @@ void SignalProxy::IODevicePeer::dispatchSignal(const RequestType &requestType, c dispatchPackedFunc(QVariant(packedFunc)); } +QString SignalProxy::IODevicePeer::address() const { + QAbstractSocket *socket = qobject_cast(_device); + if(socket) + return socket->peerAddress().toString(); + else + return QString(); +} + void SignalProxy::SignalProxyPeer::dispatchSignal(const RequestType &requestType, const QVariantList ¶ms) { Qt::ConnectionType type = QThread::currentThread() == receiver->thread() ? Qt::DirectConnection @@ -241,7 +255,7 @@ SignalProxy::SignalProxy(ProxyMode mode, QIODevice* device, QObject* parent) setProxyMode(mode); addPeer(device); init(); -} +} SignalProxy::~SignalProxy() { QList senders = _relayHash.keys(); @@ -289,7 +303,7 @@ void SignalProxy::initClient() { bool SignalProxy::addPeer(QIODevice* iodev) { if(!iodev) return false; - + if(_peers.contains(iodev)) return true; @@ -298,16 +312,16 @@ bool SignalProxy::addPeer(QIODevice* iodev) { return false; } - if(!iodev->isOpen()) - qWarning("SignalProxy::the device you passed is not open!"); + if(!iodev->isOpen()) { + qWarning("SignalProxy::addPeer(QIODevice *iodev): iodev needs to be open!"); + return false; + } connect(iodev, SIGNAL(disconnected()), this, SLOT(removePeerBySender())); connect(iodev, SIGNAL(readyRead()), this, SLOT(dataAvailable())); - QAbstractSocket* sock = qobject_cast(iodev); - if(sock) { - connect(sock, SIGNAL(disconnected()), this, SLOT(removePeerBySender())); - } + if(!iodev->parent()) + iodev->setParent(this); _peers[iodev] = new IODevicePeer(iodev, iodev->property("UseCompression").toBool()); @@ -336,7 +350,6 @@ bool SignalProxy::addPeer(SignalProxy* proxy) { } _peers[proxy] = new SignalProxyPeer(this, proxy); - proxy->addPeer(this); if(_peers.count() == 1) @@ -373,6 +386,15 @@ void SignalProxy::removePeer(QObject* dev) { if(peer->type() == AbstractPeer::IODevicePeer) emit peerRemoved(static_cast(dev)); + if(peer->type() == AbstractPeer::SignalProxyPeer) { + SignalProxy *proxy = static_cast(dev); + if(proxy->_peers.contains(this)) + proxy->removePeer(this); + } + + if(dev->parent() == this) + dev->deleteLater(); + delete peer; if(_peers.isEmpty()) @@ -391,7 +413,7 @@ void SignalProxy::objectRenamed(const QString &newname, const QString &oldname) if(proxyMode() == Client) return; - + QVariantList params; params << "__objectRenamed__" << className << newname << oldname; dispatchSignal(RpcCall, params); @@ -475,7 +497,7 @@ const QByteArray &SignalProxy::methodName(QObject *obj, int methodId) { void SignalProxy::setSyncMap(SyncableObject *obj) { const QMetaObject *meta = obj->syncMetaObject(); QHash syncMap; - + QList slotIndexes; for(int i = 0; i < meta->methodCount(); i++) { if(meta->method(i).methodType() == QMetaMethod::Slot) @@ -550,7 +572,7 @@ void SignalProxy::setReceiveMap(SyncableObject *obj) { signature = QByteArray(requestSlot.signature()); if(!signature.startsWith("request")) continue; - + paramsPos = signature.indexOf('('); if(paramsPos == -1) continue; @@ -572,7 +594,7 @@ void SignalProxy::setReceiveMap(SyncableObject *obj) { if(receiverId != -1) receiveMap[i] = receiverId; } - _classInfo[meta]->receiveMap = receiveMap; + _classInfo[meta]->receiveMap = receiveMap; } const QHash &SignalProxy::receiveMap(SyncableObject *obj) { @@ -654,7 +676,7 @@ bool SignalProxy::attachSlot(const QByteArray& sigName, QObject* recv, const cha void SignalProxy::synchronize(SyncableObject *obj) { createClassInfo(obj); setUpdatedRemotelyId(obj); - + // attaching all the Signals SignalRelay* relay; if(_relayHash.contains(obj)) @@ -740,7 +762,10 @@ void SignalProxy::dispatchSignal(const RequestType &requestType, const QVariantL case AbstractPeer::IODevicePeer: { IODevicePeer *ioPeer = static_cast(*peer); - ioPeer->dispatchPackedFunc(packedFunc); + if(ioPeer->isOpen()) + ioPeer->dispatchPackedFunc(packedFunc); + else + QCoreApplication::postEvent(this, new RemovePeerEvent(peer.key())); } break; case AbstractPeer::SignalProxyPeer: @@ -764,7 +789,7 @@ void SignalProxy::receivePackedFunc(AbstractPeer *sender, const QVariant &packed RequestType requestType = (RequestType)params.takeFirst().value(); receivePeerSignal(sender, requestType, params); } - + void SignalProxy::receivePeerSignal(AbstractPeer *sender, const RequestType &requestType, const QVariantList ¶ms) { switch(requestType) { case RpcCall: @@ -778,15 +803,15 @@ void SignalProxy::receivePeerSignal(AbstractPeer *sender, const RequestType &req case Sync: handleSync(sender, params); break; - + case InitRequest: handleInitRequest(sender, params); break; - + case InitData: handleInitData(sender, params); break; - + case HeartBeat: receiveHeartBeat(sender, params); break; @@ -814,7 +839,7 @@ void SignalProxy::handleSync(AbstractPeer *sender, QVariantList params) { qWarning() << "received invalid Sync call" << params; return; } - + QByteArray className = params.takeFirst().toByteArray(); QString objectName = params.takeFirst().toString(); QByteArray signal = params.takeFirst().toByteArray(); @@ -851,7 +876,7 @@ void SignalProxy::handleSync(AbstractPeer *sender, QVariantList params) { returnParams << returnValue; sender->dispatchSignal(Sync, returnParams); } - + // send emit update signal invokeSlot(receiver, updatedRemotelyId(receiver)); } @@ -862,10 +887,10 @@ void SignalProxy::handleInitRequest(AbstractPeer *sender, const QVariantList &pa << params; return; } - + QByteArray className(params[0].toByteArray()); QString objectName(params[1].toString()); - + if(!_syncSlave.contains(className)) { qWarning() << "SignalProxy::handleInitRequest() received initRequest for unregistered Class:" << className; @@ -877,7 +902,7 @@ void SignalProxy::handleInitRequest(AbstractPeer *sender, const QVariantList &pa << className << objectName; return; } - + SyncableObject *obj = _syncSlave[className][objectName]; QVariantList params_; @@ -895,7 +920,7 @@ void SignalProxy::handleInitData(AbstractPeer *sender, const QVariantList ¶m << params; return; } - + QByteArray className(params[0].toByteArray()); QString objectName(params[1].toString()); QVariantMap propertyMap(params[2].toMap()); @@ -944,7 +969,7 @@ bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList return false; } - void *_a[] = {0, // return type... + void *_a[] = {0, // return type... 0, 0, 0, 0 , 0, // and 10 args - that's the max size qt can handle with signals and slots 0, 0, 0, 0 , 0}; @@ -964,7 +989,7 @@ bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList if(returnValue.type() != QVariant::Invalid) _a[0] = const_cast(returnValue.constData()); - + Qt::ConnectionType type = QThread::currentThread() == receiver->thread() ? Qt::DirectConnection : Qt::QueuedConnection; @@ -976,7 +1001,7 @@ bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList // note to self: qmetaobject.cpp:990 ff return false; } - + } bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList ¶ms) { @@ -1005,7 +1030,7 @@ void SignalProxy::writeDataToDevice(QIODevice *dev, const QVariant &item, bool c QDataStream out(&block, QIODevice::WriteOnly); out.setVersion(QDataStream::Qt_4_2); out << (quint32)0; - + if(compressed) { QByteArray rawItem; QDataStream itemStream(&rawItem, QIODevice::WriteOnly); @@ -1036,30 +1061,35 @@ bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVarian } 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(); + disconnectDevice(dev, tr("Client tried to send package larger than max package size!")); return false; } - + + if(blockSize == 0) { + disconnectDevice(dev, tr("Client tried to send 0 byte package!")); + return false; + } + if(dev->bytesAvailable() < blockSize) return false; + blockSize = 0; + if(compressed) { QByteArray rawItem; in >> rawItem; - // debug check + int nbytes = rawItem.size(); - if (nbytes <= 4) { + if(nbytes <= 4) { const char *data = rawItem.constData(); - if (nbytes < 4 || (data[0]!=0 || data[1]!=0 || data[2]!=0 || data[3]!=0)) - qWarning() << "receieved corrupted compressed data:" - << blockSize << rawItem << rawItem.size() << dev; + if(nbytes < 4 || (data[0]!=0 || data[1]!=0 || data[2]!=0 || data[3]!=0)) { + disconnectDevice(dev, tr("Client sent corrupted compressed data!")); + return false; + } } - // end + rawItem = qUncompress(rawItem); - + QDataStream itemStream(&rawItem, QIODevice::ReadOnly); itemStream.setVersion(QDataStream::Qt_4_2); itemStream >> item; @@ -1067,7 +1097,10 @@ bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVarian in >> item; } - blockSize = 0; + if(!item.isValid()) { + disconnectDevice(dev, tr("Client sent corrupt data: unable to load QVariant!")); + return false; + } return true; } @@ -1128,21 +1161,20 @@ void SignalProxy::setInitData(SyncableObject *obj, const QVariantMap &properties } void SignalProxy::sendHeartBeat() { - dispatchSignal(SignalProxy::HeartBeat, QVariantList() << QTime::currentTime()); + QVariantList heartBeatParams; + heartBeatParams << QTime::currentTime(); PeerHash::iterator peer = _peers.begin(); while(peer != _peers.end()) { if((*peer)->type() == AbstractPeer::IODevicePeer) { IODevicePeer *ioPeer = static_cast(*peer); + ioPeer->dispatchSignal(SignalProxy::HeartBeat, heartBeatParams); if(ioPeer->sentHeartBeats > 0) { updateLag(ioPeer, ioPeer->sentHeartBeats * _heartBeatTimer.interval()); } if(ioPeer->sentHeartBeats > 1) { - //FIXME: proper disconnect. -// QAbstractSocket *socket = qobject_cast(peerIter.key()); -// qWarning() << "SignalProxy: Disconnecting peer:" -// << (socket ? qPrintable(socket->peerAddress().toString()) : "local client") -// << "(didn't receive a heartbeat for over" << peerIter->sentHeartBeats * _heartBeatTimer.interval() / 1000 << "seconds)"; -// peerIter.key()->close(); + qWarning() << "SignalProxy: Disconnecting peer:" << ioPeer->address() + << "(didn't receive a heartbeat for over" << ioPeer->sentHeartBeats * _heartBeatTimer.interval() / 1000 << "seconds)"; + ioPeer->close(); } else { ioPeer->sentHeartBeats++; } @@ -1165,28 +1197,43 @@ void SignalProxy::receiveHeartBeatReply(AbstractPeer *peer, const QVariantList & ioPeer->sentHeartBeats = 0; if(params.isEmpty()) { - qWarning() << "SignalProxy: received heart beat reply with less params then sent from:" << ioPeer->device(); + qWarning() << "SignalProxy: received heart beat reply with less params then sent from:" << ioPeer->address(); return; } - + QTime sendTime = params[0].value(); updateLag(ioPeer, sendTime.msecsTo(QTime::currentTime()) / 2); } void SignalProxy::customEvent(QEvent *event) { switch(event->type()) { - case QEvent::User: + case PeerSignal: { - PeerSignalEvent *sig = static_cast(event); - receivePeerSignal(sig->sender, sig->requestType, sig->params); + PeerSignalEvent *e = static_cast(event); + receivePeerSignal(e->sender, e->requestType, e->params); } event->accept(); break; + case RemovePeer: + { + RemovePeerEvent *e = static_cast(event); + removePeer(e->peer); + } + event->accept(); default: return; } } +void SignalProxy::disconnectDevice(QIODevice *dev, const QString &reason) { + if(!reason.isEmpty()) + qWarning() << qPrintable(reason); + QAbstractSocket *sock = qobject_cast(dev); + if(sock) + qWarning() << qPrintable(tr("Disconnecting")) << qPrintable(sock->peerAddress().toString()); + dev->close(); +} + void SignalProxy::updateLag(IODevicePeer *peer, int lag) { peer->lag = lag; if(proxyMode() == Client) { @@ -1209,7 +1256,7 @@ void SignalProxy::dumpProxyStats() { int slaveCount = 0; foreach(ObjectId oid, _syncSlave.values()) slaveCount += oid.count(); - + qDebug() << this; qDebug() << " Proxy Mode:" << mode; qDebug() << "attached sending Objects:" << _relayHash.count(); @@ -1226,13 +1273,13 @@ void SignalProxy::dumpSyncMap(SyncableObject *object) { QHash syncMap_ = syncMap(object); QHash::const_iterator iter = syncMap_.constBegin(); while(iter != syncMap_.constEnd()) { - qDebug() << iter.key() << "-->" << iter.value() << meta->method(iter.value()).signature(); + qDebug() << iter.key() << "-->" << iter.value() << meta->method(iter.value()).signature(); iter++; } // QHash syncMap_ = syncMap(object); // QHash::const_iterator iter = syncMap_.constBegin(); // while(iter != syncMap_.constEnd()) { -// qDebug() << iter.key() << meta->method(iter.key()).signature() << "-->" << iter.value() << meta->method(iter.value()).signature(); +// qDebug() << iter.key() << meta->method(iter.key()).signature() << "-->" << iter.value() << meta->method(iter.value()).signature(); // iter++; // } }