MultichannelTcpReceiver.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "CoreMinimal.h"
#include "NetworkMessage.h"
#include "HAL/RunnableThread.h"
#include "HAL/Runnable.h"
#include "MultichannelTcpGlobals.h"
 
class Error;
class FSocket;
 
DECLARE_DELEGATE_ThreeParams(FOnMultichannelTcpReceive, const TArray<uint8>&, uint32, bool);
 
 
class FMultichannelTcpReceiver
    : public FRunnable
{
public:
 
    FMultichannelTcpReceiver( FSocket* InSocket, const FOnMultichannelTcpReceive& InReceiveDelegate )
        : Socket(InSocket)
        , ReceiveDelegate(InReceiveDelegate)
    {
        Thread = FRunnableThread::Create(this, TEXT("FMultichannelTCPReceiver"), 8 * 1024, TPri_AboveNormal);
    }
 
    ~FMultichannelTcpReceiver()
    {
        if (Thread)
        {
            Thread->Kill(true);
            delete Thread;
        }
    }
 
public:
 
    int32 GetBytesReceived()
    {
        return BytesReceived;
    }
 
public:
 
    // FRunnable interface
 
    virtual bool Init( )
    {
        return true;
    }
 
    virtual uint32 Run( )
    {
        while (1)
        {
            // read a header and payload pair
            FArrayReader Payload; 
 
            if (FNFSMessageHeader::ReceivePayload(Payload, FSimpleAbstractSocket_FSocket(Socket)) == false)
            {
                // if we failed to receive the payload, then the client is most likely dead, so, we can kill this connection
                // @todo: Add more error codes, maybe some errors shouldn't kill the connection
                break;
            }
 
            uint32 Magic;
 
            Payload << Magic;
 
            if (Magic != MultichannelMagic)
            {
                UE_LOG(LogMultichannelTCP, Error, TEXT("Wrong magic."));
 
                break;
            }
 
            uint32 Channel;
            Payload << Channel;
            
            TArray<uint8> InnerPayload;
            Payload << InnerPayload;
 
            ReceiveDelegate.Execute(InnerPayload, Channel, Payload.ForceByteSwapping());
 
            FPlatformAtomics::InterlockedAdd(&BytesReceived, InnerPayload.Num());
        }
 
        return 0;
    }
 
    virtual void Stop( ) { }
    virtual void Exit( ) { }
 
private:
 
    int32 BytesReceived;
 
    FSocket* Socket;
 
    FRunnableThread* Thread;
 
private:
 
    FOnMultichannelTcpReceive ReceiveDelegate;
};