ReceivedPacketProcessor.h

//
//  ReceivedPacketProcessor.h
//  libraries/networking/src
//
//  Created by Brad Hefta-Gaub on 8/12/13.
//  Copyright 2013 High Fidelity, Inc.
//
//  Distributed under the Apache License, Version 2.0.
//  See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html
//
 
#ifndef hifi_ReceivedPacketProcessor_h
#define hifi_ReceivedPacketProcessor_h
 
#include <QtCore/QSharedPointer>
#include <QWaitCondition>
 
#include "NodeList.h"
 
#include "GenericThread.h"
 
class ReceivedMessage;
 
class ReceivedPacketProcessor : public GenericThread {
    Q_OBJECT
public:
    static const uint64_t MAX_WAIT_TIME { 100 }; // Max wait time in ms
 
    ReceivedPacketProcessor();
 
    void queueReceivedPacket(QSharedPointer<ReceivedMessage> message, SharedNodePointer sendingNode);
 
    bool hasPacketsToProcess() const { return _packets.size() > 0; }
 
    bool isAlive(const QUuid& nodeUUID) const {
        return _nodePacketCounts.contains(nodeUUID);
    }
 
    bool hasPacketsToProcessFrom(const SharedNodePointer& sendingNode) const {
        return hasPacketsToProcessFrom(sendingNode->getUUID());
    }
 
    bool hasPacketsToProcessFrom(const QUuid& nodeUUID) const {
        return _nodePacketCounts[nodeUUID] > 0;
    }
 
    int packetsToProcessCount() const { return (int)_packets.size(); }
 
    float getIncomingPPS() const { return _incomingPPS.getAverage(); }
    float getProcessedPPS() const { return _processedPPS.getAverage(); }
 
    virtual void terminating() override;
 
public slots:
    void nodeKilled(SharedNodePointer node);
 
protected:
    virtual void processPacket(QSharedPointer<ReceivedMessage> message, SharedNodePointer sendingNode) = 0;
 
    virtual bool process() override;
 
    virtual uint32_t getMaxWait() const { return MAX_WAIT_TIME; }
 
    virtual void preProcess() { }
 
    virtual void midProcess() { }
 
    virtual void postProcess() { }
 
protected:
    std::list<NodeSharedReceivedMessagePair> _packets;
    QHash<QUuid, int> _nodePacketCounts;
 
    QWaitCondition _hasPackets;
    QMutex _waitingOnPacketsMutex;
 
    quint64 _lastWindowAt = 0;
    int _lastWindowIncomingPackets = 0;
    int _lastWindowProcessedPackets = 0;
    SimpleMovingAverage _incomingPPS;
    SimpleMovingAverage _processedPPS;
};
 
#endif // hifi_ReceivedPacketProcessor_h