UDPReceiver.hh

#ifndef artdaq_ots_Generators_UDPReceiver_hh
#define artdaq_ots_Generators_UDPReceiver_hh
 
#include "artdaq-core/Data/Fragment.hh"
#include "artdaq/Generators/CommandableFragmentGenerator.hh"
#include "fhiclcpp/fwd.h"
 
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
 
#include <array>
#include <atomic>
#include <chrono>
#include <list>
#include <mutex>
#include <queue>
#include <thread>
 
namespace ots
{
enum class CommandType : uint8_t
{
    Read        = 0,
    Write       = 1,
    Start_Burst = 2,
    Stop_Burst  = 3,
};
 
enum class ReturnCode : uint8_t
{
    Read   = 0,
    First  = 1,
    Middle = 2,
    Last   = 3,
};
 
enum class DataType : uint8_t
{
    Raw    = 0,
    JSON   = 1,
    String = 2,
};
 
struct CommandPacket
{
    CommandType type;
    uint8_t     dataSize;
    uint64_t    address;
    uint64_t    data[182];
};
 
typedef std::string               packetBuffer_t;
typedef std::list<packetBuffer_t> packetBuffer_list_t;
 
class UDPReceiver : public artdaq::CommandableFragmentGenerator
{
  public:
    explicit UDPReceiver(fhicl::ParameterSet const& ps);
    virtual ~UDPReceiver();
 
  protected:
    bool         getNext_(artdaq::FragmentPtrs& output) override;
    void         start(void) override;
    virtual void start_();
    virtual void stop(void) override;
    virtual void stopNoMutex(void) override;
 
    virtual void ProcessData_(artdaq::FragmentPtrs& output, size_t totalSize);
 
    DataType getDataType(uint8_t byte)
    {
        return static_cast<DataType>((byte & 0xF0) >> 4);
    }
    ReturnCode getReturnCode(uint8_t byte) { return static_cast<ReturnCode>(byte & 0xF); }
    void       send(CommandType flag);
 
    packetBuffer_list_t packetBuffers_;
 
    bool        rawOutput_;
    std::string rawPath_;
 
    int         dataport_;
    std::string ip_;
    int         rcvbuf_;
 
    uint8_t expectedPacketNumber_;
 
    struct sockaddr_in si_data_;
    int                datasocket_;
    bool               sendCommands_;
 
  private:
    void receiveLoop_();
    bool isTimerExpired_();
 
    std::unique_ptr<std::thread> receiverThread_;
    std::mutex                   receiveBufferLock_;
    packetBuffer_list_t          receiveBuffers_;
 
    bool fakeDataMode_;
 
    double                                         fragmentWindow_;
    std::chrono::high_resolution_clock::time_point lastFrag_;
};
}  // namespace ots
 
#endif /* artdaq_demo_Generators_ToySimulator_hh */