MultichannelTcpSender.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "CoreMinimal.h"
#include "Serialization/BufferArchive.h"
#include "NetworkMessage.h"
#include "Misc/ScopedEvent.h"
#include "Misc/ScopeLock.h"
#include "HAL/RunnableThread.h"
#include "HAL/Runnable.h"
#include "MultichannelTcpGlobals.h"
 
class Error;
class FSocket;
 
DECLARE_DELEGATE_RetVal_TwoParams(bool, FOnMultichannelTcpOkToSend, int32, uint32);
 
 
class FMultichannelTcpSender
    : public FRunnable
{
    enum
    {
        MaxPacket = 128 * 1024 - 8
    };
 
public:
 
    FMultichannelTcpSender( FSocket* InSocket, const FOnMultichannelTcpOkToSend& InOkToSendDelegate )
        : BytesSent(0)
        , Socket(InSocket)
        , OkToSendDelegate(InOkToSendDelegate)
    {
        Thread = FRunnableThread::Create(this, TEXT("FMultichannelTCPSender"), 8 * 1024, TPri_AboveNormal);
    }
 
    ~FMultichannelTcpSender(  )
    {
        if (Thread)
        {
            Thread->Kill(true);
            delete Thread;
        }
    }
 
public:
 
    void AttemptResumeSending( )
    {
        FScopeLock ScopeLock(&SendBuffersCriticalSection);
        AttemptResumeSendingInternal();
    }
 
    int32 GetBytesSent()
    {
        return BytesSent;
    }
 
    void Send( const uint8* Data, int32 Count, uint32 Channel )
    {
        FScopeLock ScopeLock(&SendBuffersCriticalSection);
        check(Count);
 
        TArray<uint8>* SendBuffer = SendBuffers.FindRef(Channel);
 
        if (!SendBuffer)
        {
            SendBuffer = new TArray<uint8>();
            SendBuffers.Add(Channel, SendBuffer);
        }
 
        int32 OldCount = SendBuffer->Num();
 
        SendBuffer->AddUninitialized(Count);
 
        FMemory::Memcpy(SendBuffer->GetData() + OldCount, Data, Count);
 
        AttemptResumeSendingInternal();
    }
 
public:
 
    // FRunnable interface
 
    virtual bool Init( )
    {
        return true;
    }
 
    virtual uint32 Run( )
    {
        while (1)
        {
            TArray<uint8> Data;
            uint32 Channel = 0;
 
            FScopedEvent* LocalEventToRestart = NULL;
            {
                FScopeLock ScopeLock(&SendBuffersCriticalSection);
 
                TArray<uint8>* SendBuffer = NULL;
 
                for (TMap<uint32, TArray<uint8>*>::TIterator It(SendBuffers); It; ++It)
                {
                    check(It.Value()->Num());
 
                    if (!SendBuffer || It.Key() < Channel)
                    {
                        Channel = It.Key();
                        SendBuffer = It.Value();
                    }
                }
 
                if (SendBuffer)
                {
                    int32 Num = SendBuffer->Num();
                    check(Num > 0);
 
                    int32 Size = FMath::Min<int32>(Num, MaxPacket);
 
                    if (OkToSendDelegate.Execute(Size, Channel))
                    {
                        Data.AddUninitialized(Size);
                        FMemory::Memcpy(Data.GetData(), SendBuffer->GetData(), Size);
 
                        if (Size < Num)
                        {
                            SendBuffer->RemoveAt(0, Size);
                        }
                        else
                        {
                            delete SendBuffer;
 
                            SendBuffers.Remove(Channel);
                        }
 
                        check(Data.Num());
                    }
                }
                else
                {
                    LocalEventToRestart = new FScopedEvent();
                    EventToRestart = LocalEventToRestart;
                }
            }
 
            if (Data.Num())
            {
                FBufferArchive Ar;
 
                uint32 Magic = MultichannelMagic;
 
                Ar << Magic;
                Ar << Channel;
                Ar << Data;
 
                if (!FNFSMessageHeader::WrapAndSendPayload(Ar, FSimpleAbstractSocket_FSocket(Socket)))
                {
                    UE_LOG(LogMultichannelTCP, Error, TEXT("Failed to send payload."));
 
                    break;
                }
 
                FPlatformAtomics::InterlockedAdd(&BytesSent, Data.Num());
            }
 
            delete LocalEventToRestart; // block here if we don't have any data
        }
 
        return 0;
    }
 
    virtual void Stop( ) { }
 
    virtual void Exit( )
    {
        FScopeLock ScopeLock(&SendBuffersCriticalSection);
 
        for (TMap<uint32, TArray<uint8>*>::TIterator It(SendBuffers); It; ++It)
        {
            delete It.Value();
        }
    }
 
protected:
 
    void AttemptResumeSendingInternal( )
    {
        if (EventToRestart)
        {
            FScopedEvent* LocalEventToRestart = EventToRestart;
            EventToRestart = NULL;
            LocalEventToRestart->Trigger();
        }
    }
 
private:
 
    int32 BytesSent;
 
    FScopedEvent* EventToRestart;
 
    TMap<uint32, TArray<uint8>*> SendBuffers;
 
    FCriticalSection SendBuffersCriticalSection;
 
    FSocket* Socket;
 
    FRunnableThread* Thread;
 
private:
 
    FOnMultichannelTcpOkToSend OkToSendDelegate;
};