【UE】Unity エディタと Unreal エディタでローカルマルチプレイしよう！～UE 編～

#pragma once

#include "CoreMinimal.h"
#include "Kismet/BlueprintAsyncActionBase.h"
#include "AsyncActionSendMessage.generated.h"

DECLARE_DYNAMIC_MULTICAST_DELEGATE_OneParam(FOnCompleted, bool, bSucceeded);

UCLASS()
class LOCALCONNECTSAMPLE_API UAsyncActionSendMessage : public UBlueprintAsyncActionBase
{
    GENERATED_BODY()
	
public:
    UFUNCTION(BlueprintCallable, meta = (BlueprintInternalUseOnly = "true", WorldContext = "WorldContextObject"))
    static UAsyncActionSendMessage* AsyncSendMessage(UObject* WorldContextObject, const FString& InMessage, const FString& InAddress = TEXT("127.0.0.1"),  const int32 InPort = 8000);

    //~ Begin UBlueprintAsyncActionBase Interface
    virtual void Activate() override;
    //~ End UBlueprintAsyncActionBase Interface

private:
    UPROPERTY(BlueprintAssignable)
    FOnCompleted OnCompleted;

    FString Message;
    FString Address;
    int32 Port;

    void CallOnCompleted(const bool bSucceeded);
};

#include "AsyncActionSendMessage.h"
#include "Sockets.h"
#include "SocketSubsystem.h"
#include "Interfaces/IPv4/IPv4Address.h"
#include "Interfaces/IPv4/IPv4Endpoint.h"

UAsyncActionSendMessage* UAsyncActionSendMessage::AsyncSendMessage(UObject* WorldContextObject, const FString& InMessage, const FString& InAddress, const int32 InPort)
{
    check(WorldContextObject);
    UAsyncActionSendMessage* Action = NewObject<UAsyncActionSendMessage>();

    Action->Message = InMessage;
    Action->Address = InAddress;
    Action->Port = InPort;

    Action->RegisterWithGameInstance(WorldContextObject);

    return Action;
}

void UAsyncActionSendMessage::Activate()
{
    TWeakObjectPtr<UAsyncActionSendMessage> WeakThis = this;
    AsyncTask(ENamedThreads::AnyBackgroundThreadNormalTask, [WeakThis]()
    {
        if (!WeakThis.IsValid())
        {
            return;
        }
		
        bool bSucceeded = true;
		
        if (ISocketSubsystem* SocketSubsystem = ISocketSubsystem::Get(PLATFORM_SOCKETSUBSYSTEM))
        {
            const FString Message = WeakThis->Message;
            const FString Address = WeakThis->Address;
            const int32 Port = WeakThis->Port;
		
            FIPv4Address ParsedAddress;
            if (FIPv4Address::Parse(Address, ParsedAddress))
            {
                if (FSocket* ClientSocket = SocketSubsystem->CreateSocket(NAME_Stream, TEXT("ClientSocket")))
                {
                    ClientSocket->SetNonBlocking(false);
                    const FIPv4Endpoint Endpoint(ParsedAddress, Port);

                    if (ClientSocket->Connect(*Endpoint.ToInternetAddr()))
                    {
                        const FTCHARToUTF8 Utf8Message(*Message);
                        int32 BytesSent = 0;

                        if (!ClientSocket->Send(reinterpret_cast<const uint8*>(Utf8Message.Get()), Utf8Message.Length(), BytesSent) || BytesSent != Utf8Message.Length())
                        {
                            UE_LOG(LogTemp, Error, TEXT("Failed to send message. Error: %s"), SocketSubsystem->GetSocketError(SocketSubsystem->GetLastErrorCode()));
                            bSucceeded = false;
                        }
                    }
                    else
                    {
                        UE_LOG(LogTemp, Error, TEXT("Failed to connect to %s:%d. Error: %s"), *Address, Port, SocketSubsystem->GetSocketError(SocketSubsystem->GetLastErrorCode()));
                        bSucceeded = false;
                    }
			
                    ClientSocket->Close();
                }
                else
                {
                    UE_LOG(LogTemp, Error, TEXT("Failed to create client socket."));
                    bSucceeded = false;
                }
            }
            else
            {
                UE_LOG(LogTemp, Error, TEXT("Failed to parse %s."), *Address);
                bSucceeded = false;
            }
        }
        else
        {
            UE_LOG(LogTemp, Error, TEXT("Failed to get SocketSubsystem."));
            bSucceeded = false;
        }

        AsyncTask(ENamedThreads::GameThread, [WeakThis, bSucceeded]()
        {
            if (WeakThis.IsValid())
            {
                WeakThis->CallOnCompleted(bSucceeded);
            }
        });
    });
}

void UAsyncActionSendMessage::CallOnCompleted(const bool bSucceeded)
{
    OnCompleted.Broadcast(bSucceeded);
    SetReadyToDestroy();
}

#pragma once

#include "CoreMinimal.h"
#include "Interfaces/IPv4/IPv4Address.h"
#include "Kismet/BlueprintAsyncActionBase.h"
#include "AsyncActionStartListening.generated.h"

DECLARE_DYNAMIC_MULTICAST_DELEGATE_OneParam(FOnReceivedMessage, const FString&, ReceivedMessage);
DECLARE_DYNAMIC_MULTICAST_DELEGATE(FOnFailed);

UCLASS()
class LOCALCONNECTSAMPLE_API UAsyncActionStartListening : public UBlueprintAsyncActionBase, public FTickableGameObject
{
    GENERATED_BODY()
	
public:
UFUNCTION(BlueprintCallable, meta = (BlueprintInternalUseOnly = "true", WorldContext = "WorldContextObject"))
    static UAsyncActionStartListening* AsyncStartListening(UObject* WorldContextObject, const FString& ListenerAddress = TEXT("127.0.0.1"),  const int32 ListenerPort = 8000);

    //~ Begin UBlueprintAsyncActionBase Interface
    virtual void Activate() override;
    //~ End UBlueprintAsyncActionBase Interface
	
    //~ Begin FTickableObjectBase Interface
    virtual void Tick(float DeltaTime) override;
    virtual TStatId GetStatId() const override;
    virtual ETickableTickType GetTickableTickType() const override;
    //~ End FTickableObjectBase Interface

    //~ Begin UObject Interface
    virtual void BeginDestroy() override;
    //~ End UObject Interface
	
    void CallOnReceivedMessage(const FString& ReceivedMessage) const;
    void CallOnFailed();
	
private:
    UPROPERTY(BlueprintAssignable)
    FOnReceivedMessage OnReceivedMessage;
	
    UPROPERTY(BlueprintAssignable)
    FOnFailed OnFailed;
	
    bool bIsListening;
	
    FIPv4Address ListenerIPv4Address;
    FString ListenerAddress;
    int32 ListenerPort;
	
    TSharedPtr<FSocket> ListenerSocket;
    FRunnableThread* ListenerThread;
    TWeakObjectPtr<UObject> WorldContextObject;

    void StopListening();
};

#include "AsyncActionStartListening.h"
#include "ListenerRunnable.h"
#include "Sockets.h"
#include "SocketSubsystem.h"
#include "Interfaces/IPv4/IPv4Endpoint.h"

UAsyncActionStartListening* UAsyncActionStartListening::AsyncStartListening(UObject* WorldContextObject, const FString& ListenerAddress, const int32 ListenerPort)
{
    check(WorldContextObject);
    UAsyncActionStartListening* Action = NewObject<UAsyncActionStartListening>();

    Action->ListenerAddress = ListenerAddress;
    Action->ListenerPort = ListenerPort;
    Action->WorldContextObject = WorldContextObject;

    Action->RegisterWithGameInstance(WorldContextObject);

    return Action;
}

void UAsyncActionStartListening::Activate()
{
    if (ListenerPort < 0 || 65535 < ListenerPort)
    {
        UE_LOG(LogTemp, Error, TEXT("Listener port %d is incorrect value."), ListenerPort);
        CallOnFailed();
		
        return;
    }

    ISocketSubsystem* SocketSubsystem = ISocketSubsystem::Get(PLATFORM_SOCKETSUBSYSTEM);

    if (!SocketSubsystem)
    {
        UE_LOG(LogTemp, Error, TEXT("Failed to get SocketSubsystem."));
        CallOnFailed();
		
        return;
    }
	
    const TSharedPtr<FSocket> CreatedListenerSocket = MakeShareable(SocketSubsystem->CreateSocket(NAME_Stream, TEXT("ListenerSocket")));

    if (!CreatedListenerSocket.IsValid())
    {
        UE_LOG(LogTemp, Error, TEXT("Failed to create listener socket."));
        CallOnFailed();
		
        return;
    }
	
    CreatedListenerSocket->SetReuseAddr(true);
    CreatedListenerSocket->SetNonBlocking(true);
	
    if (!FIPv4Address::Parse(ListenerAddress, ListenerIPv4Address))
    {
        UE_LOG(LogTemp, Error, TEXT("Failed to parse %s."), *ListenerAddress);
        CallOnFailed();
		
        return;
    }
	
    const FIPv4Endpoint Endpoint(ListenerIPv4Address, ListenerPort);
    const TSharedRef<FInternetAddr> InternetAddress = Endpoint.ToInternetAddr();
    
    if (!CreatedListenerSocket->Bind(*InternetAddress))
    {
        UE_LOG(LogTemp, Error, TEXT("Failed to bind listener socket to %s:%d."), *ListenerIPv4Address.ToString(), ListenerPort);
        CallOnFailed();
		
        return;
    }
	
    if (!CreatedListenerSocket->Listen(1))
    {
        UE_LOG(LogTemp, Error, TEXT("Failed to start listening."));
        CallOnFailed();
		
        return;
    }
	
    ListenerSocket = CreatedListenerSocket;
    bIsListening = true;

    FListenerRunnable* ListenerRunnable = new FListenerRunnable(this, ListenerSocket);
    ListenerThread = FRunnableThread::Create(ListenerRunnable, TEXT("ListenerThread"));
	
    if (!ListenerThread)
    {
        UE_LOG(LogTemp, Error, TEXT("Failed to create listener thread."));
        CallOnFailed();
    }
}

void UAsyncActionStartListening::CallOnReceivedMessage(const FString& ReceivedMessage) const
{
    OnReceivedMessage.Broadcast(ReceivedMessage);
}

void UAsyncActionStartListening::CallOnFailed()
{
    StopListening();
    OnFailed.Broadcast();
    SetReadyToDestroy();
}

void UAsyncActionStartListening::Tick(float DeltaTime)
{
    if (!WorldContextObject.IsValid() && bIsListening)
    {
        StopListening();
        SetReadyToDestroy();
    }
}

TStatId UAsyncActionStartListening::GetStatId() const
{
    RETURN_QUICK_DECLARE_CYCLE_STAT(UAsyncActionStartListening, STATGROUP_Tickables);
}

ETickableTickType UAsyncActionStartListening::GetTickableTickType() const
{
    return IsTemplate() ? ETickableTickType::Never : ETickableTickType::Always;
}

void UAsyncActionStartListening::BeginDestroy()
{
    StopListening();
    Super::BeginDestroy();
}

void UAsyncActionStartListening::StopListening()
{
    if (!bIsListening)
    {
        return;
    }

    if (ListenerThread)
    {
        ListenerThread->Kill(false);
        delete ListenerThread;
        ListenerThread = nullptr;
    }
	
    if (ListenerSocket.IsValid())
    {
        if (ISocketSubsystem* SocketSubsystem = ISocketSubsystem::Get(PLATFORM_SOCKETSUBSYSTEM))
        {
            ListenerSocket->Close();
	    SocketSubsystem->DestroySocket(ListenerSocket.Get());
            ListenerSocket.Reset();
        }
        else
        {
            UE_LOG(LogTemp, Error, TEXT("Failed to get SocketSubsystem."));
        }
    }
	
    bIsListening = false;
}

#pragma once

#include "CoreMinimal.h"
#include "HAL/Runnable.h"

class UAsyncActionStartListening;

class FListenerRunnable : public FRunnable
{
public:
    explicit FListenerRunnable(const TWeakObjectPtr<UAsyncActionStartListening> InAsyncActionStartListening, const TSharedPtr<FSocket>& InListenerSocket);

    //~ Begin FRunnable Interface
    virtual bool Init() override;
    virtual uint32 Run() override;
    virtual void Stop() override;
    virtual void Exit() override;
    //~ End FRunnable Interface

private:
    bool bListenerThreadIsRunning;
    TWeakObjectPtr<UAsyncActionStartListening> AsyncActionStartListening;
	
    TSharedPtr<FSocket> ListenerSocket;
    TSharedPtr<FSocket> ClientSocket;
};

#include "ListenerRunnable.h"
#include "AsyncActionStartListening.h"
#include "Sockets.h"
#include "SocketSubsystem.h"

FListenerRunnable::FListenerRunnable(const TWeakObjectPtr<UAsyncActionStartListening> InAsyncActionStartListening, const TSharedPtr<FSocket>& InListenerSocket)
    : bListenerThreadIsRunning(true)
    , AsyncActionStartListening(InAsyncActionStartListening)
    , ListenerSocket(InListenerSocket)
{

}

bool FListenerRunnable::Init()
{
    return true;
}

uint32 FListenerRunnable::Run()
{
    while (bListenerThreadIsRunning && AsyncActionStartListening.IsValid())
    {
        bool bHasPendingConnection = false;
        if (!ListenerSocket->HasPendingConnection(bHasPendingConnection) || !bHasPendingConnection)
        {
            FPlatformProcess::Sleep(0.1f);
            continue;
        }
        
        ClientSocket = MakeShareable(ListenerSocket->Accept(TEXT("ClientSocket")));
        
        if (!ClientSocket.IsValid())
        {
            FPlatformProcess::Sleep(0.1f);
            continue;
        }
        
        int32 SocketBufferSize = 65536;
        ClientSocket->SetNoDelay(true);
        ClientSocket->SetSendBufferSize(SocketBufferSize, SocketBufferSize);
        ClientSocket->SetReceiveBufferSize(SocketBufferSize, SocketBufferSize);
        
        uint8 Buffer[8192];
        
        while (bListenerThreadIsRunning && AsyncActionStartListening.IsValid())
        {
            int32 BytesRead = 0;
            if (ClientSocket->Recv(Buffer, sizeof(Buffer), BytesRead))
            {
                if (BytesRead == 0 || !AsyncActionStartListening.IsValid())
                {
                    break;
                }

                Buffer[BytesRead] = '\0';
                const FString ReceivedMessage = UTF8_TO_TCHAR(Buffer);
                
                AsyncActionStartListening->CallOnReceivedMessage(ReceivedMessage);
            }

            ISocketSubsystem* SocketSubsystem = ISocketSubsystem::Get(PLATFORM_SOCKETSUBSYSTEM);
            const ESocketErrors LastErrorCode = SocketSubsystem->GetLastErrorCode();
            
            bool bShouldBreak = true;
            
            if (LastErrorCode == SE_EWOULDBLOCK) 
            {
                bShouldBreak = false;
                FPlatformProcess::Sleep(0.01f);
            }
            else if (LastErrorCode == SE_EINTR)
            {
                bShouldBreak = false;
            }
            
            if (bShouldBreak)
            {
                break;
            }
        }

        if (ClientSocket.IsValid())
        {
            ClientSocket->Close();
        }
    }
     
    if (AsyncActionStartListening.IsValid() && bListenerThreadIsRunning)
    {
        AsyncActionStartListening->CallOnFailed();
    }

    return 0;
}

void FListenerRunnable::Stop()
{
    bListenerThreadIsRunning = false;
}

void FListenerRunnable::Exit()
{

}