<commit_before>#include "networkmanager.h"
#include "../networkhelper.h"
#include "../log/logger.h"
#include "../scheduler/scheduler.h"
#include "../task/taskexecutor.h"
#include "../settings.h"
#include "../timing/immediatetiming.h"
#include "../timing/periodictiming.h"

#include "tcpsocket.h"
#include "udpsocket.h"

#include <QDataStream>
#include <QJsonDocument>
#include <QJsonObject>
#include <QPointer>
#include <QTcpServer>
#include <QTcpSocket>
#include <QUdpSocket>
#include <QStringList>
#include <QHash>
#include <QMutex>
#include <QReadWriteLock>
#include <QTimer>
#include <QDebug>
#include <QNetworkConfigurationManager>

LOGGER(NetworkManager);

class PeerRequest
{
public:
    QVariant toVariant()
    {
        QVariantMap map;
        map.insert("type", "peer_request");
        map.insert("measurementDefinition", measurementDefinition);
        map.insert("taskId", taskId);
        map.insert("measurementUuid", measurementUuid);
        map.insert("measurement", measurement);
        map.insert("peer", peer);
        map.insert("port", port);
        map.insert("protocol", protocol);
        return map;
    }

    static PeerRequest fromVariant(const QVariant& variant)
    {
        QVariantMap map = variant.toMap();

        PeerRequest request;
        request.measurementDefinition = map.value("measurementDefinition");
        request.taskId = map.value("taskId").toUuid();
        request.measurementUuid = map.value("measurementUuid").toUuid();
        request.measurement = map.value("measurement").toString();
        request.peer = map.value("peer").toString();
        request.port = map.value("port").toUInt();
        request.protocol = (NetworkManager::SocketType)map.value("protocol").toInt();
        return request;
    }

    QVariant measurementDefinition;
    QUuid taskId;
    QUuid measurementUuid;
    QString measurement;
    QString peer;
    quint16 port;
    NetworkManager::SocketType protocol;
};

class NetworkManager::Private : public QObject
{
    Q_OBJECT

public:
    Private(NetworkManager* q)
        : q(q)
    {
        connect(&timer, SIGNAL(timeout()), this, SLOT(timeout()));
    }

    NetworkManager* q;

    // Properties
    QMutex mutex;
    QHash<QString, QObject*> objectHash;
    QHash<QString, QAbstractSocket*> socketHash;

    QTimer timer;
    QPointer<QUdpSocket> socket;
    QPointer<Scheduler> scheduler;
    QPointer<Settings> settings;

    QSet<QUuid> handledMeasureUuids;

    quint16 localPort;

    QNetworkConfigurationManager ncm;

    // Functions
    QAbstractSocket* createSocket(NetworkManager::SocketType socketType);

    void updateSocket();
    void updateTimer();
    void processDatagram(const QByteArray& datagram, const QHostAddress& host, quint16 port);

public slots:
    void socketDestroyed(QObject* obj);
    void responseChanged();
    void timeout();
    void onDatagramReady();
};

QAbstractSocket *NetworkManager::Private::createSocket(NetworkManager::SocketType socketType)
{
    QAbstractSocket* socket = NULL;

    switch (socketType)
    {
    case TcpSocket:
        socket = new ::TcpSocket;
        break;

    case UdpSocket:
        socket = new ::UdpSocket;
        break;

    case UtpSocket:
        //socket = new UtpSocket;
        break;

    default:
        break;
    }

    if (socket)
    {
        connect(socket, SIGNAL(destroyed(QObject*)), this, SLOT(socketDestroyed(QObject*)));
    }

    return socket;
}

void NetworkManager::Private::socketDestroyed(QObject *obj)
{
    QString hostname = objectHash.key(obj);
    if (hostname.isEmpty())
    {
        return;
    }

    objectHash.remove(hostname);
    socketHash.remove(hostname);
}

void NetworkManager::Private::updateSocket()
{
    if (!socket.isNull())
    {
        delete socket.data();
        socket.clear();
    }

    /*
    QString keepaliveAddress = settings->config()->keepaliveAddress();
    RemoteHost remote = NetworkHelper::remoteHost(keepaliveAddress);
    if (!remote.isValid())
    {
        return;
    }
    */

    localPort = remote.port;

    socket = qobject_cast<QUdpSocket*>( q->createConnection(NetworkManager::UdpSocket) );
    socket->setParent(this);
    connect(socket.data(), SIGNAL(readyRead()), this, SLOT(onDatagramReady()));
    if (!socket->bind(remote.port))
    {
        LOG_ERROR(QString("Unable to bind port %1: %2").arg(remote.port).arg(socket->errorString()));
    }
}

void NetworkManager::Private::updateTimer()
{
    TimingPtr timing = settings->config()->keepaliveSchedule();
    if (!timing)
    {
        return;
    }

    QSharedPointer<PeriodicTiming> periodicTiming = timing.dynamicCast<PeriodicTiming>();
    Q_ASSERT(periodicTiming);

    int interval = periodicTiming->period();
    if (interval < 1000)
    {
        LOG_INFO("Keepalive interval < 1 sec will not be accepted.");<commit_msg>Hardcoded ports<commit_after>#include "networkmanager.h"
#include "../networkhelper.h"
#include "../log/logger.h"
#include "../scheduler/scheduler.h"
#include "../task/taskexecutor.h"
#include "../settings.h"
#include "../timing/immediatetiming.h"
#include "../timing/periodictiming.h"

#include "tcpsocket.h"
#include "udpsocket.h"

#include <QDataStream>
#include <QJsonDocument>
#include <QJsonObject>
#include <QPointer>
#include <QTcpServer>
#include <QTcpSocket>
#include <QUdpSocket>
#include <QStringList>
#include <QHash>
#include <QMutex>
#include <QReadWriteLock>
#include <QTimer>
#include <QDebug>
#include <QNetworkConfigurationManager>

LOGGER(NetworkManager);

class PeerRequest
{
public:
    QVariant toVariant()
    {
        QVariantMap map;
        map.insert("type", "peer_request");
        map.insert("measurementDefinition", measurementDefinition);
        map.insert("taskId", taskId);
        map.insert("measurementUuid", measurementUuid);
        map.insert("measurement", measurement);
        map.insert("peer", peer);
        map.insert("port", port);
        map.insert("protocol", protocol);
        return map;
    }

    static PeerRequest fromVariant(const QVariant& variant)
    {
        QVariantMap map = variant.toMap();

        PeerRequest request;
        request.measurementDefinition = map.value("measurementDefinition");
        request.taskId = map.value("taskId").toUuid();
        request.measurementUuid = map.value("measurementUuid").toUuid();
        request.measurement = map.value("measurement").toString();
        request.peer = map.value("peer").toString();
        request.port = map.value("port").toUInt();
        request.protocol = (NetworkManager::SocketType)map.value("protocol").toInt();
        return request;
    }

    QVariant measurementDefinition;
    QUuid taskId;
    QUuid measurementUuid;
    QString measurement;
    QString peer;
    quint16 port;
    NetworkManager::SocketType protocol;
};

class NetworkManager::Private : public QObject
{
    Q_OBJECT

public:
    Private(NetworkManager* q)
        : q(q)
    {
        connect(&timer, SIGNAL(timeout()), this, SLOT(timeout()));
    }

    NetworkManager* q;

    // Properties
    QMutex mutex;
    QHash<QString, QObject*> objectHash;
    QHash<QString, QAbstractSocket*> socketHash;

    QTimer timer;
    QPointer<QUdpSocket> socket;
    QPointer<Scheduler> scheduler;
    QPointer<Settings> settings;

    QSet<QUuid> handledMeasureUuids;

    quint16 localPort;

    QNetworkConfigurationManager ncm;

    // Functions
    QAbstractSocket* createSocket(NetworkManager::SocketType socketType);

    void updateSocket();
    void updateTimer();
    void processDatagram(const QByteArray& datagram, const QHostAddress& host, quint16 port);

public slots:
    void socketDestroyed(QObject* obj);
    void responseChanged();
    void timeout();
    void onDatagramReady();
};

QAbstractSocket *NetworkManager::Private::createSocket(NetworkManager::SocketType socketType)
{
    QAbstractSocket* socket = NULL;

    switch (socketType)
    {
    case TcpSocket:
        socket = new ::TcpSocket;
        break;

    case UdpSocket:
        socket = new ::UdpSocket;
        break;

    case UtpSocket:
        //socket = new UtpSocket;
        break;

    default:
        break;
    }

    if (socket)
    {
        connect(socket, SIGNAL(destroyed(QObject*)), this, SLOT(socketDestroyed(QObject*)));
    }

    return socket;
}

void NetworkManager::Private::socketDestroyed(QObject *obj)
{
    QString hostname = objectHash.key(obj);
    if (hostname.isEmpty())
    {
        return;
    }

    objectHash.remove(hostname);
    socketHash.remove(hostname);
}

void NetworkManager::Private::updateSocket()
{
    if (!socket.isNull())
    {
        delete socket.data();
        socket.clear();
    }

    /*
    QString keepaliveAddress = settings->config()->keepaliveAddress();
    RemoteHost remote = NetworkHelper::remoteHost(keepaliveAddress);
    if (!remote.isValid())
    {
        return;
    }
    */

    localPort = 5106; //remote.port;

    socket = qobject_cast<QUdpSocket*>( q->createConnection(NetworkManager::UdpSocket) );
    socket->setParent(this);
    connect(socket.data(), SIGNAL(readyRead()), this, SLOT(onDatagramReady()));
    if (!socket->bind(5106))
    {
        LOG_ERROR(QString("Unable to bind port %1: %2").arg(5106).arg(socket->errorString()));
    }
}

void NetworkManager::Private::updateTimer()
{
    TimingPtr timing = settings->config()->keepaliveSchedule();
    if (!timing)
    {
        return;
    }

    QSharedPointer<PeriodicTiming> periodicTiming = timing.dynamicCast<PeriodicTiming>();
    Q_ASSERT(periodicTiming);

    int interval = periodicTiming->period();
    if (interval < 1000)
    {
        LOG_INFO("Keepalive interval < 1 sec will not be accepted.");
{'fraction_non_alphanumeric': 0.13204625232294032, 'fraction_numerical': 0.0039231881065455295, 'mean_word_length': 3.385242191036668, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6258365', 'n_tokens_mistral': 2969, 'n_tokens_neox': 2894, 'n_words': 575}