UdpSocketSender.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "CoreMinimal.h"
#include "Containers/Queue.h"
#include "HAL/Event.h"
#include "HAL/PlatformProcess.h"
#include "HAL/Runnable.h"
#include "HAL/RunnableThread.h"
#include "Misc/SingleThreadRunnable.h"
#include "Sockets.h"
 
#include "Interfaces/IPv4/IPv4Endpoint.h"
 
class FUdpSocketSender
    : public FRunnable
    , private FSingleThreadRunnable
{
    // Structure for outbound packets.
    struct FPacket
    {
        TSharedPtr<TArray<uint8>, ESPMode::ThreadSafe> Data;
 
        FIPv4Endpoint Recipient;
 
        FPacket() { }
 
        FPacket(const TSharedRef<TArray<uint8>, ESPMode::ThreadSafe>& InData, const FIPv4Endpoint& InRecipient)
            : Data(InData)
            , Recipient(InRecipient)
        { }
    };
 
public:
 
    FUdpSocketSender(FSocket* InSocket, const TCHAR* ThreadDescription)
        : SendRate(0)
        , Socket(InSocket)
        , Stopping(false)
        , WaitTime(FTimespan::FromMilliseconds(100))
    {
        check(Socket != nullptr);
        check(Socket->GetSocketType() == SOCKTYPE_Datagram);
 
        int32 NewSize = 0;
        Socket->SetSendBufferSize(512 * 1024, NewSize);
 
        WorkEvent = FPlatformProcess::GetSynchEventFromPool();
        Thread = FRunnableThread::Create(this, ThreadDescription, 128 * 1024, TPri_AboveNormal, FPlatformAffinity::GetPoolThreadMask());
    }
 
    virtual ~FUdpSocketSender()
    {
        if (Thread != nullptr)
        {
            Thread->Kill(true);
            delete Thread;
        }
 
        FPlatformProcess::ReturnSynchEventToPool(WorkEvent);
        WorkEvent = nullptr;
    }
 
public:
 
    uint32 GetSendRate() const
    {
        return SendRate;
    }
 
    uint32 GetThroughput() const
    {
        return Throughput;
    }
 
    bool Send(const TSharedRef<TArray<uint8>, ESPMode::ThreadSafe>& Data, const FIPv4Endpoint& Recipient)
    {
        if (!Stopping && SendQueue.Enqueue(FPacket(Data, Recipient)))
        {
            WorkEvent->Trigger();
 
            return true;
        }
 
        return false;
    }
 
    void SetSendRate(uint32 Rate)
    {
        SendRate = Rate;
    }
 
    void SetWaitTime(const FTimespan& Timespan)
    {
        WaitTime = Timespan;
    }
 
public:
 
    //~ FRunnable interface
 
    virtual FSingleThreadRunnable* GetSingleThreadInterface() override
    {
        return this;
    }
 
    virtual bool Init() override
    {
        return true;
    }
 
    virtual uint32 Run() override
    {
        while (!Stopping)
        {
            if (!Update(WaitTime))
            {
                Stopping = true;
 
                return 0;
            }
 
            WorkEvent->Wait(WaitTime);
        }
 
        return 0;
    }
 
    virtual void Stop() override
    {
        Stopping = true;
        WorkEvent->Trigger();
    }
 
    virtual void Exit() override { }
 
protected:
 
    bool Update(const FTimespan& SocketWaitTime)
    {
        while (!SendQueue.IsEmpty())
        {
            if (!Socket->Wait(ESocketWaitConditions::WaitForWrite, SocketWaitTime))
            {
                break;
            }
 
            FPacket Packet;
            int32 Sent = 0;
 
            SendQueue.Dequeue(Packet);
            Socket->SendTo(Packet.Data->GetData(), Packet.Data->Num(), Sent, *Packet.Recipient.ToInternetAddr());
 
            if (Sent != Packet.Data->Num())
            {
                return false;
            }
        }
 
        return true;
    }
 
protected:
 
    //~ FSingleThreadRunnable interface
 
    virtual void Tick() override
    {
        Update(FTimespan::Zero());
    }
 
private:
 
    TQueue<FPacket, EQueueMode::Mpsc> SendQueue;
 
    uint32 SendRate;
 
    FSocket* Socket;
 
    bool Stopping;
 
    FRunnableThread* Thread;
 
    uint32 Throughput;
 
    FTimespan WaitTime;
 
    FEvent* WorkEvent;
};