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