src/common/signalproxy.cpp

   1 /***************************************************************************
   2  *   Copyright (C) 2005-09 by the Quassel Project                          *
   3  *   devel@quassel-irc.org                                                 *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) version 3.                                           *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15  *   You should have received a copy of the GNU General Public License     *
  16  *   along with this program; if not, write to the                         *
  17  *   Free Software Foundation, Inc.,                                       *
  18  *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
  19  ***************************************************************************/
  20
  21 #include "signalproxy.h"
  22
  23 #include <QObject>
  24 #include <QIODevice>
  25 #include <QAbstractSocket>
  26 #include <QHostAddress>
  27 #include <QHash>
  28 #include <QMultiHash>
  29 #include <QList>
  30 #include <QSet>
  31 #include <QDebug>
  32 #include <QMetaMethod>
  33 #include <QMetaProperty>
  34 #include <QRegExp>
  35 #ifdef HAVE_SSL
  36 #include <QSslSocket>
  37 #endif
  38 #include <QThread>
  39 #include <QTime>
  40 #include <QEvent>
  41 #include <QCoreApplication>
  42
  43 #include "syncableobject.h"
  44 #include "util.h"
  45
  46 // ==================================================
  47 //  PeerSignalEvent
  48 // ==================================================
  49 class PeerSignalEvent : public QEvent {
  50 public:
  51   PeerSignalEvent(SignalProxy *sender, SignalProxy::RequestType requestType, const QVariantList &params) : QEvent(QEvent::Type(SignalProxy::PeerSignal)), sender(sender), requestType(requestType), params(params) {}
  52   SignalProxy *sender;
  53   SignalProxy::RequestType requestType;
  54   QVariantList params;
  55 };
  56
  57 class RemovePeerEvent : public QEvent {
  58 public:
  59   RemovePeerEvent(QObject *peer) : QEvent(QEvent::Type(SignalProxy::RemovePeer)), peer(peer) {}
  60   QObject *peer;
  61 };
  62
  63 // ==================================================
  64 //  SignalRelay
  65 // ==================================================
  66 class SignalProxy::SignalRelay : public QObject {
  67 /* Q_OBJECT is not necessary or even allowed, because we implement
  68    qt_metacall ourselves (and don't use any other features of the meta
  69    object system)
  70 */
  71 public:
  72   SignalRelay(SignalProxy *parent) : QObject(parent), _proxy(parent) {}
  73   inline SignalProxy *proxy() const { return _proxy; }
  74
  75   int qt_metacall(QMetaObject::Call _c, int _id, void **_a);
  76
  77   void attachSignal(QObject *sender, int signalId, const QByteArray &funcName);
  78   void detachSignal(QObject *sender, int signalId = -1);
  79
  80 private:
  81   struct Signal {
  82     QObject *sender;
  83     int signalId;
  84     QByteArray signature;
  85     Signal(QObject *sender, int sigId, const QByteArray &signature) : sender(sender), signalId(sigId), signature(signature) {}
  86     Signal() : sender(0), signalId(-1) {}
  87   };
  88
  89   SignalProxy *_proxy;
  90   QHash<int, Signal> _slots;
  91 };
  92
  93 void SignalProxy::SignalRelay::attachSignal(QObject *sender, int signalId, const QByteArray &funcName) {
  94   // we ride without safetybelts here... all checking for valid method etc pp has to be done by the caller
  95   // all connected methodIds are offset by the standard methodCount of QObject
  96   int slotId;
  97   for(int i = 0;; i++) {
  98     if(!_slots.contains(i)) {
  99       slotId = i;
 100       break;
 101     }
 102   }
 103
 104   QByteArray fn;
 105   if(!funcName.isEmpty()) {
 106     fn = QMetaObject::normalizedSignature(funcName);
 107   } else {
 108     fn = SIGNAL(fakeMethodSignature());
 109     fn = fn.replace("fakeMethodSignature()", sender->metaObject()->method(signalId).signature());
 110   }
 111
 112   _slots[slotId] = Signal(sender, signalId, fn);
 113
 114   QMetaObject::connect(sender, signalId, this, QObject::staticMetaObject.methodCount() + slotId);
 115 }
 116
 117 void SignalProxy::SignalRelay::detachSignal(QObject *sender, int signalId) {
 118   QHash<int, Signal>::iterator slotIter = _slots.begin();
 119   while(slotIter != _slots.end()) {
 120     if(slotIter->sender == sender && (signalId == -1 || slotIter->signalId == signalId)) {
 121       slotIter = _slots.erase(slotIter);
 122       if(signalId != -1)
 123         break;
 124     } else {
 125       slotIter++;
 126     }
 127   }
 128 }
 129
 130 int SignalProxy::SignalRelay::qt_metacall(QMetaObject::Call _c, int _id, void **_a) {
 131   _id = QObject::qt_metacall(_c, _id, _a);
 132   if(_id < 0)
 133     return _id;
 134
 135   if(_c == QMetaObject::InvokeMetaMethod) {
 136     if(_slots.contains(_id)) {
 137       QObject *caller = sender();
 138
 139       SignalProxy::ExtendedMetaObject *eMeta = proxy()->extendedMetaObject(caller->metaObject());
 140       Q_ASSERT(eMeta);
 141
 142       const Signal &signal = _slots[_id];
 143
 144       QVariantList params;
 145       params << signal.signature;
 146
 147       const QList<int> &argTypes = eMeta->argTypes(signal.signalId);
 148       for(int i = 0; i < argTypes.size(); i++) {
 149         if(argTypes[i] == 0) {
 150           qWarning() << "SignalRelay::qt_metacall(): received invalid data for argument number" << i << "of signal" << QString("%1::%2").arg(caller->metaObject()->className()).arg(caller->metaObject()->method(_id).signature());
 151           qWarning() << "                            - make sure all your data types are known by the Qt MetaSystem";
 152           return _id;
 153         }
 154         params << QVariant(argTypes[i], _a[i+1]);
 155       }
 156
 157       proxy()->dispatchSignal(SignalProxy::RpcCall, params);
 158     }
 159     _id -= _slots.count();
 160   }
 161   return _id;
 162 }
 163
 164 // ==================================================
 165 //  Peers
 166 // ==================================================
 167 void SignalProxy::IODevicePeer::dispatchSignal(const RequestType &requestType, const QVariantList &params) {
 168   QVariantList packedFunc;
 169   packedFunc << (qint16)requestType
 170              << params;
 171   dispatchPackedFunc(QVariant(packedFunc));
 172 }
 173
 174 bool SignalProxy::IODevicePeer::isSecure() const {
 175 #ifdef HAVE_SSL
 176   QSslSocket *sslSocket = qobject_cast<QSslSocket *>(_device);
 177   if(sslSocket)
 178     return sslSocket->isEncrypted() || sslSocket->localAddress() == QHostAddress::LocalHost || sslSocket->localAddress() == QHostAddress::LocalHostIPv6;
 179 #endif
 180
 181   QAbstractSocket *socket = qobject_cast<QAbstractSocket *>(_device);
 182   if(socket)
 183     return socket->localAddress() == QHostAddress::LocalHost || socket->localAddress() == QHostAddress::LocalHostIPv6;
 184
 185   return false;
 186 }
 187
 188 QString SignalProxy::IODevicePeer::address() const {
 189   QAbstractSocket *socket = qobject_cast<QAbstractSocket *>(_device);
 190   if(socket)
 191     return socket->peerAddress().toString();
 192   else
 193     return QString();
 194 }
 195
 196 void SignalProxy::SignalProxyPeer::dispatchSignal(const RequestType &requestType, const QVariantList &params) {
 197   Qt::ConnectionType type = QThread::currentThread() == receiver->thread()
 198     ? Qt::DirectConnection
 199     : Qt::QueuedConnection;
 200
 201   if(type == Qt::DirectConnection) {
 202     receiver->receivePeerSignal(sender, requestType, params);
 203   } else {
 204     QCoreApplication::postEvent(receiver, new PeerSignalEvent(sender, requestType, params));
 205   }
 206 }
 207
 208 // ==================================================
 209 //  SignalProxy
 210 // ==================================================
 211 SignalProxy::SignalProxy(QObject* parent)
 212   : QObject(parent)
 213 {
 214   setProxyMode(Client);
 215   init();
 216 }
 217
 218 SignalProxy::SignalProxy(ProxyMode mode, QObject* parent)
 219   : QObject(parent)
 220 {
 221   setProxyMode(mode);
 222   init();
 223 }
 224
 225 SignalProxy::SignalProxy(ProxyMode mode, QIODevice* device, QObject* parent)
 226   : QObject(parent)
 227 {
 228   setProxyMode(mode);
 229   addPeer(device);
 230   init();
 231 }
 232
 233 SignalProxy::~SignalProxy() {
 234   QHash<QByteArray, ObjectId>::iterator classIter = _syncSlave.begin();
 235   while(classIter != _syncSlave.end()) {
 236     ObjectId::iterator objIter = classIter->begin();
 237     while(objIter != classIter->end()) {
 238       SyncableObject *obj = objIter.value();
 239       objIter = classIter->erase(objIter);
 240       obj->stopSynchronize(this);
 241     }
 242     classIter++;
 243   }
 244   _syncSlave.clear();
 245
 246   removeAllPeers();
 247 }
 248
 249 void SignalProxy::setProxyMode(ProxyMode mode) {
 250   PeerHash::iterator peer = _peers.begin();
 251   while(peer != _peers.end()) {
 252     if((*peer)->type() != AbstractPeer::IODevicePeer) {
 253       IODevicePeer *ioPeer = static_cast<IODevicePeer *>(*peer);
 254       if(ioPeer->isOpen()) {
 255         qWarning() << "SignalProxy: Cannot change proxy mode while connected";
 256         return;
 257       }
 258     }
 259     if((*peer)->type() != AbstractPeer::SignalProxyPeer) {
 260       qWarning() << "SignalProxy: Cannot change proxy mode while connected to another internal SignalProxy";
 261       return;
 262     }
 263     peer++;
 264   }
 265
 266   _proxyMode = mode;
 267   if(mode == Server)
 268     initServer();
 269   else
 270     initClient();
 271 }
 272
 273 void SignalProxy::init() {
 274   _signalRelay = new SignalRelay(this);
 275   connect(&_heartBeatTimer, SIGNAL(timeout()), this, SLOT(sendHeartBeat()));
 276   _heartBeatTimer.start(30 * 1000);
 277   _secure = false;
 278   updateSecureState();
 279 }
 280
 281 void SignalProxy::initServer() {
 282 }
 283
 284 void SignalProxy::initClient() {
 285   attachSlot("__objectRenamed__", this, SLOT(objectRenamed(QByteArray, QString, QString)));
 286 }
 287
 288 bool SignalProxy::addPeer(QIODevice* iodev) {
 289   if(!iodev)
 290     return false;
 291
 292   if(_peers.contains(iodev))
 293     return true;
 294
 295   if(proxyMode() == Client && !_peers.isEmpty()) {
 296     qWarning("SignalProxy: only one peer allowed in client mode!");
 297     return false;
 298   }
 299
 300   if(!iodev->isOpen()) {
 301     qWarning("SignalProxy::addPeer(QIODevice *iodev): iodev needs to be open!");
 302     return false;
 303   }
 304
 305   connect(iodev, SIGNAL(disconnected()), this, SLOT(removePeerBySender()));
 306   connect(iodev, SIGNAL(readyRead()), this, SLOT(dataAvailable()));
 307
 308 #ifdef HAVE_SSL
 309   QSslSocket *sslSocket = qobject_cast<QSslSocket *>(iodev);
 310   if(sslSocket) {
 311     connect(iodev, SIGNAL(encrypted()), this, SLOT(updateSecureState()));
 312   }
 313 #endif
 314
 315   if(!iodev->parent())
 316     iodev->setParent(this);
 317
 318   _peers[iodev] = new IODevicePeer(iodev, iodev->property("UseCompression").toBool());
 319
 320   if(_peers.count() == 1)
 321     emit connected();
 322
 323   updateSecureState();
 324   return true;
 325 }
 326
 327 bool SignalProxy::addPeer(SignalProxy* proxy) {
 328   if(!proxy)
 329     return false;
 330
 331   if(proxyMode() == proxy->proxyMode()) {
 332     qWarning() << "SignalProxy::addPeer(): adding a SignalProxy as peer requires one proxy to be server and one client!";
 333     return false;
 334   }
 335
 336   if(_peers.contains(proxy)) {
 337     return true;
 338   }
 339
 340   if(proxyMode() == Client && !_peers.isEmpty()) {
 341     qWarning("SignalProxy: only one peer allowed in client mode!");
 342     return false;
 343   }
 344
 345   _peers[proxy] = new SignalProxyPeer(this, proxy);
 346   proxy->addPeer(this);
 347
 348   if(_peers.count() == 1)
 349     emit connected();
 350
 351   updateSecureState();
 352   return true;
 353 }
 354
 355 void SignalProxy::removeAllPeers() {
 356   Q_ASSERT(proxyMode() == Server || _peers.count() <= 1);
 357   // wee need to copy that list since we modify it in the loop
 358   QList<QObject *> peers = _peers.keys();
 359   foreach(QObject *peer, peers) {
 360     removePeer(peer);
 361   }
 362 }
 363
 364 void SignalProxy::removePeer(QObject* dev) {
 365   if(_peers.isEmpty()) {
 366     qWarning() << "SignalProxy::removePeer(): No peers in use!";
 367     return;
 368   }
 369
 370   Q_ASSERT(dev);
 371   if(!_peers.contains(dev)) {
 372     qWarning() << "SignalProxy: unknown Peer" << dev;
 373     return;
 374   }
 375
 376   AbstractPeer *peer = _peers[dev];
 377   _peers.remove(dev);
 378
 379   disconnect(dev, 0, this, 0);
 380   if(peer->type() == AbstractPeer::IODevicePeer)
 381     emit peerRemoved(static_cast<QIODevice *>(dev));
 382
 383   if(peer->type() == AbstractPeer::SignalProxyPeer) {
 384     SignalProxy *proxy = static_cast<SignalProxy *>(dev);
 385     if(proxy->_peers.contains(this))
 386       proxy->removePeer(this);
 387   }
 388
 389   if(dev->parent() == this)
 390     dev->deleteLater();
 391
 392   delete peer;
 393
 394   updateSecureState();
 395
 396   if(_peers.isEmpty())
 397     emit disconnected();
 398 }
 399
 400 void SignalProxy::removePeerBySender() {
 401   removePeer(sender());
 402 }
 403
 404 void SignalProxy::renameObject(const SyncableObject *obj, const QString &newname, const QString &oldname) {
 405   if(proxyMode() == Client)
 406     return;
 407
 408   const QMetaObject *meta = obj->syncMetaObject();
 409   const QByteArray className(meta->className());
 410   objectRenamed(className, newname, oldname);
 411
 412   QVariantList params;
 413   params << "__objectRenamed__" << className << newname << oldname;
 414   dispatchSignal(RpcCall, params);
 415 }
 416
 417 void SignalProxy::objectRenamed(const QByteArray &classname, const QString &newname, const QString &oldname) {
 418   if(_syncSlave.contains(classname) && _syncSlave[classname].contains(oldname) && oldname != newname) {
 419     SyncableObject *obj = _syncSlave[classname][newname] = _syncSlave[classname].take(oldname);
 420     requestInit(obj);
 421   }
 422 }
 423
 424 const QMetaObject *SignalProxy::metaObject(const QObject *obj) {
 425   if(const SyncableObject *syncObject = qobject_cast<const SyncableObject *>(obj))
 426     return syncObject->syncMetaObject();
 427   else
 428     return obj->metaObject();
 429 }
 430
 431 SignalProxy::ExtendedMetaObject *SignalProxy::extendedMetaObject(const QMetaObject *meta) const {
 432   if(_extendedMetaObjects.contains(meta))
 433     return _extendedMetaObjects[meta];
 434   else
 435     return 0;
 436 }
 437
 438 SignalProxy::ExtendedMetaObject *SignalProxy::createExtendedMetaObject(const QMetaObject *meta, bool checkConflicts) {
 439   if(!_extendedMetaObjects.contains(meta)) {
 440     _extendedMetaObjects[meta] = new ExtendedMetaObject(meta, checkConflicts);
 441   }
 442   return _extendedMetaObjects[meta];
 443 }
 444
 445 bool SignalProxy::attachSignal(QObject *sender, const char* signal, const QByteArray& sigName) {
 446   const QMetaObject* meta = sender->metaObject();
 447   QByteArray sig(meta->normalizedSignature(signal).mid(1));
 448   int methodId = meta->indexOfMethod(sig.constData());
 449   if(methodId == -1 || meta->method(methodId).methodType() != QMetaMethod::Signal) {
 450     qWarning() << "SignalProxy::attachSignal(): No such signal" << signal;
 451     return false;
 452   }
 453
 454   createExtendedMetaObject(meta);
 455   _signalRelay->attachSignal(sender, methodId, sigName);
 456
 457   disconnect(sender, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 458   connect(sender, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 459   return true;
 460 }
 461
 462 bool SignalProxy::attachSlot(const QByteArray& sigName, QObject* recv, const char* slot) {
 463   const QMetaObject* meta = recv->metaObject();
 464   int methodId = meta->indexOfMethod(meta->normalizedSignature(slot).mid(1));
 465   if(methodId == -1 || meta->method(methodId).methodType() == QMetaMethod::Method) {
 466     qWarning() << "SignalProxy::attachSlot(): No such slot" << slot;
 467     return false;
 468   }
 469
 470   createExtendedMetaObject(meta);
 471
 472   QByteArray funcName = QMetaObject::normalizedSignature(sigName.constData());
 473   _attachedSlots.insert(funcName, qMakePair(recv, methodId));
 474
 475   disconnect(recv, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 476   connect(recv, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 477   return true;
 478 }
 479
 480 void SignalProxy::synchronize(SyncableObject *obj) {
 481   createExtendedMetaObject(obj, true);
 482
 483   // attaching as slave to receive sync Calls
 484   QByteArray className(obj->syncMetaObject()->className());
 485   _syncSlave[className][obj->objectName()] = obj;
 486
 487   if(proxyMode() == Server) {
 488     obj->setInitialized();
 489     emit objectInitialized(obj);
 490   } else {
 491     if(obj->isInitialized())
 492       emit objectInitialized(obj);
 493     else
 494       requestInit(obj);
 495   }
 496
 497   obj->synchronize(this);
 498 }
 499
 500 void SignalProxy::detachObject(QObject *obj) {
 501   detachSignals(obj);
 502   detachSlots(obj);
 503 }
 504
 505 void SignalProxy::detachSignals(QObject *sender) {
 506   _signalRelay->detachSignal(sender);
 507 }
 508
 509 void SignalProxy::detachSlots(QObject* receiver) {
 510   SlotHash::iterator slotIter = _attachedSlots.begin();
 511   while(slotIter != _attachedSlots.end()) {
 512     if(slotIter.value().first == receiver) {
 513       slotIter = _attachedSlots.erase(slotIter);
 514     } else
 515       slotIter++;
 516   }
 517 }
 518
 519 void SignalProxy::stopSynchronize(SyncableObject *obj) {
 520   // we can't use a className here, since it might be effed up, if we receive the call as a result of a decon
 521   // gladly the objectName() is still valid. So we have only to iterate over the classes not each instance! *sigh*
 522   QHash<QByteArray, ObjectId>::iterator classIter = _syncSlave.begin();
 523   while(classIter != _syncSlave.end()) {
 524     if(classIter->contains(obj->objectName()) && classIter.value()[obj->objectName()] == obj) {
 525       classIter->remove(obj->objectName());
 526       break;
 527     }
 528     classIter++;
 529   }
 530   obj->stopSynchronize(this);
 531 }
 532
 533 void SignalProxy::dispatchSignal(const RequestType &requestType, const QVariantList &params) {
 534   QVariant packedFunc(QVariantList() << (qint16)requestType << params);
 535   PeerHash::iterator peer = _peers.begin();
 536   while(peer != _peers.end()) {
 537     switch((*peer)->type()) {
 538     case AbstractPeer::IODevicePeer:
 539       {
 540         IODevicePeer *ioPeer = static_cast<IODevicePeer *>(*peer);
 541         if(ioPeer->isOpen())
 542           ioPeer->dispatchPackedFunc(packedFunc);
 543         else
 544           QCoreApplication::postEvent(this, new RemovePeerEvent(peer.key()));
 545       }
 546       break;
 547     case AbstractPeer::SignalProxyPeer:
 548       (*peer)->dispatchSignal(requestType, params);
 549       break;
 550     default:
 551       Q_ASSERT(false); // there shouldn't be any peers with wrong / unknown type
 552     }
 553     peer++;
 554   }
 555 }
 556
 557 void SignalProxy::receivePackedFunc(AbstractPeer *sender, const QVariant &packedFunc) {
 558   QVariantList params(packedFunc.toList());
 559
 560   if(params.isEmpty()) {
 561     qWarning() << "SignalProxy::receivePeerSignal(): received incompatible Data:" << packedFunc;
 562     return;
 563   }
 564
 565   RequestType requestType = (RequestType)params.takeFirst().value<int>();
 566   receivePeerSignal(sender, requestType, params);
 567 }
 568
 569 void SignalProxy::receivePeerSignal(AbstractPeer *sender, const RequestType &requestType, const QVariantList &params) {
 570   switch(requestType) {
 571     // list all RequestTypes that shouldnot trigger a heartbeat counter reset here
 572   case HeartBeatReply:
 573     break;
 574   default:
 575     if(sender->type() == AbstractPeer::IODevicePeer) {
 576       IODevicePeer *ioPeer = static_cast<IODevicePeer *>(sender);
 577       ioPeer->sentHeartBeats = 0;
 578     }
 579   }
 580
 581   // qDebug() << "SignalProxy::receivePeerSignal)" << requestType << params;
 582   switch(requestType) {
 583   case RpcCall:
 584     if(params.empty())
 585       qWarning() << "SignalProxy::receivePeerSignal(): received empty RPC-Call";
 586     else
 587       handleSignal(params);
 588       //handleSignal(params.takeFirst().toByteArray(), params);
 589     break;
 590
 591   case Sync:
 592     handleSync(sender, params);
 593     break;
 594
 595   case InitRequest:
 596     handleInitRequest(sender, params);
 597     break;
 598
 599   case InitData:
 600     handleInitData(sender, params);
 601     break;
 602
 603   case HeartBeat:
 604     receiveHeartBeat(sender, params);
 605     break;
 606
 607   case HeartBeatReply:
 608     receiveHeartBeatReply(sender, params);
 609     break;
 610
 611   default:
 612     qWarning() << "SignalProxy::receivePeerSignal(): received undefined CallType" << requestType << params;
 613   }
 614 }
 615
 616 void SignalProxy::receivePeerSignal(SignalProxy *sender, const RequestType &requestType, const QVariantList &params) {
 617   if(!_peers.contains(sender)) {
 618     // 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.
 619     qWarning() << "SignalProxy::receivePeerSignal(): received Signal from unknown Proxy" << reinterpret_cast<void *>(sender);
 620     return;
 621   }
 622   receivePeerSignal(_peers[sender], requestType, params);
 623 }
 624
 625 void SignalProxy::handleSync(AbstractPeer *sender, QVariantList params) {
 626   if(params.count() < 3) {
 627     qWarning() << "received invalid Sync call" << params;
 628     return;
 629   }
 630
 631   QByteArray className = params.takeFirst().toByteArray();
 632   QString objectName = params.takeFirst().toString();
 633   QByteArray slot = params.takeFirst().toByteArray();
 634
 635   if(!_syncSlave.contains(className) || !_syncSlave[className].contains(objectName)) {
 636     qWarning() << QString("no registered receiver for sync call: %1::%2 (objectName=\"%3\"). Params are:").arg(QString(className)).arg(QString(slot)).arg(objectName)
 637                << params;
 638     return;
 639   }
 640
 641   SyncableObject *receiver = _syncSlave[className][objectName];
 642   ExtendedMetaObject *eMeta = extendedMetaObject(receiver);
 643   if(!eMeta->slotMap().contains(slot)) {
 644     qWarning() << QString("no matching slot for sync call: %1::%2 (objectName=\"%3\"). Params are:").arg(QString(className)).arg(QString(slot)).arg(objectName)
 645                << params;
 646     return;
 647   }
 648
 649   int slotId = eMeta->slotMap()[slot];
 650   if(proxyMode() != eMeta->receiverMode(slotId)) {
 651     qWarning("SignalProxy::handleSync(): invokeMethod for \"%s\" failed. Wrong ProxyMode!", eMeta->methodName(slotId).constData());
 652     return;
 653   }
 654
 655   QVariant returnValue((QVariant::Type)eMeta->returnType(slotId));
 656   if(!invokeSlot(receiver, slotId, params, returnValue)) {
 657     qWarning("SignalProxy::handleSync(): invokeMethod for \"%s\" failed ", eMeta->methodName(slotId).constData());
 658     return;
 659   }
 660
 661
 662   if(returnValue.type() != QVariant::Invalid && eMeta->receiveMap().contains(slotId)) {
 663     int receiverId = eMeta->receiveMap()[slotId];
 664     QVariantList returnParams;
 665     returnParams << className
 666                  << objectName
 667                  << eMeta->methodName(receiverId);
 668     //QByteArray(receiver->metaObject()->method(receiverId).signature());
 669     if(eMeta->argTypes(receiverId).count() > 1)
 670       returnParams << params;
 671     returnParams << returnValue;
 672     sender->dispatchSignal(Sync, returnParams);
 673   }
 674
 675   // send emit update signal
 676   invokeSlot(receiver, eMeta->updatedRemotelyId());
 677 }
 678
 679 void SignalProxy::handleInitRequest(AbstractPeer *sender, const QVariantList &params) {
 680   if(params.count() != 2) {
 681     qWarning() << "SignalProxy::handleInitRequest() received initRequest with invalid param Count:"
 682                << params;
 683     return;
 684   }
 685
 686   QByteArray className(params[0].toByteArray());
 687   QString objectName(params[1].toString());
 688
 689   if(!_syncSlave.contains(className)) {
 690     qWarning() << "SignalProxy::handleInitRequest() received initRequest for unregistered Class:"
 691                << className;
 692     return;
 693   }
 694
 695   if(!_syncSlave[className].contains(objectName)) {
 696     qWarning() << "SignalProxy::handleInitRequest() received initRequest for unregistered Object:"
 697                << className << objectName;
 698     return;
 699   }
 700
 701   SyncableObject *obj = _syncSlave[className][objectName];
 702
 703   QVariantList params_;
 704   params_ << className
 705           << objectName
 706           << initData(obj);
 707
 708   sender->dispatchSignal(InitData, params_);
 709 }
 710
 711 void SignalProxy::handleInitData(AbstractPeer *sender, const QVariantList &params) {
 712   Q_UNUSED(sender)
 713   if(params.count() != 3) {
 714     qWarning() << "SignalProxy::handleInitData() received initData with invalid param Count:"
 715                << params;
 716     return;
 717   }
 718
 719   QByteArray className(params[0].toByteArray());
 720   QString objectName(params[1].toString());
 721   QVariantMap propertyMap(params[2].toMap());
 722
 723   if(!_syncSlave.contains(className)) {
 724     qWarning() << "SignalProxy::handleInitData() received initData for unregistered Class:"
 725                << className;
 726     return;
 727   }
 728
 729   if(!_syncSlave[className].contains(objectName)) {
 730     qWarning() << "SignalProxy::handleInitData() received initData for unregistered Object:"
 731                << className << objectName;
 732     return;
 733   }
 734
 735   SyncableObject *obj = _syncSlave[className][objectName];
 736   setInitData(obj, propertyMap);
 737 }
 738
 739 //void SignalProxy::handleSignal(const QByteArray &funcName, const QVariantList &params) {
 740 void SignalProxy::handleSignal(const QVariantList &data) {
 741   QVariantList params = data;
 742   QByteArray funcName = params.takeFirst().toByteArray();
 743
 744   QObject* receiver;
 745   int methodId;
 746   SlotHash::const_iterator slot = _attachedSlots.constFind(funcName);
 747   while(slot != _attachedSlots.constEnd() && slot.key() == funcName) {
 748     receiver = (*slot).first;
 749     methodId = (*slot).second;
 750     if(!invokeSlot(receiver, methodId, params)) {
 751       ExtendedMetaObject *eMeta = extendedMetaObject(receiver);
 752       qWarning("SignalProxy::handleSignal(): invokeMethod for \"%s\" failed ", eMeta->methodName(methodId).constData());
 753     }
 754     slot++;
 755   }
 756 }
 757
 758 bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList &params, QVariant &returnValue) {
 759   ExtendedMetaObject *eMeta = extendedMetaObject(receiver);
 760   const QList<int> args = eMeta->argTypes(methodId);
 761   const int numArgs = params.count() < args.count()
 762     ? params.count()
 763     : args.count();
 764
 765   if(eMeta->minArgCount(methodId) > params.count()) {
 766       qWarning() << "SignalProxy::invokeSlot(): not enough params to invoke" << eMeta->methodName(methodId);
 767       return false;
 768   }
 769
 770   void *_a[] = {0,              // return type...
 771                 0, 0, 0, 0 , 0, // and 10 args - that's the max size qt can handle with signals and slots
 772                 0, 0, 0, 0 , 0};
 773
 774   // check for argument compatibility and build params array
 775   for(int i = 0; i < numArgs; i++) {
 776     if(!params[i].isValid()) {
 777       qWarning() << "SignalProxy::invokeSlot(): received invalid data for argument number" << i << "of method" << QString("%1::%2()").arg(receiver->metaObject()->className()).arg(receiver->metaObject()->method(methodId).signature());
 778       qWarning() << "                            - make sure all your data types are known by the Qt MetaSystem";
 779       return false;
 780     }
 781     if(args[i] != QMetaType::type(params[i].typeName())) {
 782       qWarning() << "SignalProxy::invokeSlot(): incompatible param types to invoke" << eMeta->methodName(methodId);
 783       return false;
 784     }
 785     _a[i+1] = const_cast<void *>(params[i].constData());
 786   }
 787
 788   if(returnValue.type() != QVariant::Invalid)
 789     _a[0] = const_cast<void *>(returnValue.constData());
 790
 791   Qt::ConnectionType type = QThread::currentThread() == receiver->thread()
 792     ? Qt::DirectConnection
 793     : Qt::QueuedConnection;
 794
 795   if(type == Qt::DirectConnection) {
 796     return receiver->qt_metacall(QMetaObject::InvokeMetaMethod, methodId, _a) < 0;
 797   } else {
 798     qWarning() << "Queued Connections are not implemented yet";
 799     // note to self: qmetaobject.cpp:990 ff
 800     return false;
 801   }
 802
 803 }
 804
 805 bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList &params) {
 806   QVariant ret;
 807   return invokeSlot(receiver, methodId, params, ret);
 808 }
 809
 810 void SignalProxy::dataAvailable() {
 811   // yet again. it's a private slot. no need for checks.
 812   QIODevice* ioDev = qobject_cast<QIODevice* >(sender());
 813   Q_ASSERT(_peers.contains(ioDev) && _peers[ioDev]->type() == AbstractPeer::IODevicePeer);
 814   IODevicePeer *peer = static_cast<IODevicePeer *>(_peers[ioDev]);
 815   QVariant var;
 816   while(peer->readData(var))
 817     receivePackedFunc(peer, var);
 818 }
 819
 820 void SignalProxy::writeDataToDevice(QIODevice *dev, const QVariant &item, bool compressed) {
 821   QAbstractSocket* sock  = qobject_cast<QAbstractSocket*>(dev);
 822   if(!dev->isOpen() || (sock && sock->state()!=QAbstractSocket::ConnectedState)) {
 823     qWarning("SignalProxy: Can't call write on a closed device");
 824     return;
 825   }
 826
 827   QByteArray block;
 828   QDataStream out(&block, QIODevice::WriteOnly);
 829   out.setVersion(QDataStream::Qt_4_2);
 830   out << (quint32)0;
 831
 832   if(compressed) {
 833     QByteArray rawItem;
 834     QDataStream itemStream(&rawItem, QIODevice::WriteOnly);
 835
 836     itemStream.setVersion(QDataStream::Qt_4_2);
 837     itemStream << item;
 838
 839     rawItem = qCompress(rawItem);
 840
 841     out << rawItem;
 842   } else {
 843     out << item;
 844   }
 845
 846   out.device()->seek(0);
 847   out << (quint32)(block.size() - sizeof(quint32));
 848
 849   dev->write(block);
 850 }
 851
 852 bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVariant &item, bool compressed) {
 853   if(!dev)
 854     return false;
 855
 856   QDataStream in(dev);
 857   in.setVersion(QDataStream::Qt_4_2);
 858
 859   if(blockSize == 0) {
 860     if(dev->bytesAvailable() < (int)sizeof(quint32)) return false;
 861     in >> blockSize;
 862   }
 863
 864   if(blockSize > 1 << 22) {
 865     disconnectDevice(dev, tr("Peer tried to send package larger than max package size!"));
 866     return false;
 867   }
 868
 869   if(blockSize == 0) {
 870     disconnectDevice(dev, tr("Peer tried to send 0 byte package!"));
 871     return false;
 872   }
 873
 874   if(dev->bytesAvailable() < blockSize)
 875     return false;
 876
 877   blockSize = 0;
 878
 879   if(compressed) {
 880     QByteArray rawItem;
 881     in >> rawItem;
 882
 883     int nbytes = rawItem.size();
 884     if(nbytes <= 4) {
 885       const char *data = rawItem.constData();
 886       if(nbytes < 4 || (data[0]!=0 || data[1]!=0 || data[2]!=0 || data[3]!=0)) {
 887         disconnectDevice(dev, tr("Peer sent corrupted compressed data!"));
 888         return false;
 889       }
 890     }
 891
 892     rawItem = qUncompress(rawItem);
 893
 894     QDataStream itemStream(&rawItem, QIODevice::ReadOnly);
 895     itemStream.setVersion(QDataStream::Qt_4_2);
 896     itemStream >> item;
 897   } else {
 898     in >> item;
 899   }
 900
 901   if(!item.isValid()) {
 902     disconnectDevice(dev, tr("Peer sent corrupt data: unable to load QVariant!"));
 903     return false;
 904   }
 905
 906   return true;
 907 }
 908
 909 void SignalProxy::requestInit(SyncableObject *obj) {
 910   if(proxyMode() == Server || obj->isInitialized())
 911     return;
 912
 913   QVariantList params;
 914   params << obj->syncMetaObject()->className()
 915          << obj->objectName();
 916   dispatchSignal(InitRequest, params);
 917 }
 918
 919 QVariantMap SignalProxy::initData(SyncableObject *obj) const {
 920   return obj->toVariantMap();
 921 }
 922
 923 void SignalProxy::setInitData(SyncableObject *obj, const QVariantMap &properties) {
 924   if(obj->isInitialized())
 925     return;
 926   obj->fromVariantMap(properties);
 927   obj->setInitialized();
 928   emit objectInitialized(obj);
 929   invokeSlot(obj, extendedMetaObject(obj)->updatedRemotelyId());
 930 }
 931
 932 void SignalProxy::sendHeartBeat() {
 933   QVariantList heartBeatParams;
 934   heartBeatParams << QTime::currentTime();
 935   QList<IODevicePeer *> toClose;
 936
 937   PeerHash::iterator peer = _peers.begin();
 938   while(peer != _peers.end()) {
 939     if((*peer)->type() == AbstractPeer::IODevicePeer) {
 940       IODevicePeer *ioPeer = static_cast<IODevicePeer *>(*peer);
 941       ioPeer->dispatchSignal(SignalProxy::HeartBeat, heartBeatParams);
 942       if(ioPeer->sentHeartBeats > 0) {
 943         updateLag(ioPeer, ioPeer->sentHeartBeats * _heartBeatTimer.interval());
 944       }
 945       if(ioPeer->sentHeartBeats > 1)
 946         toClose.append(ioPeer);
 947       else
 948         ioPeer->sentHeartBeats++;
 949     }
 950     ++peer;
 951   }
 952
 953   foreach(IODevicePeer *ioPeer, toClose) {
 954     qWarning() << "SignalProxy: Disconnecting peer:" << ioPeer->address()
 955                << "(didn't receive a heartbeat for over" << ioPeer->sentHeartBeats * _heartBeatTimer.interval() / 1000 << "seconds)";
 956     ioPeer->close();
 957   }
 958 }
 959
 960 void SignalProxy::receiveHeartBeat(AbstractPeer *peer, const QVariantList &params) {
 961   peer->dispatchSignal(SignalProxy::HeartBeatReply, params);
 962 }
 963
 964 void SignalProxy::receiveHeartBeatReply(AbstractPeer *peer, const QVariantList &params) {
 965   if(peer->type() != AbstractPeer::IODevicePeer) {
 966     qWarning() << "SignalProxy::receiveHeartBeatReply: received heart beat from a non IODevicePeer!";
 967     return;
 968   }
 969
 970   IODevicePeer *ioPeer = static_cast<IODevicePeer *>(peer);
 971   ioPeer->sentHeartBeats = 0;
 972
 973   if(params.isEmpty()) {
 974     qWarning() << "SignalProxy: received heart beat reply with less params then sent from:" << ioPeer->address();
 975     return;
 976   }
 977
 978   QTime sendTime = params[0].value<QTime>();
 979   updateLag(ioPeer, sendTime.msecsTo(QTime::currentTime()) / 2);
 980 }
 981
 982 void SignalProxy::customEvent(QEvent *event) {
 983   switch(event->type()) {
 984   case PeerSignal:
 985     {
 986       PeerSignalEvent *e = static_cast<PeerSignalEvent *>(event);
 987       receivePeerSignal(e->sender, e->requestType, e->params);
 988     }
 989     event->accept();
 990     break;
 991   case RemovePeer:
 992     {
 993       RemovePeerEvent *e = static_cast<RemovePeerEvent *>(event);
 994       removePeer(e->peer);
 995     }
 996     event->accept();
 997   default:
 998     return;
 999   }
1000 }
1001
1002 void SignalProxy::sync_call__(const SyncableObject *obj, SignalProxy::ProxyMode modeType, const char *funcname, va_list ap) {
1003   // qDebug() << obj << modeType << "(" << _proxyMode << ")" << funcname;
1004   if(modeType != _proxyMode)
1005     return;
1006
1007   ExtendedMetaObject *eMeta = extendedMetaObject(obj);
1008
1009   QVariantList params;
1010   params << eMeta->metaObject()->className()
1011          << obj->objectName()
1012          << QByteArray(funcname);
1013
1014   const QList<int> &argTypes = eMeta->argTypes(eMeta->methodId(QByteArray(funcname)));
1015
1016   for(int i = 0; i < argTypes.size(); i++) {
1017     if(argTypes[i] == 0) {
1018       qWarning() << Q_FUNC_INFO << "received invalid data for argument number" << i << "of signal" << QString("%1::%2").arg(eMeta->metaObject()->className()).arg(funcname);
1019       qWarning() << "        - make sure all your data types are known by the Qt MetaSystem";
1020       return;
1021     }
1022     params << QVariant(argTypes[i], va_arg(ap, void *));
1023   }
1024
1025   dispatchSignal(Sync, params);
1026 }
1027
1028
1029
1030 void SignalProxy::disconnectDevice(QIODevice *dev, const QString &reason) {
1031   if(!reason.isEmpty())
1032     qWarning() << qPrintable(reason);
1033   QAbstractSocket *sock  = qobject_cast<QAbstractSocket *>(dev);
1034   if(sock)
1035     qWarning() << qPrintable(tr("Disconnecting")) << qPrintable(sock->peerAddress().toString());
1036   dev->close();
1037 }
1038
1039 void SignalProxy::updateLag(IODevicePeer *peer, int lag) {
1040   peer->lag = lag;
1041   if(proxyMode() == Client) {
1042     emit lagUpdated(lag);
1043   }
1044 }
1045
1046 void SignalProxy::dumpProxyStats() {
1047   QString mode;
1048   if(proxyMode() == Server)
1049     mode = "Server";
1050   else
1051     mode = "Client";
1052
1053   int slaveCount = 0;
1054   foreach(ObjectId oid, _syncSlave.values())
1055     slaveCount += oid.count();
1056
1057   qDebug() << this;
1058   qDebug() << "              Proxy Mode:" << mode;
1059   qDebug() << "          attached Slots:" << _attachedSlots.count();
1060   qDebug() << " number of synced Slaves:" << slaveCount;
1061   qDebug() << "number of Classes cached:" << _extendedMetaObjects.count();
1062 }
1063
1064 void SignalProxy::updateSecureState() {
1065   bool wasSecure = _secure;
1066
1067   _secure = !_peers.isEmpty();
1068   PeerHash::const_iterator peerIter;
1069   for(peerIter = _peers.constBegin(); peerIter != _peers.constEnd(); peerIter++) {
1070     _secure &= (*peerIter)->isSecure();
1071   }
1072
1073   if(wasSecure != _secure)
1074     emit secureStateChanged(_secure);
1075 }
1076
1077
1078 // ==================================================
1079 //  ExtendedMetaObject
1080 // ==================================================
1081 SignalProxy::ExtendedMetaObject::ExtendedMetaObject(const QMetaObject *meta, bool checkConflicts)
1082   : _meta(meta),
1083     _updatedRemotelyId(_meta->indexOfSignal("updatedRemotely()"))
1084 {
1085   for(int i = 0; i < _meta->methodCount(); i++) {
1086     if(_meta->method(i).methodType() != QMetaMethod::Slot)
1087       continue;
1088
1089     if(QByteArray(_meta->method(i).signature()).contains('*'))
1090       continue; // skip methods with ptr params
1091
1092     QByteArray method = methodName(_meta->method(i));
1093     if(method.startsWith("init"))
1094       continue; // skip initializers
1095
1096     if(_methodIds.contains(method)) {
1097       /* funny... moc creates for methods containing default parameters multiple metaMethod with separate methodIds.
1098          we don't care... we just need the full fledged version
1099        */
1100       const QMetaMethod &current = _meta->method(_methodIds[method]);
1101       const QMetaMethod &candidate = _meta->method(i);
1102       if(current.parameterTypes().count() > candidate.parameterTypes().count()) {
1103         int minCount = candidate.parameterTypes().count();
1104         QList<QByteArray> commonParams = current.parameterTypes().mid(0, minCount);
1105         if(commonParams == candidate.parameterTypes())
1106           continue; // we already got the full featured version
1107       } else {
1108         int minCount = current.parameterTypes().count();
1109         QList<QByteArray> commonParams = candidate.parameterTypes().mid(0, minCount);
1110         if(commonParams == current.parameterTypes()) {
1111           _methodIds[method] = i; // use the new one
1112           continue;
1113         }
1114       }
1115       if(checkConflicts) {
1116         qWarning() << "class" << meta->className() << "contains overloaded methods which is currently not supported!";
1117         qWarning() << " - " << _meta->method(i).signature() << "conflicts with" << _meta->method(_methodIds[method]).signature();
1118       }
1119       continue;
1120     }
1121     _methodIds[method] = i;
1122   }
1123 }
1124
1125 const SignalProxy::ExtendedMetaObject::MethodDescriptor &SignalProxy::ExtendedMetaObject::methodDescriptor(int methodId) {
1126   if(!_methods.contains(methodId)) {
1127     _methods[methodId] = MethodDescriptor(_meta->method(methodId));
1128   }
1129   return _methods[methodId];
1130 }
1131
1132 const QHash<int, int> &SignalProxy::ExtendedMetaObject::receiveMap() {
1133   if(_receiveMap.isEmpty()) {
1134     QHash<int, int> receiveMap;
1135
1136     QMetaMethod requestSlot;
1137     QByteArray returnTypeName;
1138     QByteArray signature;
1139     QByteArray methodName;
1140     QByteArray params;
1141     int paramsPos;
1142     int receiverId;
1143     const int methodCount = _meta->methodCount();
1144     for(int i = 0; i < methodCount; i++) {
1145       requestSlot = _meta->method(i);
1146       if(requestSlot.methodType() != QMetaMethod::Slot)
1147         continue;
1148
1149       returnTypeName = requestSlot.typeName();
1150       if(QMetaType::Void == (QMetaType::Type)returnType(i))
1151         continue;
1152
1153       signature = QByteArray(requestSlot.signature());
1154       if(!signature.startsWith("request"))
1155         continue;
1156
1157       paramsPos = signature.indexOf('(');
1158       if(paramsPos == -1)
1159         continue;
1160
1161       methodName = signature.left(paramsPos);
1162       params = signature.mid(paramsPos);
1163
1164       methodName = methodName.replace("request", "receive");
1165       params = params.left(params.count() - 1) + ", " + returnTypeName + ")";
1166
1167       signature = QMetaObject::normalizedSignature(methodName + params);
1168       receiverId = _meta->indexOfSlot(signature);
1169
1170       if(receiverId == -1) {
1171         signature = QMetaObject::normalizedSignature(methodName + "(" + returnTypeName + ")");
1172         receiverId = _meta->indexOfSlot(signature);
1173       }
1174
1175       if(receiverId != -1) {
1176         receiveMap[i] = receiverId;
1177       }
1178     }
1179     _receiveMap = receiveMap;
1180   }
1181   return _receiveMap;
1182 }
1183
1184 QByteArray SignalProxy::ExtendedMetaObject::methodName(const QMetaMethod &method) {
1185   QByteArray sig(method.signature());
1186   return sig.left(sig.indexOf("("));
1187 }
1188
1189 QString SignalProxy::ExtendedMetaObject::methodBaseName(const QMetaMethod &method) {
1190   QString methodname = QString(method.signature()).section("(", 0, 0);
1191
1192   // determine where we have to chop:
1193   int upperCharPos;
1194   if(method.methodType() == QMetaMethod::Slot) {
1195     // we take evertyhing from the first uppercase char if it's slot
1196     upperCharPos = methodname.indexOf(QRegExp("[A-Z]"));
1197     if(upperCharPos == -1)
1198       return QString();
1199     methodname = methodname.mid(upperCharPos);
1200   } else {
1201     // and if it's a signal we discard everything from the last uppercase char
1202     upperCharPos = methodname.lastIndexOf(QRegExp("[A-Z]"));
1203     if(upperCharPos == -1)
1204       return QString();
1205     methodname = methodname.left(upperCharPos);
1206   }
1207
1208   methodname[0] = methodname[0].toUpper();
1209
1210   return methodname;
1211 }
1212
1213 SignalProxy::ExtendedMetaObject::MethodDescriptor::MethodDescriptor(const QMetaMethod &method)
1214   : _methodName(SignalProxy::ExtendedMetaObject::methodName(method)),
1215     _returnType(QMetaType::type(method.typeName()))
1216 {
1217   // determine argTypes
1218   QList<QByteArray> paramTypes = method.parameterTypes();
1219   QList<int> argTypes;
1220   for(int i = 0; i < paramTypes.count(); i++) {
1221     argTypes.append(QMetaType::type(paramTypes[i]));
1222   }
1223   _argTypes = argTypes;
1224
1225   // determine minArgCount
1226   QString signature(method.signature());
1227   _minArgCount = method.parameterTypes().count() - signature.count("=");
1228
1229   _receiverMode = (_methodName.startsWith("request"))
1230     ? SignalProxy::Server
1231     : SignalProxy::Client;
1232 }
1233