X-Git-Url: https://git.quassel-irc.org/?p=quassel.git;a=blobdiff_plain;f=src%2Fcommon%2Fsignalproxy.cpp;h=94aa063f55e6b909298ac5ceff66f92cdbd96bc3;hp=467041eee6c97f7b07291c55e61e07ffcd385562;hb=cc0cad583046c34cb1296329016e1834995d171d;hpb=6fec875b14593ce6c83a9095fa5334afac6b6777 diff --git a/src/common/signalproxy.cpp b/src/common/signalproxy.cpp index 467041ee..94aa063f 100644 --- a/src/common/signalproxy.cpp +++ b/src/common/signalproxy.cpp @@ -34,12 +34,27 @@ #include #include #include +#include +#include #include "syncableobject.h" #include "util.h" -class SignalRelay: public QObject { +// ================================================== +// PeerSignalEvent +// ================================================== +class PeerSignalEvent : public QEvent { +public: + PeerSignalEvent(SignalProxy *sender, SignalProxy::RequestType requestType, const QVariantList ¶ms) : QEvent(QEvent::User), sender(sender), requestType(requestType), params(params) {} + SignalProxy *sender; + SignalProxy::RequestType requestType; + QVariantList params; +}; +// ================================================== +// SIGNALRELAY +// ================================================== +class SignalRelay: public QObject { /* Q_OBJECT is not necessary or even allowed, because we implement qt_metacall ourselves (and don't use any other features of the meta object system) @@ -180,14 +195,40 @@ void SignalRelay::attachSignal(int methodId, const QByteArray &func) { } sigNames.insert(methodId, fn); } -// ==================== -// /SIGNALRELAY -// ==================== +// ================================================== +// Peers +// ================================================== +void SignalProxy::IODevicePeer::dispatchSignal(const RequestType &requestType, const QVariantList ¶ms) { + QVariantList packedFunc; + packedFunc << (qint16)requestType + << params; + 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 + : Qt::QueuedConnection; + + if(type == Qt::DirectConnection) { + receiver->receivePeerSignal(sender, requestType, params); + } else { + QCoreApplication::postEvent(receiver, new PeerSignalEvent(sender, requestType, params)); + } +} + +// ================================================== // SignalProxy -// ==================== +// ================================================== SignalProxy::SignalProxy(QObject* parent) : QObject(parent) { @@ -214,21 +255,26 @@ SignalProxy::~SignalProxy() { QList senders = _relayHash.keys(); foreach(QObject* sender, senders) detachObject(sender); - - // close peer connections - foreach(QIODevice *device, _peers.keys()) { - device->close(); - delete device; - } + removeAllPeers(); } void SignalProxy::setProxyMode(ProxyMode mode) { - foreach(QIODevice* peer, _peers.keys()) { - if(peer->isOpen()) { - qWarning() << "SignalProxy: Cannot change proxy mode while connected"; + PeerHash::iterator peer = _peers.begin(); + while(peer != _peers.end()) { + if((*peer)->type() != AbstractPeer::IODevicePeer) { + IODevicePeer *ioPeer = static_cast(*peer); + if(ioPeer->isOpen()) { + qWarning() << "SignalProxy: Cannot change proxy mode while connected"; + return; + } + } + if((*peer)->type() != AbstractPeer::SignalProxyPeer) { + qWarning() << "SignalProxy: Cannot change proxy mode while connected to another internal SignalProxy"; return; } + peer++; } + _proxyMode = mode; if(mode == Server) initServer(); @@ -271,9 +317,7 @@ bool SignalProxy::addPeer(QIODevice* iodev) { connect(sock, SIGNAL(disconnected()), this, SLOT(removePeerBySender())); } - _peers[iodev] = peerInfo(); - if(iodev->property("UseCompression").toBool()) - _peers[iodev].usesCompression = true; + _peers[iodev] = new IODevicePeer(iodev, iodev->property("UseCompression").toBool()); if(_peers.count() == 1) emit connected(); @@ -281,43 +325,70 @@ bool SignalProxy::addPeer(QIODevice* iodev) { return true; } -void SignalProxy::removePeer(QIODevice* iodev) { - if(_peers.isEmpty()) { - qWarning() << "SignalProxy::removePeer(): No peers in use!"; - return; +bool SignalProxy::addPeer(SignalProxy* proxy) { + if(!proxy) + return false; + + if(proxyMode() == proxy->proxyMode()) { + qWarning() << "SignalProxy::addPeer(): adding a SignalProxy as peer requires one proxy to be server and one client!"; + return false; + } + + if(_peers.contains(proxy)) { + return true; } - if(proxyMode() == Server && !iodev) { - // disconnect all - QList peers = _peers.keys(); - foreach(QIODevice *peer, peers) - removePeer(peer); + if(proxyMode() == Client && !_peers.isEmpty()) { + qWarning("SignalProxy: only one peer allowed in client mode!"); + return false; } - if(proxyMode() != Server && !iodev) - iodev = _peers.keys().first(); + _peers[proxy] = new SignalProxyPeer(this, proxy); + + proxy->addPeer(this); - Q_ASSERT(iodev); + if(_peers.count() == 1) + emit connected(); + + return true; +} - if(!_peers.contains(iodev)) { - qWarning() << "SignalProxy: unknown QIODevice" << iodev; +void SignalProxy::removeAllPeers() { + Q_ASSERT(proxyMode() == Server || _peers.count() <= 1); + // wee need to copy that list since we modify it in the loop + QList peers = _peers.keys(); + foreach(QObject *peer, peers) { + removePeer(peer); + } +} + +void SignalProxy::removePeer(QObject* dev) { + if(_peers.isEmpty()) { + qWarning() << "SignalProxy::removePeer(): No peers in use!"; return; } - _peers.remove(iodev); + Q_ASSERT(dev); + if(!_peers.contains(dev)) { + qWarning() << "SignalProxy: unknown Peer" << dev; + return; + } + + AbstractPeer *peer = _peers[dev]; + _peers.remove(dev); + + disconnect(dev, 0, this, 0); + if(peer->type() == AbstractPeer::IODevicePeer) + emit peerRemoved(static_cast(dev)); - disconnect(iodev, 0, this, 0); - emit peerRemoved(iodev); + delete peer; 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); + removePeer(sender()); } void SignalProxy::objectRenamed(const QString &newname, const QString &oldname) { @@ -669,41 +740,47 @@ 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), _peers[receiver].usesCompression); -} - void SignalProxy::dispatchSignal(const RequestType &requestType, const QVariantList ¶ms) { - // yes I know we have a little code duplication here... it's for the sake of performance - QVariantList packedFunc; - packedFunc << (qint16)requestType; - packedFunc << params; - foreach(QIODevice* dev, _peers.keys()) { - Q_ASSERT(_peers.contains(dev)); - writeDataToDevice(dev, QVariant(packedFunc), _peers[dev].usesCompression); + QVariant packedFunc(QVariantList() << (qint16)requestType << params); + PeerHash::iterator peer = _peers.begin(); + while(peer != _peers.end()) { + switch((*peer)->type()) { + case AbstractPeer::IODevicePeer: + { + IODevicePeer *ioPeer = static_cast(*peer); + ioPeer->dispatchPackedFunc(packedFunc); + } + break; + case AbstractPeer::SignalProxyPeer: + (*peer)->dispatchSignal(requestType, params); + break; + default: + Q_ASSERT(false); // there shouldn't be any peers with wrong / unknown type + } + peer++; } } -void SignalProxy::receivePeerSignal(QIODevice *sender, const QVariant &packedFunc) { +void SignalProxy::receivePackedFunc(AbstractPeer *sender, const QVariant &packedFunc) { QVariantList params(packedFunc.toList()); if(params.isEmpty()) { qWarning() << "SignalProxy::receivePeerSignal(): received incompatible Data:" << packedFunc; return; } - - int callType = params.takeFirst().value(); - switch(callType) { + 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: if(params.empty()) qWarning() << "SignalProxy::receivePeerSignal(): received empty RPC-Call"; else - handleSignal(params.takeFirst().toByteArray(), params); + handleSignal(params); + //handleSignal(params.takeFirst().toByteArray(), params); break; case Sync: @@ -727,11 +804,20 @@ void SignalProxy::receivePeerSignal(QIODevice *sender, const QVariant &packedFun break; default: - qWarning() << "SignalProxy::receivePeerSignal(): received undefined CallType" << callType << params; + qWarning() << "SignalProxy::receivePeerSignal(): received undefined CallType" << requestType << params; + } +} + +void SignalProxy::receivePeerSignal(SignalProxy *sender, const RequestType &requestType, const QVariantList ¶ms) { + if(!_peers.contains(sender)) { + // we output only the pointer value. otherwise Qt would try to pretty print. As the object might already been destroyed, this is not a good idea. + qWarning() << "SignalProxy::receivePeerSignal(): received Signal from unknown Proxy" << reinterpret_cast(sender); + return; } + receivePeerSignal(_peers[sender], requestType, params); } -void SignalProxy::handleSync(QIODevice *sender, QVariantList params) { +void SignalProxy::handleSync(AbstractPeer *sender, QVariantList params) { if(params.count() < 3) { qWarning() << "received invalid Sync call" << params; return; @@ -771,14 +857,14 @@ void SignalProxy::handleSync(QIODevice *sender, QVariantList params) { if(argTypes(receiver, receiverId).count() > 1) returnParams << params; returnParams << returnValue; - dispatchSignal(sender, Sync, returnParams); + sender->dispatchSignal(Sync, returnParams); } // send emit update signal invokeSlot(receiver, updatedRemotelyId(receiver)); } -void SignalProxy::handleInitRequest(QIODevice *sender, const QVariantList ¶ms) { +void SignalProxy::handleInitRequest(AbstractPeer *sender, const QVariantList ¶ms) { if(params.count() != 2) { qWarning() << "SignalProxy::handleInitRequest() received initRequest with invalid param Count:" << params; @@ -807,10 +893,10 @@ void SignalProxy::handleInitRequest(QIODevice *sender, const QVariantList ¶m << objectName << initData(obj); - dispatchSignal(sender, InitData, params_); + sender->dispatchSignal(InitData, params_); } -void SignalProxy::handleInitData(QIODevice *sender, const QVariantList ¶ms) { +void SignalProxy::handleInitData(AbstractPeer *sender, const QVariantList ¶ms) { Q_UNUSED(sender) if(params.count() != 3) { qWarning() << "SignalProxy::handleInitData() received initData with invalid param Count:" @@ -838,7 +924,11 @@ void SignalProxy::handleInitData(QIODevice *sender, const QVariantList ¶ms) setInitData(obj, propertyMap); } -void SignalProxy::handleSignal(const QByteArray &funcName, const QVariantList ¶ms) { +//void SignalProxy::handleSignal(const QByteArray &funcName, const QVariantList ¶ms) { +void SignalProxy::handleSignal(const QVariantList &data) { + QVariantList params = data; + QByteArray funcName = params.takeFirst().toByteArray(); + QObject* receiver; int methodId; SlotHash::const_iterator slot = _attachedSlots.constFind(funcName); @@ -905,10 +995,11 @@ 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)); + Q_ASSERT(_peers.contains(ioDev) && _peers[ioDev]->type() == AbstractPeer::IODevicePeer); + IODevicePeer *peer = static_cast(_peers[ioDev]); QVariant var; - while(readDataFromDevice(ioDev, _peers[ioDev].byteCount, var, _peers[ioDev].usesCompression)) - receivePeerSignal(ioDev, var); + while(peer->readData(var)) + receivePackedFunc(peer, var); } void SignalProxy::writeDataToDevice(QIODevice *dev, const QVariant &item, bool compressed) { @@ -966,6 +1057,15 @@ bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVarian if(compressed) { QByteArray rawItem; in >> rawItem; + // debug check + int nbytes = rawItem.size(); + 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; + } + // end rawItem = qUncompress(rawItem); QDataStream itemStream(&rawItem, QIODevice::ReadOnly); @@ -1036,55 +1136,67 @@ void SignalProxy::setInitData(SyncableObject *obj, const QVariantMap &properties } void SignalProxy::sendHeartBeat() { - dispatchSignal(SignalProxy::HeartBeat, QVariantList() << QTime::currentTime()); - QHash::iterator peerIter = _peers.begin(); - QHash::iterator peerIterEnd = _peers.end(); - while(peerIter != peerIterEnd) { - if(peerIter->sentHeartBeats > 0) { - updateLag(peerIter.key(), (float)_heartBeatTimer.interval()); - } - if(peerIter->sentHeartBeats > 1) { - 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(); - } else { - peerIter->sentHeartBeats++; + 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) { + qWarning() << "SignalProxy: Disconnecting peer:" << ioPeer->address() + << "(didn't receive a heartbeat for over" << ioPeer->sentHeartBeats * _heartBeatTimer.interval() / 1000 << "seconds)"; + ioPeer->close(); + } else { + ioPeer->sentHeartBeats++; + } } - peerIter++; + peer++; } } -void SignalProxy::receiveHeartBeat(QIODevice *dev, const QVariantList ¶ms) { - if(!_peers.contains(dev)) { - qWarning() << "SignalProxy: received heart beat from unknown Device:" << dev; - } - dispatchSignal(dev, SignalProxy::HeartBeatReply, params); +void SignalProxy::receiveHeartBeat(AbstractPeer *peer, const QVariantList ¶ms) { + peer->dispatchSignal(SignalProxy::HeartBeatReply, params); } -void SignalProxy::receiveHeartBeatReply(QIODevice *dev, const QVariantList ¶ms) { - if(!_peers.contains(dev)) { - qWarning() << "SignalProxy: received heart beat reply from unknown Device:" << dev; +void SignalProxy::receiveHeartBeatReply(AbstractPeer *peer, const QVariantList ¶ms) { + if(peer->type() != AbstractPeer::IODevicePeer) { + qWarning() << "SignalProxy::receiveHeartBeatReply: received heart beat from a non IODevicePeer!"; return; } - _peers[dev].sentHeartBeats = 0; + IODevicePeer *ioPeer = static_cast(peer); + ioPeer->sentHeartBeats = 0; if(params.isEmpty()) { - qWarning() << "SignalProxy: received heart beat reply with less params then sent from:" << dev; + qWarning() << "SignalProxy: received heart beat reply with less params then sent from:" << ioPeer->address(); return; } QTime sendTime = params[0].value(); - updateLag(dev, (float)sendTime.msecsTo(QTime::currentTime()) / 2); + updateLag(ioPeer, sendTime.msecsTo(QTime::currentTime()) / 2); +} + +void SignalProxy::customEvent(QEvent *event) { + switch(event->type()) { + case QEvent::User: + { + PeerSignalEvent *sig = static_cast(event); + receivePeerSignal(sig->sender, sig->requestType, sig->params); + } + event->accept(); + break; + default: + return; + } } -void SignalProxy::updateLag(QIODevice *dev, float lag) { - Q_ASSERT(_peers.contains(dev)); - _peers[dev].lag = lag; +void SignalProxy::updateLag(IODevicePeer *peer, int lag) { + peer->lag = lag; if(proxyMode() == Client) { - qDebug() << "LAG" << lag; emit lagUpdated(lag); } }