Implement UI and pvsDaemon components for outgoing Multicast Transfer

7 files changed, 360 insertions, 6 deletions

diff --git a/src/net/mcast/McastConfiguration.cpp b/src/net/mcast/McastConfiguration.cpp
index 1a1c0a8..489eb3f 100644
--- a/src/net/mcast/McastConfiguration.cpp
+++ b/src/net/mcast/McastConfiguration.cpp
@@ -5,5 +5,44 @@
  *      Author: brs
  */
 
+#include <QSettings>
+
 #include "McastConfiguration.h"
 
+void McastConfiguration::loadFrom(QSettings* _settings, char const* group)
+{
+	if (group)
+		_settings->beginGroup(group);
+
+	_multicastAddress = _settings->value("groupAddress", DEFAULT_MULTICAST_ADDRESS).toString();
+	_multicastInterface = _settings->value("interface", DEFAULT_MULTICAST_INTERFACE).toString();
+	_multicastMTU = _settings->value("mtu", DEFAULT_MULTICAST_MTU).value<quint16>();
+	_multicastRate = _settings->value("rate", DEFAULT_MULTICAST_RATE).value<quint32>();
+	_multicastUseUDP = _settings->value("use-udp", DEFAULT_MULTICAST_USEUDP).toBool();
+	_multicastWinSize = _settings->value("winsize", DEFAULT_MULTICAST_WSIZ).value<quint16>();
+	_multicastUDPPortBase = _settings->value("portbase", DEFAULT_MULTICAST_UDPPORT).value<quint16>();
+	_multicastDPort = _settings->value("dport", DEFAULT_MULTICAST_DPORT).value<quint16>();
+	_multicastSPort = _settings->value("sport", DEFAULT_MULTICAST_SPORT).value<quint16>();
+
+	if (group)
+		_settings->endGroup();
+}
+
+void McastConfiguration::writeTo(QSettings* _settings, char const* group) const
+{
+	if (group)
+		_settings->beginGroup(group);
+
+	_settings->setValue("groupAddress", _multicastAddress);
+	_settings->setValue("interface", _multicastInterface);
+	_settings->setValue("mtu", _multicastMTU);
+	_settings->setValue("rate", _multicastRate);
+	_settings->setValue("use-udp", _multicastUseUDP);
+	_settings->setValue("winsize", _multicastWinSize);
+	_settings->setValue("portbase", _multicastUDPPortBase);
+	_settings->setValue("dport", _multicastDPort);
+	_settings->setValue("sport", _multicastSPort);
+
+	if (group)
+		_settings->endGroup();
+}
diff --git a/src/net/mcast/McastConfiguration.h b/src/net/mcast/McastConfiguration.h
index 2e64678..b010f60 100644
--- a/src/net/mcast/McastConfiguration.h
+++ b/src/net/mcast/McastConfiguration.h
@@ -23,6 +23,8 @@
 
 #include "McastConstants.h"
 
+class QSettings;
+
 class McastConfiguration: public QObject
 {
 	Q_OBJECT
@@ -164,6 +166,9 @@ public:
 		emit changed();
 	}
 
+	void loadFrom(QSettings* settings, char const* group = 0);
+	void writeTo(QSettings* settings, char const* group = 0) const;
+
 signals:
 	void changed();
 
diff --git a/src/net/mcast/McastPGMSocket.cpp b/src/net/mcast/McastPGMSocket.cpp
index 105bea9..7952f00 100644
--- a/src/net/mcast/McastPGMSocket.cpp
+++ b/src/net/mcast/McastPGMSocket.cpp
@@ -567,6 +567,11 @@ void McastPGMSocket::sendPacket(QByteArray const& bytes)
 
 void McastPGMSocket::finish()
 {
+    if (_finished)
+    {
+        return;
+    }
+
 	qDebug() << "finish()";
 
 	Q_FOREACH(QSocketNotifier* notif, _priv->_notifs)
@@ -582,8 +587,11 @@ void McastPGMSocket::finish()
 		_priv->send_notif = 0;
 	}
 
-	pgm_close(_priv->socket, 1);
-	_priv->socket = 0;
+    if (_priv->socket)
+    {
+        pgm_close(_priv->socket, 1);
+        _priv->socket = 0;
+    }
 
 	_finished = true;
 
diff --git a/src/net/mcast/McastSender.cpp b/src/net/mcast/McastSender.cpp
index 322b751..e25ec86 100644
--- a/src/net/mcast/McastSender.cpp
+++ b/src/net/mcast/McastSender.cpp
@@ -111,6 +111,6 @@ void McastSender::close()
 
 void McastSender::socketFinished()
 {
-	delete _socket;
+	_socket->deleteLater();
 	emit finished();
 }
diff --git a/src/net/pvsOutgoingMulticastTransfer.cpp b/src/net/pvsOutgoingMulticastTransfer.cpp
new file mode 100644
index 0000000..2f24d49
--- /dev/null
+++ b/src/net/pvsOutgoingMulticastTransfer.cpp
@@ -0,0 +1,209 @@
+/*
+# Copyright (c) 2009 - OpenSLX Project, Computer Center University of Freiburg
+#
+# This program is free software distributed under the GPL version 2.
+# See http://openslx.org/COPYING
+#
+# If you have any feedback please consult http://openslx.org/feedback and
+# send your suggestions, praise, or complaints to feedback@openslx.org
+#
+# General information about OpenSLX can be found at http://openslx.org/
+# -----------------------------------------------------------------------------
+# src/net/pvsOutgoingMulticastTransfer.cpp
+#    - wrap McastSender functionality in PVS daemon
+# -----------------------------------------------------------------------------
+*/
+
+#include "pvsOutgoingMulticastTransfer.h"
+
+#include <QDataStream>
+#include <QFile>
+#include <QFileInfo>
+#include <QHostInfo>
+#include <QSettings>
+#include <QTimer>
+
+#include <src/net/pvsMsg.h>
+#include <src/net/mcast/McastConfiguration.h>
+#include <src/net/mcast/McastPGMSocket.h>
+#include <src/net/mcast/McastSender.h>
+#include <src/util/consoleLogger.h>
+
+PVSOutgoingMulticastTransfer::PVSOutgoingMulticastTransfer(QString senderName, quint64 id, QString filename, QObject* parent) :
+    QObject(parent),
+    _file(0),
+    _progress(0),
+    _config(0),
+    _socket(0),
+    _progressTimer(0),
+    _prepareTimer(0),
+    _senderName(senderName),
+    _id(id),
+    _error(false)
+{
+    QFileInfo finfo(filename);
+    if(!finfo.exists())
+    {
+        error("File does not exist");
+    }
+
+    if(!finfo.isReadable())
+    {
+        error("File is not readable");
+    }
+
+    _file = new QFile(filename, this);
+    _length = _file->size();
+}
+
+PVSOutgoingMulticastTransfer::~PVSOutgoingMulticastTransfer()
+{
+    if(_file)
+        delete _file;
+    if(_config)
+        delete _config;
+}
+
+void PVSOutgoingMulticastTransfer::prepare()
+{
+    if (_socket && !_socketInacceptable)
+    {
+        _prepareTimer->stop();
+        delete _prepareTimer;
+        _prepareTimer = 0;
+
+        QTimer::singleShot(0, this, SLOT(doStart()));
+        return;
+    }
+    else if (_socket)
+    {
+        delete _socket;
+        _socket = 0;
+    }
+
+    QSettings settings;
+    quint16 portbase = settings.value("multicast/port-base", "6966").value<quint16>();
+    quint16 portlimit = settings.value("multicast/port-limit", "7966").value<quint16>();
+
+    int tries_remaining = 5;
+    while(tries_remaining > 0)
+    {
+        quint16 port = portbase + (qrand() % (portlimit - portbase + 1));
+
+        if (!_config)
+        {
+            _config = new McastConfiguration();
+        }
+        _config->loadFrom(&settings, "multicast");
+        _config->multicastUDPPortBase(port);
+
+        _socket = new McastPGMSocket(this);
+        if(_socket->open(_config, McastPGMSocket::PSOCK_WRITE))
+        {
+            break;
+        }
+        else
+        {
+            delete _socket;
+            _socket = 0;
+        }
+    }
+
+    if (!_socket)
+    {
+        emit failed(_id, "Could not open socket");
+        delete _prepareTimer;
+        _prepareTimer = 0;
+        return;
+    }
+    else
+    {
+        _socketInacceptable = false;
+        // announce the transfer:
+        QFileInfo info(*_file);
+        QString message = QString("%1:%2:%3:%4:%5").arg(_senderName).arg(_id).arg(info.baseName()).arg(info.size()).arg(_config->multicastUDPPortBase());
+        PVSMsg msg(PVSCOMMAND, "MCASTFTANNOUNCE", message);
+        emit announce(msg);
+        _prepareTimer->start(5000);
+    }
+}
+
+void PVSOutgoingMulticastTransfer::doStart()
+{
+    ConsoleLog writeLine(QString("Starting multicast transfer %1").arg(_id));
+
+    _file->open(QIODevice::ReadOnly);
+
+    _sender = new McastSender(_file, _config, this);
+    connect(_sender, SIGNAL(finished()), SLOT(senderFinished()));
+    connect(_sender, SIGNAL(progress(quint64)), SLOT(senderProgress(quint64)));
+    // connect(_sender, SIGNAL(allSent()), SIGNAL(allSent()));
+    _socket->setParent(_sender);
+    _sender->start(_socket);
+
+    emit started(_id);
+
+    _progressTimer = new QTimer(this);
+    connect(_progressTimer, SIGNAL(timeout()), SLOT(reportTimeout()));
+    _progressTimer->setInterval(333);
+    _progressTimer->start();
+}
+
+void PVSOutgoingMulticastTransfer::senderProgress(quint64 bytes)
+
+{
+    _progress = bytes;
+}
+
+void PVSOutgoingMulticastTransfer::senderFinished()
+{
+    if(_progressTimer)
+    {
+        _progressTimer->stop();
+        delete _progressTimer;
+        _progressTimer = 0;
+    }
+    emit finished(_id);
+    _sender->close();
+    delete _sender;
+}
+
+void PVSOutgoingMulticastTransfer::reportTimeout()
+{
+    emit progress(_id, _progress, _length);
+}
+
+void PVSOutgoingMulticastTransfer::retry()
+{
+    bool first = !_socketInacceptable;
+    _socketInacceptable = true;
+
+    if(first)
+    {
+        _prepareTimer->setInterval(1000);
+    }
+}
+
+void PVSOutgoingMulticastTransfer::start()
+{
+    if (!_prepareTimer)
+    {
+        _prepareTimer = new QTimer(this);
+        _prepareTimer->setSingleShot(true);
+        connect(_prepareTimer, SIGNAL(timeout()), SLOT(prepare()));
+    }
+    QTimer::singleShot(0, this, SLOT(prepare()));
+}
+
+void PVSOutgoingMulticastTransfer::abort()
+{
+    if (_sender)
+        _sender->close();
+}
+
+void PVSOutgoingMulticastTransfer::error(QString const& reason)
+{
+    qCritical() << "Could not create an outgoing mcast transfer: " << reason;
+    _error = true;
+    _reason = reason;
+}
diff --git a/src/net/pvsOutgoingMulticastTransfer.h b/src/net/pvsOutgoingMulticastTransfer.h
new file mode 100644
index 0000000..5fd6a3d
--- /dev/null
+++ b/src/net/pvsOutgoingMulticastTransfer.h
@@ -0,0 +1,92 @@
+/*
+# Copyright (c) 2009 - OpenSLX Project, Computer Center University of Freiburg
+#
+# This program is free software distributed under the GPL version 2.
+# See http://openslx.org/COPYING
+#
+# If you have any feedback please consult http://openslx.org/feedback and
+# send your suggestions, praise, or complaints to feedback@openslx.org
+#
+# General information about OpenSLX can be found at http://openslx.org/
+# -----------------------------------------------------------------------------
+# src/net/pvsOutgoingMulticastTransfer.h
+#    - wrap McastSender functionality in PVS daemon
+# -----------------------------------------------------------------------------
+*/
+
+#ifndef PVSOUTGOINGMULTICASTTRANSFER_H_
+#define PVSOUTGOINGMULTICASTTRANSFER_H_
+
+#include <QtGlobal>
+#include <QObject>
+#include <QString>
+
+#include <src/net/pvsMsg.h>
+
+class QFile;
+class QTimer;
+class McastConfiguration;
+class McastPGMSocket;
+class McastSender;
+
+class PVSOutgoingMulticastTransfer : public QObject
+{
+    Q_OBJECT
+public:
+    PVSOutgoingMulticastTransfer(QString senderName, quint64 id, QString filename, QObject* parent = 0);
+    virtual ~PVSOutgoingMulticastTransfer();
+
+    quint64 id() const
+    {
+        return _id;
+    }
+
+    bool isError() const
+    {
+        return _error;
+    }
+
+    QString reason() const
+    {
+        return _reason;
+    }
+
+signals:
+    void started(qulonglong id);
+    void progress(qulonglong id, qulonglong bytes, qulonglong of);
+    void allSent(qulonglong id);
+    void finished(qulonglong id);
+    void failed(qulonglong id, QString const reason);
+    void announce(PVSMsg announcement);
+
+private slots:
+    void senderProgress(quint64 bytes);
+    void senderFinished();
+    void reportTimeout();
+    void doStart();
+    void prepare();
+
+public slots:
+    void start();
+    void abort();
+    void retry();
+
+private:
+    QFile* _file;
+    quint64 _length;
+    quint64 _progress;
+    McastConfiguration* _config;
+    McastPGMSocket* _socket;
+    McastSender* _sender;
+    QTimer* _progressTimer;
+    QTimer* _prepareTimer;
+    QString _senderName;
+    quint64 _id;
+    bool _error;
+    QString _reason;
+    bool _socketInacceptable;
+
+    void error(QString const& reason);
+};
+
+#endif /* PVSOUTGOINGMULTICASTTRANSFER_H_ */
diff --git a/src/net/pvsServerConnection.h b/src/net/pvsServerConnection.h
index 0669d88..c6ef015 100644
--- a/src/net/pvsServerConnection.h
+++ b/src/net/pvsServerConnection.h
@@ -11,9 +11,9 @@
 #define _PVSSERVERCONNECTION_H_
 
 #include "src/util/dispatcher.h"
+#include "src/net/pvsMsg.h"
 #include <QtNetwork/QSslSocket>
 
-class PVSMsg;
 class PVS;
 class PVSDiscoveredServer;
 
@@ -30,8 +30,6 @@ public:
         return _socket != NULL && _socket->state() == QAbstractSocket::ConnectedState;
     }
 
-    void sendMessage(PVSMsg newMessage);
-
     void ping();
     QString getServerName();
 
@@ -63,6 +61,9 @@ public:
         _commandDispatcher.removeListener(ident, who, func);
     };
 
+public Q_SLOTS:
+	void sendMessage(PVSMsg newMessage);
+
 protected:
     void timerEvent(QTimerEvent *event);