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