+ QVariantList heartBeatParams;
+ heartBeatParams << QTime::currentTime();
+ QList<IODevicePeer *> toClose;
+
+ PeerHash::iterator peer = _peers.begin();
+ while(peer != _peers.end()) {
+ if((*peer)->type() == AbstractPeer::IODevicePeer) {
+ IODevicePeer *ioPeer = static_cast<IODevicePeer *>(*peer);
+ ioPeer->dispatchSignal(SignalProxy::HeartBeat, heartBeatParams);
+ if(ioPeer->sentHeartBeats > 0) {
+ updateLag(ioPeer, ioPeer->sentHeartBeats * _heartBeatTimer.interval());
+ }
+ if(maxHeartBeatCount() >= 0 && ioPeer->sentHeartBeats >= maxHeartBeatCount())
+ toClose.append(ioPeer);
+ else
+ ioPeer->sentHeartBeats++;
+ }
+ ++peer;
+ }
+
+ foreach(IODevicePeer *ioPeer, toClose) {
+ qWarning() << "SignalProxy: Disconnecting peer:" << ioPeer->address()
+ << "(didn't receive a heartbeat for over" << ioPeer->sentHeartBeats * _heartBeatTimer.interval() / 1000 << "seconds)";
+ ioPeer->close();
+ }
+}
+
+void SignalProxy::receiveHeartBeat(AbstractPeer *peer, const QVariantList ¶ms) {
+ peer->dispatchSignal(SignalProxy::HeartBeatReply, params);
+}
+
+void SignalProxy::receiveHeartBeatReply(AbstractPeer *peer, const QVariantList ¶ms) {
+ if(peer->type() != AbstractPeer::IODevicePeer) {
+ qWarning() << "SignalProxy::receiveHeartBeatReply: received heart beat from a non IODevicePeer!";
+ return;
+ }
+
+ IODevicePeer *ioPeer = static_cast<IODevicePeer *>(peer);
+ ioPeer->sentHeartBeats = 0;
+
+ if(params.isEmpty()) {
+ qWarning() << "SignalProxy: received heart beat reply with less params then sent from:" << ioPeer->address();
+ return;
+ }
+
+ QTime sendTime = params[0].value<QTime>();
+ updateLag(ioPeer, sendTime.msecsTo(QTime::currentTime()) / 2);
+}
+
+void SignalProxy::customEvent(QEvent *event) {
+ switch(+event->type()) {
+ case PeerSignal:
+ {
+ PeerSignalEvent *e = static_cast<PeerSignalEvent *>(event);
+ receivePeerSignal(e->sender, e->requestType, e->params);
+ }
+ event->accept();
+ break;
+ case RemovePeer:
+ {
+ RemovePeerEvent *e = static_cast<RemovePeerEvent *>(event);
+ removePeer(e->peer);
+ }
+ event->accept();
+ default:
+ return;
+ }
+}
+
+void SignalProxy::sync_call__(const SyncableObject *obj, SignalProxy::ProxyMode modeType, const char *funcname, va_list ap) {
+ // qDebug() << obj << modeType << "(" << _proxyMode << ")" << funcname;
+ if(modeType != _proxyMode)
+ return;
+
+ ExtendedMetaObject *eMeta = extendedMetaObject(obj);
+
+ QVariantList params;
+ params << eMeta->metaObject()->className()
+ << obj->objectName()
+ << QByteArray(funcname);
+
+ const QList<int> &argTypes = eMeta->argTypes(eMeta->methodId(QByteArray(funcname)));
+
+ for(int i = 0; i < argTypes.size(); i++) {
+ if(argTypes[i] == 0) {
+ qWarning() << Q_FUNC_INFO << "received invalid data for argument number" << i << "of signal" << QString("%1::%2").arg(eMeta->metaObject()->className()).arg(funcname);
+ qWarning() << " - make sure all your data types are known by the Qt MetaSystem";
+ return;
+ }
+ params << QVariant(argTypes[i], va_arg(ap, void *));
+ }
+
+ dispatchSignal(Sync, params);
+}
+
+
+
+void SignalProxy::disconnectDevice(QIODevice *dev, const QString &reason) {
+ if(!reason.isEmpty())
+ qWarning() << qPrintable(reason);
+ QAbstractSocket *sock = qobject_cast<QAbstractSocket *>(dev);
+ if(sock)
+ qWarning() << qPrintable(tr("Disconnecting")) << qPrintable(sock->peerAddress().toString());
+ dev->close();
+}
+
+void SignalProxy::updateLag(IODevicePeer *peer, int lag) {
+ peer->lag = lag;
+ if(proxyMode() == Client) {
+ emit lagUpdated(lag);
+ }