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   setHeartBeatInterval(30);
 277   setMaxHeartBeatCount(2);
 278   _secure = false;
 279   updateSecureState();
 280 }
 281
 282 void SignalProxy::initServer() {
 283 }
 284
 285 void SignalProxy::initClient() {
 286   attachSlot("__objectRenamed__", this, SLOT(objectRenamed(QByteArray, QString, QString)));
 287 }
 288
 289 bool SignalProxy::addPeer(QIODevice* iodev) {
 290   if(!iodev)
 291     return false;
 292
 293   if(_peers.contains(iodev))
 294     return true;
 295
 296   if(proxyMode() == Client && !_peers.isEmpty()) {
 297     qWarning("SignalProxy: only one peer allowed in client mode!");
 298     return false;
 299   }
 300
 301   if(!iodev->isOpen()) {
 302     qWarning("SignalProxy::addPeer(QIODevice *iodev): iodev needs to be open!");
 303     return false;
 304   }
 305
 306   connect(iodev, SIGNAL(disconnected()), this, SLOT(removePeerBySender()));
 307   connect(iodev, SIGNAL(readyRead()), this, SLOT(dataAvailable()));
 308
 309 #ifdef HAVE_SSL
 310   QSslSocket *sslSocket = qobject_cast<QSslSocket *>(iodev);
 311   if(sslSocket) {
 312     connect(iodev, SIGNAL(encrypted()), this, SLOT(updateSecureState()));
 313   }
 314 #endif
 315
 316   if(!iodev->parent())
 317     iodev->setParent(this);
 318
 319   _peers[iodev] = new IODevicePeer(iodev, iodev->property("UseCompression").toBool());
 320
 321   if(_peers.count() == 1)
 322     emit connected();
 323
 324   updateSecureState();
 325   return true;
 326 }
 327
 328 void SignalProxy::setHeartBeatInterval(int secs) {
 329   if(secs != _heartBeatInterval) {
 330     _heartBeatInterval = secs;
 331     _heartBeatTimer.setInterval(secs * 1000);
 332   }
 333 }
 334
 335 void SignalProxy::setMaxHeartBeatCount(int max) {
 336   if(max < 0)
 337     _heartBeatTimer.stop();
 338   else {
 339     _maxHeartBeatCount = max;
 340     if(!_heartBeatTimer.isActive())
 341       _heartBeatTimer.start();
 342   }
 343 }
 344
 345 bool SignalProxy::addPeer(SignalProxy* proxy) {
 346   if(!proxy)
 347     return false;
 348
 349   if(proxyMode() == proxy->proxyMode()) {
 350     qWarning() << "SignalProxy::addPeer(): adding a SignalProxy as peer requires one proxy to be server and one client!";
 351     return false;
 352   }
 353
 354   if(_peers.contains(proxy)) {
 355     return true;
 356   }
 357
 358   if(proxyMode() == Client && !_peers.isEmpty()) {
 359     qWarning("SignalProxy: only one peer allowed in client mode!");
 360     return false;
 361   }
 362
 363   _peers[proxy] = new SignalProxyPeer(this, proxy);
 364   proxy->addPeer(this);
 365
 366   if(_peers.count() == 1)
 367     emit connected();
 368
 369   updateSecureState();
 370   return true;
 371 }
 372
 373 void SignalProxy::removeAllPeers() {
 374   Q_ASSERT(proxyMode() == Server || _peers.count() <= 1);
 375   // wee need to copy that list since we modify it in the loop
 376   QList<QObject *> peers = _peers.keys();
 377   foreach(QObject *peer, peers) {
 378     removePeer(peer);
 379   }
 380 }
 381
 382 void SignalProxy::removePeer(QObject* dev) {
 383   if(_peers.isEmpty()) {
 384     qWarning() << "SignalProxy::removePeer(): No peers in use!";
 385     return;
 386   }
 387
 388   Q_ASSERT(dev);
 389   if(!_peers.contains(dev)) {
 390     qWarning() << "SignalProxy: unknown Peer" << dev;
 391     return;
 392   }
 393
 394   AbstractPeer *peer = _peers[dev];
 395   _peers.remove(dev);
 396
 397   disconnect(dev, 0, this, 0);
 398   if(peer->type() == AbstractPeer::IODevicePeer)
 399     emit peerRemoved(static_cast<QIODevice *>(dev));
 400
 401   if(peer->type() == AbstractPeer::SignalProxyPeer) {
 402     SignalProxy *proxy = static_cast<SignalProxy *>(dev);
 403     if(proxy->_peers.contains(this))
 404       proxy->removePeer(this);
 405   }
 406
 407   if(dev->parent() == this)
 408     dev->deleteLater();
 409
 410   delete peer;
 411
 412   updateSecureState();
 413
 414   if(_peers.isEmpty())
 415     emit disconnected();
 416 }
 417
 418 void SignalProxy::removePeerBySender() {
 419   removePeer(sender());
 420 }
 421
 422 void SignalProxy::renameObject(const SyncableObject *obj, const QString &newname, const QString &oldname) {
 423   if(proxyMode() == Client)
 424     return;
 425
 426   const QMetaObject *meta = obj->syncMetaObject();
 427   const QByteArray className(meta->className());
 428   objectRenamed(className, newname, oldname);
 429
 430   QVariantList params;
 431   params << "__objectRenamed__" << className << newname << oldname;
 432   dispatchSignal(RpcCall, params);
 433 }
 434
 435 void SignalProxy::objectRenamed(const QByteArray &classname, const QString &newname, const QString &oldname) {
 436   if(_syncSlave.contains(classname) && _syncSlave[classname].contains(oldname) && oldname != newname) {
 437     SyncableObject *obj = _syncSlave[classname][newname] = _syncSlave[classname].take(oldname);
 438     requestInit(obj);
 439   }
 440 }
 441
 442 const QMetaObject *SignalProxy::metaObject(const QObject *obj) {
 443   if(const SyncableObject *syncObject = qobject_cast<const SyncableObject *>(obj))
 444     return syncObject->syncMetaObject();
 445   else
 446     return obj->metaObject();
 447 }
 448
 449 SignalProxy::ExtendedMetaObject *SignalProxy::extendedMetaObject(const QMetaObject *meta) const {
 450   if(_extendedMetaObjects.contains(meta))
 451     return _extendedMetaObjects[meta];
 452   else
 453     return 0;
 454 }
 455
 456 SignalProxy::ExtendedMetaObject *SignalProxy::createExtendedMetaObject(const QMetaObject *meta, bool checkConflicts) {
 457   if(!_extendedMetaObjects.contains(meta)) {
 458     _extendedMetaObjects[meta] = new ExtendedMetaObject(meta, checkConflicts);
 459   }
 460   return _extendedMetaObjects[meta];
 461 }
 462
 463 bool SignalProxy::attachSignal(QObject *sender, const char* signal, const QByteArray& sigName) {
 464   const QMetaObject* meta = sender->metaObject();
 465   QByteArray sig(meta->normalizedSignature(signal).mid(1));
 466   int methodId = meta->indexOfMethod(sig.constData());
 467   if(methodId == -1 || meta->method(methodId).methodType() != QMetaMethod::Signal) {
 468     qWarning() << "SignalProxy::attachSignal(): No such signal" << signal;
 469     return false;
 470   }
 471
 472   createExtendedMetaObject(meta);
 473   _signalRelay->attachSignal(sender, methodId, sigName);
 474
 475   disconnect(sender, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 476   connect(sender, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 477   return true;
 478 }
 479
 480 bool SignalProxy::attachSlot(const QByteArray& sigName, QObject* recv, const char* slot) {
 481   const QMetaObject* meta = recv->metaObject();
 482   int methodId = meta->indexOfMethod(meta->normalizedSignature(slot).mid(1));
 483   if(methodId == -1 || meta->method(methodId).methodType() == QMetaMethod::Method) {
 484     qWarning() << "SignalProxy::attachSlot(): No such slot" << slot;
 485     return false;
 486   }
 487
 488   createExtendedMetaObject(meta);
 489
 490   QByteArray funcName = QMetaObject::normalizedSignature(sigName.constData());
 491   _attachedSlots.insert(funcName, qMakePair(recv, methodId));
 492
 493   disconnect(recv, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 494   connect(recv, SIGNAL(destroyed(QObject *)), this, SLOT(detachObject(QObject *)));
 495   return true;
 496 }
 497
 498 void SignalProxy::synchronize(SyncableObject *obj) {
 499   createExtendedMetaObject(obj, true);
 500
 501   // attaching as slave to receive sync Calls
 502   QByteArray className(obj->syncMetaObject()->className());
 503   _syncSlave[className][obj->objectName()] = obj;
 504
 505   if(proxyMode() == Server) {
 506     obj->setInitialized();
 507     emit objectInitialized(obj);
 508   } else {
 509     if(obj->isInitialized())
 510       emit objectInitialized(obj);
 511     else
 512       requestInit(obj);
 513   }
 514
 515   obj->synchronize(this);
 516 }
 517
 518 void SignalProxy::detachObject(QObject *obj) {
 519   detachSignals(obj);
 520   detachSlots(obj);
 521 }
 522
 523 void SignalProxy::detachSignals(QObject *sender) {
 524   _signalRelay->detachSignal(sender);
 525 }
 526
 527 void SignalProxy::detachSlots(QObject* receiver) {
 528   SlotHash::iterator slotIter = _attachedSlots.begin();
 529   while(slotIter != _attachedSlots.end()) {
 530     if(slotIter.value().first == receiver) {
 531       slotIter = _attachedSlots.erase(slotIter);
 532     } else
 533       slotIter++;
 534   }
 535 }
 536
 537 void SignalProxy::stopSynchronize(SyncableObject *obj) {
 538   // we can't use a className here, since it might be effed up, if we receive the call as a result of a decon
 539   // gladly the objectName() is still valid. So we have only to iterate over the classes not each instance! *sigh*
 540   QHash<QByteArray, ObjectId>::iterator classIter = _syncSlave.begin();
 541   while(classIter != _syncSlave.end()) {
 542     if(classIter->contains(obj->objectName()) && classIter.value()[obj->objectName()] == obj) {
 543       classIter->remove(obj->objectName());
 544       break;
 545     }
 546     classIter++;
 547   }
 548   obj->stopSynchronize(this);
 549 }
 550
 551 void SignalProxy::dispatchSignal(const RequestType &requestType, const QVariantList &params) {
 552   QVariant packedFunc(QVariantList() << (qint16)requestType << params);
 553   PeerHash::iterator peer = _peers.begin();
 554   while(peer != _peers.end()) {
 555     switch((*peer)->type()) {
 556     case AbstractPeer::IODevicePeer:
 557       {
 558         IODevicePeer *ioPeer = static_cast<IODevicePeer *>(*peer);
 559         if(ioPeer->isOpen())
 560           ioPeer->dispatchPackedFunc(packedFunc);
 561         else
 562           QCoreApplication::postEvent(this, new RemovePeerEvent(peer.key()));
 563       }
 564       break;
 565     case AbstractPeer::SignalProxyPeer:
 566       (*peer)->dispatchSignal(requestType, params);
 567       break;
 568     default:
 569       Q_ASSERT(false); // there shouldn't be any peers with wrong / unknown type
 570     }
 571     peer++;
 572   }
 573 }
 574
 575 void SignalProxy::receivePackedFunc(AbstractPeer *sender, const QVariant &packedFunc) {
 576   QVariantList params(packedFunc.toList());
 577
 578   if(params.isEmpty()) {
 579     qWarning() << "SignalProxy::receivePeerSignal(): received incompatible Data:" << packedFunc;
 580     return;
 581   }
 582
 583   RequestType requestType = (RequestType)params.takeFirst().value<int>();
 584   receivePeerSignal(sender, requestType, params);
 585 }
 586
 587 void SignalProxy::receivePeerSignal(AbstractPeer *sender, const RequestType &requestType, const QVariantList &params) {
 588   switch(requestType) {
 589     // list all RequestTypes that shouldnot trigger a heartbeat counter reset here
 590   case HeartBeatReply:
 591     break;
 592   default:
 593     if(sender->type() == AbstractPeer::IODevicePeer) {
 594       IODevicePeer *ioPeer = static_cast<IODevicePeer *>(sender);
 595       ioPeer->sentHeartBeats = 0;
 596     }
 597   }
 598
 599   // qDebug() << "SignalProxy::receivePeerSignal)" << requestType << params;
 600   switch(requestType) {
 601   case RpcCall:
 602     if(params.empty())
 603       qWarning() << "SignalProxy::receivePeerSignal(): received empty RPC-Call";
 604     else
 605       handleSignal(params);
 606       //handleSignal(params.takeFirst().toByteArray(), params);
 607     break;
 608
 609   case Sync:
 610     handleSync(sender, params);
 611     break;
 612
 613   case InitRequest:
 614     handleInitRequest(sender, params);
 615     break;
 616
 617   case InitData:
 618     handleInitData(sender, params);
 619     break;
 620
 621   case HeartBeat:
 622     receiveHeartBeat(sender, params);
 623     break;
 624
 625   case HeartBeatReply:
 626     receiveHeartBeatReply(sender, params);
 627     break;
 628
 629   default:
 630     qWarning() << "SignalProxy::receivePeerSignal(): received undefined CallType" << requestType << params;
 631   }
 632 }
 633
 634 void SignalProxy::receivePeerSignal(SignalProxy *sender, const RequestType &requestType, const QVariantList &params) {
 635   if(!_peers.contains(sender)) {
 636     // 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.
 637     qWarning() << "SignalProxy::receivePeerSignal(): received Signal from unknown Proxy" << reinterpret_cast<void *>(sender);
 638     return;
 639   }
 640   receivePeerSignal(_peers[sender], requestType, params);
 641 }
 642
 643 void SignalProxy::handleSync(AbstractPeer *sender, QVariantList params) {
 644   if(params.count() < 3) {
 645     qWarning() << "received invalid Sync call" << params;
 646     return;
 647   }
 648
 649   QByteArray className = params.takeFirst().toByteArray();
 650   QString objectName = params.takeFirst().toString();
 651   QByteArray slot = params.takeFirst().toByteArray();
 652
 653   if(!_syncSlave.contains(className) || !_syncSlave[className].contains(objectName)) {
 654     qWarning() << QString("no registered receiver for sync call: %1::%2 (objectName=\"%3\"). Params are:").arg(QString(className)).arg(QString(slot)).arg(objectName)
 655                << params;
 656     return;
 657   }
 658
 659   SyncableObject *receiver = _syncSlave[className][objectName];
 660   ExtendedMetaObject *eMeta = extendedMetaObject(receiver);
 661   if(!eMeta->slotMap().contains(slot)) {
 662     qWarning() << QString("no matching slot for sync call: %1::%2 (objectName=\"%3\"). Params are:").arg(QString(className)).arg(QString(slot)).arg(objectName)
 663                << params;
 664     return;
 665   }
 666
 667   int slotId = eMeta->slotMap()[slot];
 668   if(proxyMode() != eMeta->receiverMode(slotId)) {
 669     qWarning("SignalProxy::handleSync(): invokeMethod for \"%s\" failed. Wrong ProxyMode!", eMeta->methodName(slotId).constData());
 670     return;
 671   }
 672
 673   QVariant returnValue((QVariant::Type)eMeta->returnType(slotId));
 674   if(!invokeSlot(receiver, slotId, params, returnValue)) {
 675     qWarning("SignalProxy::handleSync(): invokeMethod for \"%s\" failed ", eMeta->methodName(slotId).constData());
 676     return;
 677   }
 678
 679
 680   if(returnValue.type() != QVariant::Invalid && eMeta->receiveMap().contains(slotId)) {
 681     int receiverId = eMeta->receiveMap()[slotId];
 682     QVariantList returnParams;
 683     returnParams << className
 684                  << objectName
 685                  << eMeta->methodName(receiverId);
 686     //QByteArray(receiver->metaObject()->method(receiverId).signature());
 687     if(eMeta->argTypes(receiverId).count() > 1)
 688       returnParams << params;
 689     returnParams << returnValue;
 690     sender->dispatchSignal(Sync, returnParams);
 691   }
 692
 693   // send emit update signal
 694   invokeSlot(receiver, eMeta->updatedRemotelyId());
 695 }
 696
 697 void SignalProxy::handleInitRequest(AbstractPeer *sender, const QVariantList &params) {
 698   if(params.count() != 2) {
 699     qWarning() << "SignalProxy::handleInitRequest() received initRequest with invalid param Count:"
 700                << params;
 701     return;
 702   }
 703
 704   QByteArray className(params[0].toByteArray());
 705   QString objectName(params[1].toString());
 706
 707   if(!_syncSlave.contains(className)) {
 708     qWarning() << "SignalProxy::handleInitRequest() received initRequest for unregistered Class:"
 709                << className;
 710     return;
 711   }
 712
 713   if(!_syncSlave[className].contains(objectName)) {
 714     qWarning() << "SignalProxy::handleInitRequest() received initRequest for unregistered Object:"
 715                << className << objectName;
 716     return;
 717   }
 718
 719   SyncableObject *obj = _syncSlave[className][objectName];
 720
 721   QVariantList params_;
 722   params_ << className
 723           << objectName
 724           << initData(obj);
 725
 726   sender->dispatchSignal(InitData, params_);
 727 }
 728
 729 void SignalProxy::handleInitData(AbstractPeer *sender, const QVariantList &params) {
 730   Q_UNUSED(sender)
 731   if(params.count() != 3) {
 732     qWarning() << "SignalProxy::handleInitData() received initData with invalid param Count:"
 733                << params;
 734     return;
 735   }
 736
 737   QByteArray className(params[0].toByteArray());
 738   QString objectName(params[1].toString());
 739   QVariantMap propertyMap(params[2].toMap());
 740
 741   if(!_syncSlave.contains(className)) {
 742     qWarning() << "SignalProxy::handleInitData() received initData for unregistered Class:"
 743                << className;
 744     return;
 745   }
 746
 747   if(!_syncSlave[className].contains(objectName)) {
 748     qWarning() << "SignalProxy::handleInitData() received initData for unregistered Object:"
 749                << className << objectName;
 750     return;
 751   }
 752
 753   SyncableObject *obj = _syncSlave[className][objectName];
 754   setInitData(obj, propertyMap);
 755 }
 756
 757 //void SignalProxy::handleSignal(const QByteArray &funcName, const QVariantList &params) {
 758 void SignalProxy::handleSignal(const QVariantList &data) {
 759   QVariantList params = data;
 760   QByteArray funcName = params.takeFirst().toByteArray();
 761
 762   QObject* receiver;
 763   int methodId;
 764   SlotHash::const_iterator slot = _attachedSlots.constFind(funcName);
 765   while(slot != _attachedSlots.constEnd() && slot.key() == funcName) {
 766     receiver = (*slot).first;
 767     methodId = (*slot).second;
 768     if(!invokeSlot(receiver, methodId, params)) {
 769       ExtendedMetaObject *eMeta = extendedMetaObject(receiver);
 770       qWarning("SignalProxy::handleSignal(): invokeMethod for \"%s\" failed ", eMeta->methodName(methodId).constData());
 771     }
 772     slot++;
 773   }
 774 }
 775
 776 bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList &params, QVariant &returnValue) {
 777   ExtendedMetaObject *eMeta = extendedMetaObject(receiver);
 778   const QList<int> args = eMeta->argTypes(methodId);
 779   const int numArgs = params.count() < args.count()
 780     ? params.count()
 781     : args.count();
 782
 783   if(eMeta->minArgCount(methodId) > params.count()) {
 784       qWarning() << "SignalProxy::invokeSlot(): not enough params to invoke" << eMeta->methodName(methodId);
 785       return false;
 786   }
 787
 788   void *_a[] = {0,              // return type...
 789                 0, 0, 0, 0 , 0, // and 10 args - that's the max size qt can handle with signals and slots
 790                 0, 0, 0, 0 , 0};
 791
 792   // check for argument compatibility and build params array
 793   for(int i = 0; i < numArgs; i++) {
 794     if(!params[i].isValid()) {
 795       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());
 796       qWarning() << "                            - make sure all your data types are known by the Qt MetaSystem";
 797       return false;
 798     }
 799     if(args[i] != QMetaType::type(params[i].typeName())) {
 800       qWarning() << "SignalProxy::invokeSlot(): incompatible param types to invoke" << eMeta->methodName(methodId);
 801       return false;
 802     }
 803     _a[i+1] = const_cast<void *>(params[i].constData());
 804   }
 805
 806   if(returnValue.type() != QVariant::Invalid)
 807     _a[0] = const_cast<void *>(returnValue.constData());
 808
 809   Qt::ConnectionType type = QThread::currentThread() == receiver->thread()
 810     ? Qt::DirectConnection
 811     : Qt::QueuedConnection;
 812
 813   if(type == Qt::DirectConnection) {
 814     return receiver->qt_metacall(QMetaObject::InvokeMetaMethod, methodId, _a) < 0;
 815   } else {
 816     qWarning() << "Queued Connections are not implemented yet";
 817     // note to self: qmetaobject.cpp:990 ff
 818     return false;
 819   }
 820
 821 }
 822
 823 bool SignalProxy::invokeSlot(QObject *receiver, int methodId, const QVariantList &params) {
 824   QVariant ret;
 825   return invokeSlot(receiver, methodId, params, ret);
 826 }
 827
 828 void SignalProxy::dataAvailable() {
 829   // yet again. it's a private slot. no need for checks.
 830   QIODevice* ioDev = qobject_cast<QIODevice* >(sender());
 831   Q_ASSERT(_peers.contains(ioDev) && _peers[ioDev]->type() == AbstractPeer::IODevicePeer);
 832   IODevicePeer *peer = static_cast<IODevicePeer *>(_peers[ioDev]);
 833   QVariant var;
 834   while(peer->readData(var))
 835     receivePackedFunc(peer, var);
 836 }
 837
 838 void SignalProxy::writeDataToDevice(QIODevice *dev, const QVariant &item, bool compressed) {
 839   QAbstractSocket* sock  = qobject_cast<QAbstractSocket*>(dev);
 840   if(!dev->isOpen() || (sock && sock->state()!=QAbstractSocket::ConnectedState)) {
 841     qWarning("SignalProxy: Can't call write on a closed device");
 842     return;
 843   }
 844
 845   QByteArray block;
 846   QDataStream out(&block, QIODevice::WriteOnly);
 847   out.setVersion(QDataStream::Qt_4_2);
 848   out << (quint32)0;
 849
 850   if(compressed) {
 851     QByteArray rawItem;
 852     QDataStream itemStream(&rawItem, QIODevice::WriteOnly);
 853
 854     itemStream.setVersion(QDataStream::Qt_4_2);
 855     itemStream << item;
 856
 857     rawItem = qCompress(rawItem);
 858
 859     out << rawItem;
 860   } else {
 861     out << item;
 862   }
 863
 864   out.device()->seek(0);
 865   out << (quint32)(block.size() - sizeof(quint32));
 866
 867   dev->write(block);
 868 }
 869
 870 bool SignalProxy::readDataFromDevice(QIODevice *dev, quint32 &blockSize, QVariant &item, bool compressed) {
 871   if(!dev)
 872     return false;
 873
 874   QDataStream in(dev);
 875   in.setVersion(QDataStream::Qt_4_2);
 876
 877   if(blockSize == 0) {
 878     if(dev->bytesAvailable() < (int)sizeof(quint32)) return false;
 879     in >> blockSize;
 880   }
 881
 882   if(blockSize > 1 << 22) {
 883     disconnectDevice(dev, tr("Peer tried to send package larger than max package size!"));
 884     return false;
 885   }
 886
 887   if(blockSize == 0) {
 888     disconnectDevice(dev, tr("Peer tried to send 0 byte package!"));
 889     return false;
 890   }
 891
 892   if(dev->bytesAvailable() < blockSize)
 893     return false;
 894
 895   blockSize = 0;
 896
 897   if(compressed) {
 898     QByteArray rawItem;
 899     in >> rawItem;
 900
 901     int nbytes = rawItem.size();
 902     if(nbytes <= 4) {
 903       const char *data = rawItem.constData();
 904       if(nbytes < 4 || (data[0]!=0 || data[1]!=0 || data[2]!=0 || data[3]!=0)) {
 905         disconnectDevice(dev, tr("Peer sent corrupted compressed data!"));
 906         return false;
 907       }
 908     }
 909
 910     rawItem = qUncompress(rawItem);
 911
 912     QDataStream itemStream(&rawItem, QIODevice::ReadOnly);
 913     itemStream.setVersion(QDataStream::Qt_4_2);
 914     itemStream >> item;
 915   } else {
 916     in >> item;
 917   }
 918
 919   if(!item.isValid()) {
 920     disconnectDevice(dev, tr("Peer sent corrupt data: unable to load QVariant!"));
 921     return false;
 922   }
 923
 924   return true;
 925 }
 926
 927 void SignalProxy::requestInit(SyncableObject *obj) {
 928   if(proxyMode() == Server || obj->isInitialized())
 929     return;
 930
 931   QVariantList params;
 932   params << obj->syncMetaObject()->className()
 933          << obj->objectName();
 934   dispatchSignal(InitRequest, params);
 935 }
 936
 937 QVariantMap SignalProxy::initData(SyncableObject *obj) const {
 938   return obj->toVariantMap();
 939 }
 940
 941 void SignalProxy::setInitData(SyncableObject *obj, const QVariantMap &properties) {
 942   if(obj->isInitialized())
 943     return;
 944   obj->fromVariantMap(properties);
 945   obj->setInitialized();
 946   emit objectInitialized(obj);
 947   invokeSlot(obj, extendedMetaObject(obj)->updatedRemotelyId());
 948 }
 949
 950 void SignalProxy::sendHeartBeat() {
 951   QVariantList heartBeatParams;
 952   heartBeatParams << QTime::currentTime();
 953   QList<IODevicePeer *> toClose;
 954
 955   PeerHash::iterator peer = _peers.begin();
 956   while(peer != _peers.end()) {
 957     if((*peer)->type() == AbstractPeer::IODevicePeer) {
 958       IODevicePeer *ioPeer = static_cast<IODevicePeer *>(*peer);
 959       ioPeer->dispatchSignal(SignalProxy::HeartBeat, heartBeatParams);
 960       if(ioPeer->sentHeartBeats > 0) {
 961         updateLag(ioPeer, ioPeer->sentHeartBeats * _heartBeatTimer.interval());
 962       }
 963       if(ioPeer->sentHeartBeats >= maxHeartBeatCount())
 964         toClose.append(ioPeer);
 965       else
 966         ioPeer->sentHeartBeats++;
 967     }
 968     ++peer;
 969   }
 970
 971   foreach(IODevicePeer *ioPeer, toClose) {
 972     qWarning() << "SignalProxy: Disconnecting peer:" << ioPeer->address()
 973                << "(didn't receive a heartbeat for over" << ioPeer->sentHeartBeats * _heartBeatTimer.interval() / 1000 << "seconds)";
 974     ioPeer->close();
 975   }
 976 }
 977
 978 void SignalProxy::receiveHeartBeat(AbstractPeer *peer, const QVariantList &params) {
 979   peer->dispatchSignal(SignalProxy::HeartBeatReply, params);
 980 }
 981
 982 void SignalProxy::receiveHeartBeatReply(AbstractPeer *peer, const QVariantList &params) {
 983   if(peer->type() != AbstractPeer::IODevicePeer) {
 984     qWarning() << "SignalProxy::receiveHeartBeatReply: received heart beat from a non IODevicePeer!";
 985     return;
 986   }
 987
 988   IODevicePeer *ioPeer = static_cast<IODevicePeer *>(peer);
 989   ioPeer->sentHeartBeats = 0;
 990
 991   if(params.isEmpty()) {
 992     qWarning() << "SignalProxy: received heart beat reply with less params then sent from:" << ioPeer->address();
 993     return;
 994   }
 995
 996   QTime sendTime = params[0].value<QTime>();
 997   updateLag(ioPeer, sendTime.msecsTo(QTime::currentTime()) / 2);
 998 }
 999
1000 void SignalProxy::customEvent(QEvent *event) {
1001   switch(event->type()) {
1002   case PeerSignal:
1003     {
1004       PeerSignalEvent *e = static_cast<PeerSignalEvent *>(event);
1005       receivePeerSignal(e->sender, e->requestType, e->params);
1006     }
1007     event->accept();
1008     break;
1009   case RemovePeer:
1010     {
1011       RemovePeerEvent *e = static_cast<RemovePeerEvent *>(event);
1012       removePeer(e->peer);
1013     }
1014     event->accept();
1015   default:
1016     return;
1017   }
1018 }
1019
1020 void SignalProxy::sync_call__(const SyncableObject *obj, SignalProxy::ProxyMode modeType, const char *funcname, va_list ap) {
1021   // qDebug() << obj << modeType << "(" << _proxyMode << ")" << funcname;
1022   if(modeType != _proxyMode)
1023     return;
1024
1025   ExtendedMetaObject *eMeta = extendedMetaObject(obj);
1026
1027   QVariantList params;
1028   params << eMeta->metaObject()->className()
1029          << obj->objectName()
1030          << QByteArray(funcname);
1031
1032   const QList<int> &argTypes = eMeta->argTypes(eMeta->methodId(QByteArray(funcname)));
1033
1034   for(int i = 0; i < argTypes.size(); i++) {
1035     if(argTypes[i] == 0) {
1036       qWarning() << Q_FUNC_INFO << "received invalid data for argument number" << i << "of signal" << QString("%1::%2").arg(eMeta->metaObject()->className()).arg(funcname);
1037       qWarning() << "        - make sure all your data types are known by the Qt MetaSystem";
1038       return;
1039     }
1040     params << QVariant(argTypes[i], va_arg(ap, void *));
1041   }
1042
1043   dispatchSignal(Sync, params);
1044 }
1045
1046
1047
1048 void SignalProxy::disconnectDevice(QIODevice *dev, const QString &reason) {
1049   if(!reason.isEmpty())
1050     qWarning() << qPrintable(reason);
1051   QAbstractSocket *sock  = qobject_cast<QAbstractSocket *>(dev);
1052   if(sock)
1053     qWarning() << qPrintable(tr("Disconnecting")) << qPrintable(sock->peerAddress().toString());
1054   dev->close();
1055 }
1056
1057 void SignalProxy::updateLag(IODevicePeer *peer, int lag) {
1058   peer->lag = lag;
1059   if(proxyMode() == Client) {
1060     emit lagUpdated(lag);
1061   }
1062 }
1063
1064 void SignalProxy::dumpProxyStats() {
1065   QString mode;
1066   if(proxyMode() == Server)
1067     mode = "Server";
1068   else
1069     mode = "Client";
1070
1071   int slaveCount = 0;
1072   foreach(ObjectId oid, _syncSlave.values())
1073     slaveCount += oid.count();
1074
1075   qDebug() << this;
1076   qDebug() << "              Proxy Mode:" << mode;
1077   qDebug() << "          attached Slots:" << _attachedSlots.count();
1078   qDebug() << " number of synced Slaves:" << slaveCount;
1079   qDebug() << "number of Classes cached:" << _extendedMetaObjects.count();
1080 }
1081
1082 void SignalProxy::updateSecureState() {
1083   bool wasSecure = _secure;
1084
1085   _secure = !_peers.isEmpty();
1086   PeerHash::const_iterator peerIter;
1087   for(peerIter = _peers.constBegin(); peerIter != _peers.constEnd(); peerIter++) {
1088     _secure &= (*peerIter)->isSecure();
1089   }
1090
1091   if(wasSecure != _secure)
1092     emit secureStateChanged(_secure);
1093 }
1094
1095
1096 // ==================================================
1097 //  ExtendedMetaObject
1098 // ==================================================
1099 SignalProxy::ExtendedMetaObject::ExtendedMetaObject(const QMetaObject *meta, bool checkConflicts)
1100   : _meta(meta),
1101     _updatedRemotelyId(_meta->indexOfSignal("updatedRemotely()"))
1102 {
1103   for(int i = 0; i < _meta->methodCount(); i++) {
1104     if(_meta->method(i).methodType() != QMetaMethod::Slot)
1105       continue;
1106
1107     if(QByteArray(_meta->method(i).signature()).contains('*'))
1108       continue; // skip methods with ptr params
1109
1110     QByteArray method = methodName(_meta->method(i));
1111     if(method.startsWith("init"))
1112       continue; // skip initializers
1113
1114     if(_methodIds.contains(method)) {
1115       /* funny... moc creates for methods containing default parameters multiple metaMethod with separate methodIds.
1116          we don't care... we just need the full fledged version
1117        */
1118       const QMetaMethod &current = _meta->method(_methodIds[method]);
1119       const QMetaMethod &candidate = _meta->method(i);
1120       if(current.parameterTypes().count() > candidate.parameterTypes().count()) {
1121         int minCount = candidate.parameterTypes().count();
1122         QList<QByteArray> commonParams = current.parameterTypes().mid(0, minCount);
1123         if(commonParams == candidate.parameterTypes())
1124           continue; // we already got the full featured version
1125       } else {
1126         int minCount = current.parameterTypes().count();
1127         QList<QByteArray> commonParams = candidate.parameterTypes().mid(0, minCount);
1128         if(commonParams == current.parameterTypes()) {
1129           _methodIds[method] = i; // use the new one
1130           continue;
1131         }
1132       }
1133       if(checkConflicts) {
1134         qWarning() << "class" << meta->className() << "contains overloaded methods which is currently not supported!";
1135         qWarning() << " - " << _meta->method(i).signature() << "conflicts with" << _meta->method(_methodIds[method]).signature();
1136       }
1137       continue;
1138     }
1139     _methodIds[method] = i;
1140   }
1141 }
1142
1143 const SignalProxy::ExtendedMetaObject::MethodDescriptor &SignalProxy::ExtendedMetaObject::methodDescriptor(int methodId) {
1144   if(!_methods.contains(methodId)) {
1145     _methods[methodId] = MethodDescriptor(_meta->method(methodId));
1146   }
1147   return _methods[methodId];
1148 }
1149
1150 const QHash<int, int> &SignalProxy::ExtendedMetaObject::receiveMap() {
1151   if(_receiveMap.isEmpty()) {
1152     QHash<int, int> receiveMap;
1153
1154     QMetaMethod requestSlot;
1155     QByteArray returnTypeName;
1156     QByteArray signature;
1157     QByteArray methodName;
1158     QByteArray params;
1159     int paramsPos;
1160     int receiverId;
1161     const int methodCount = _meta->methodCount();
1162     for(int i = 0; i < methodCount; i++) {
1163       requestSlot = _meta->method(i);
1164       if(requestSlot.methodType() != QMetaMethod::Slot)
1165         continue;
1166
1167       returnTypeName = requestSlot.typeName();
1168       if(QMetaType::Void == (QMetaType::Type)returnType(i))
1169         continue;
1170
1171       signature = QByteArray(requestSlot.signature());
1172       if(!signature.startsWith("request"))
1173         continue;
1174
1175       paramsPos = signature.indexOf('(');
1176       if(paramsPos == -1)
1177         continue;
1178
1179       methodName = signature.left(paramsPos);
1180       params = signature.mid(paramsPos);
1181
1182       methodName = methodName.replace("request", "receive");
1183       params = params.left(params.count() - 1) + ", " + returnTypeName + ")";
1184
1185       signature = QMetaObject::normalizedSignature(methodName + params);
1186       receiverId = _meta->indexOfSlot(signature);
1187
1188       if(receiverId == -1) {
1189         signature = QMetaObject::normalizedSignature(methodName + "(" + returnTypeName + ")");
1190         receiverId = _meta->indexOfSlot(signature);
1191       }
1192
1193       if(receiverId != -1) {
1194         receiveMap[i] = receiverId;
1195       }
1196     }
1197     _receiveMap = receiveMap;
1198   }
1199   return _receiveMap;
1200 }
1201
1202 QByteArray SignalProxy::ExtendedMetaObject::methodName(const QMetaMethod &method) {
1203   QByteArray sig(method.signature());
1204   return sig.left(sig.indexOf("("));
1205 }
1206
1207 QString SignalProxy::ExtendedMetaObject::methodBaseName(const QMetaMethod &method) {
1208   QString methodname = QString(method.signature()).section("(", 0, 0);
1209
1210   // determine where we have to chop:
1211   int upperCharPos;
1212   if(method.methodType() == QMetaMethod::Slot) {
1213     // we take evertyhing from the first uppercase char if it's slot
1214     upperCharPos = methodname.indexOf(QRegExp("[A-Z]"));
1215     if(upperCharPos == -1)
1216       return QString();
1217     methodname = methodname.mid(upperCharPos);
1218   } else {
1219     // and if it's a signal we discard everything from the last uppercase char
1220     upperCharPos = methodname.lastIndexOf(QRegExp("[A-Z]"));
1221     if(upperCharPos == -1)
1222       return QString();
1223     methodname = methodname.left(upperCharPos);
1224   }
1225
1226   methodname[0] = methodname[0].toUpper();
1227
1228   return methodname;
1229 }
1230
1231 SignalProxy::ExtendedMetaObject::MethodDescriptor::MethodDescriptor(const QMetaMethod &method)
1232   : _methodName(SignalProxy::ExtendedMetaObject::methodName(method)),
1233     _returnType(QMetaType::type(method.typeName()))
1234 {
1235   // determine argTypes
1236   QList<QByteArray> paramTypes = method.parameterTypes();
1237   QList<int> argTypes;
1238   for(int i = 0; i < paramTypes.count(); i++) {
1239     argTypes.append(QMetaType::type(paramTypes[i]));
1240   }
1241   _argTypes = argTypes;
1242
1243   // determine minArgCount
1244   QString signature(method.signature());
1245   _minArgCount = method.parameterTypes().count() - signature.count("=");
1246
1247   _receiverMode = (_methodName.startsWith("request"))
1248     ? SignalProxy::Server
1249     : SignalProxy::Client;
1250 }
1251