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