支持单连接多通道通信

This commit is contained in:
_Redstone_c_ 2021-01-07 22:00:11 +08:00
parent af944c5b86
commit 35e1c37082
4 changed files with 141 additions and 89 deletions

View File

@ -7,39 +7,35 @@
#include "SocketSubsystem.h"
#include "..\Public\RedNetworkClient.h"
bool URedNetworkClient::Send(const TArray<uint8>& Data)
bool URedNetworkClient::Send(uint8 Channel, const TArray<uint8>& Data)
{
if (!IsActive() || !IsLogged()) return false;
return KCPUnit->Send(Data.GetData(), Data.Num()) == 0;
}
EnsureChannelCreated(Channel);
void URedNetworkClient::UDPSend(const uint8 * Data, int32 Count)
{
if (!IsActive()) return;
SendBuffer.SetNumUninitialized(8, false);
ClientPass.ToBytes(SendBuffer.GetData());
if (Count != 0) SendBuffer.Append(Data, Count);
int32 BytesSend;
SocketPtr->SendTo(SendBuffer.GetData(), SendBuffer.Num(), BytesSend, *ServerAddrPtr);
return KCPUnits[Channel]->Send(Data.GetData(), Data.Num()) == 0;
}
void URedNetworkClient::UpdateKCP()
{
if (!KCPUnit) return;
int32 Current = FPlatformTime::Cycles64() / 1000;
for (auto KCPUnit : KCPUnits)
{
if (!KCPUnit) continue;
KCPUnit->Update(Current);
}
}
void URedNetworkClient::SendHeartbeat()
{
UDPSend(nullptr, 0);
SendBuffer.SetNumUninitialized(8, false);
ClientPass.ToBytes(SendBuffer.GetData());
int32 BytesSend;
SocketPtr->SendTo(SendBuffer.GetData(), SendBuffer.Num(), BytesSend, *ServerAddrPtr);
}
void URedNetworkClient::HandleSocketRecv()
@ -71,11 +67,15 @@ void URedNetworkClient::HandleSocketRecv()
LastRecvTime = NowTime;
}
if (RecvBuffer.Num() == 8) continue;
if (RecvBuffer.Num() < 9) continue;
if (SourcePass.ID == ClientPass.ID && SourcePass.Key == ClientPass.Key)
{
KCPUnit->Input(RecvBuffer.GetData() + 8, RecvBuffer.Num() - 8);
uint8 Channel = RecvBuffer[8];
EnsureChannelCreated(Channel);
KCPUnits[Channel]->Input(RecvBuffer.GetData() + 9, RecvBuffer.Num() - 9);
}
}
}
@ -86,21 +86,17 @@ void URedNetworkClient::HandleLoginRecv(const FRedNetworkPass & SourcePass)
ClientPass = SourcePass;
KCPUnit = MakeShared<FKCPWrap>(ClientPass.ID, FString::Printf(TEXT("Client-%i"), ClientPass.ID));
KCPUnit->SetTurboMode();
KCPUnit->GetKCPCB().logmask = KCPLogMask;
KCPUnit->OutputFunc = [this](const uint8* Data, int32 Count)->int32
{
UDPSend(Data, Count);
return 0;
};
KCPUnits.SetNum(256);
OnLogin.Broadcast();
}
void URedNetworkClient::HandleKCPRecv()
{
for (int32 Channel = 0; Channel < KCPUnits.Num(); ++Channel)
{
const TSharedPtr<FKCPWrap>& KCPUnit = KCPUnits[Channel];
while (KCPUnit)
{
int32 Size = KCPUnit->PeekSize();
@ -115,7 +111,8 @@ void URedNetworkClient::HandleKCPRecv()
RecvBuffer.SetNumUninitialized(Size, false);
OnRecv.Broadcast(RecvBuffer);
OnRecv.Broadcast(Channel, RecvBuffer);
}
}
}
@ -125,7 +122,7 @@ void URedNetworkClient::HandleTimeout()
{
ClientPass.Reset();
KCPUnit = nullptr;
KCPUnits.SetNum(0);
UE_LOG(LogRedNetwork, Warning, TEXT("Red Network Client timeout."));
@ -133,6 +130,33 @@ void URedNetworkClient::HandleTimeout()
}
}
void URedNetworkClient::EnsureChannelCreated(uint8 Channel)
{
if (KCPUnits[Channel]) return;
TSharedPtr<FKCPWrap> KCPUnit = MakeShared<FKCPWrap>(0, FString::Printf(TEXT("Client-%i:%i"), ClientPass.ID, Channel));
KCPUnit->SetTurboMode();
KCPUnit->GetKCPCB().logmask = KCPLogMask;
KCPUnit->OutputFunc = [this, Channel](const uint8* Data, int32 Count)->int32
{
SendBuffer.SetNumUninitialized(9, false);
ClientPass.ToBytes(SendBuffer.GetData());
SendBuffer[8] = Channel;
if (Count != 0) SendBuffer.Append(Data, Count);
int32 BytesSend;
SocketPtr->SendTo(SendBuffer.GetData(), SendBuffer.Num(), BytesSend, *ServerAddrPtr);
return 0;
};
KCPUnits[Channel] = KCPUnit;
}
void URedNetworkClient::Tick(float DeltaTime)
{
if (!IsActive()) return;
@ -213,7 +237,7 @@ void URedNetworkClient::Deactivate()
ClientPass.Reset();
KCPUnit = nullptr;
KCPUnits.SetNum(0);
UE_LOG(LogRedNetwork, Log, TEXT("Red Network Client deactivate."));

View File

@ -8,13 +8,15 @@
#include "HAL/UnrealMemory.h"
#include "..\Public\RedNetworkServer.h"
bool URedNetworkServer::Send(int32 ClientID, const TArray<uint8>& Data)
bool URedNetworkServer::Send(int32 ClientID, uint8 Channel, const TArray<uint8>& Data)
{
if (!IsActive() || !Connections.Contains(ClientID)) return false;
const FConnectionInfo& Info = Connections[ClientID];
return Info.KCPUnit->Send(Data.GetData(), Data.Num()) == 0;
EnsureChannelCreated(ClientID, Channel);
return Info.KCPUnits[Channel]->Send(Data.GetData(), Data.Num()) == 0;
}
TSharedPtr<FInternetAddr> URedNetworkServer::GetSocketAddr() const
@ -38,29 +40,18 @@ FString URedNetworkServer::GetSocketAddrString() const
return Addr ? Addr->ToString(true) : TEXT("");
}
void URedNetworkServer::UDPSend(int32 ClientID, const uint8* Data, int32 Count)
{
if (!IsActive() || !Connections.Contains(ClientID)) return;
const FConnectionInfo& Info = Connections[ClientID];
SendBuffer.SetNumUninitialized(8, false);
Info.Pass.ToBytes(SendBuffer.GetData());
if (Count != 0) SendBuffer.Append(Data, Count);
int32 BytesSend;
SocketPtr->SendTo(SendBuffer.GetData(), SendBuffer.Num(), BytesSend, *Info.Addr);
}
void URedNetworkServer::UpdateKCP()
{
int32 Current = FPlatformTime::Cycles64() / 1000;
for (auto Info : Connections)
{
Info.Value.KCPUnit->Update(Current);
for (auto KCPUnit : Info.Value.KCPUnits)
{
if (!KCPUnit) continue;
KCPUnit->Update(Current);
}
}
}
@ -68,7 +59,12 @@ void URedNetworkServer::SendHeartbeat()
{
for (auto Info : Connections)
{
UDPSend(Info.Key, nullptr, 0);
SendBuffer.SetNumUninitialized(8, false);
Info.Value.Pass.ToBytes(SendBuffer.GetData());
int32 BytesSend;
SocketPtr->SendTo(SendBuffer.GetData(), SendBuffer.Num(), BytesSend, *Info.Value.Addr);
}
}
@ -105,11 +101,15 @@ void URedNetworkServer::HandleSocketRecv()
Connections[SourcePass.ID].RecvTime = NowTime;
if (RecvBuffer.Num() == 8) continue;
if (RecvBuffer.Num() < 9) continue;
if (Connections.Contains(SourcePass.ID))
{
Connections[SourcePass.ID].KCPUnit->Input(RecvBuffer.GetData() + 8, RecvBuffer.Num() - 8);
uint8 Channel = RecvBuffer[8];
EnsureChannelCreated(SourcePass.ID, Channel);
Connections[SourcePass.ID].KCPUnits[Channel]->Input(RecvBuffer.GetData() + 9, RecvBuffer.Num() - 9);
}
}
}
@ -167,15 +167,7 @@ void URedNetworkServer::RegisterConnection(const FRedNetworkPass& SourcePass, co
NewConnections.Heartbeat = FDateTime::MinValue();
NewConnections.Addr = SourceAddr;
NewConnections.KCPUnit = MakeShared<FKCPWrap>(NewConnections.Pass.ID, FString::Printf(TEXT("Server-%i"), NewConnections.Pass.ID));
NewConnections.KCPUnit->SetTurboMode();
NewConnections.KCPUnit->GetKCPCB().logmask = KCPLogMask;
NewConnections.KCPUnit->OutputFunc = [this, ID = NewConnections.Pass.ID](const uint8* Data, int32 Count)->int32
{
UDPSend(ID, Data, Count);
return 0;
};
NewConnections.KCPUnits.SetNum(256);
Connections.Add(SourcePass.ID, NewConnections);
@ -190,21 +182,26 @@ void URedNetworkServer::HandleKCPRecv()
{
for (auto Info : Connections)
{
while (Info.Value.KCPUnit)
for (int32 Channel = 0; Channel < Info.Value.KCPUnits.Num(); ++Channel)
{
int32 Size = Info.Value.KCPUnit->PeekSize();
const TSharedPtr<FKCPWrap>& KCPUnit = Info.Value.KCPUnits[Channel];
while (KCPUnit)
{
int32 Size = KCPUnit->PeekSize();
if (Size < 0) break;
RecvBuffer.SetNumUninitialized(Size, false);
Size = Info.Value.KCPUnit->Recv(RecvBuffer.GetData(), RecvBuffer.Num());
Size = KCPUnit->Recv(RecvBuffer.GetData(), RecvBuffer.Num());
if (Size < 0) break;
RecvBuffer.SetNumUninitialized(Size, false);
OnRecv.Broadcast(Info.Key, RecvBuffer);
OnRecv.Broadcast(Info.Key, Channel, RecvBuffer);
}
}
}
}
@ -243,6 +240,37 @@ void URedNetworkServer::HandleExpiredConnection()
}
}
void URedNetworkServer::EnsureChannelCreated(int32 ClientID, uint8 Channel)
{
FConnectionInfo& Info = Connections[ClientID];
if (Info.KCPUnits[Channel]) return;
TSharedPtr<FKCPWrap> KCPUnit = MakeShared<FKCPWrap>(0, FString::Printf(TEXT("Server-%i:%i"), ClientID, Channel));
KCPUnit->SetTurboMode();
KCPUnit->GetKCPCB().logmask = KCPLogMask;
KCPUnit->OutputFunc = [this, ClientID, Channel](const uint8* Data, int32 Count)->int32
{
const FConnectionInfo& Info = Connections[ClientID];
SendBuffer.SetNumUninitialized(9, false);
Info.Pass.ToBytes(SendBuffer.GetData());
SendBuffer[8] = Channel;
if (Count != 0) SendBuffer.Append(Data, Count);
int32 BytesSend;
SocketPtr->SendTo(SendBuffer.GetData(), SendBuffer.Num(), BytesSend, *Info.Addr);
return 0;
};
Info.KCPUnits[Channel] = KCPUnit;
}
void URedNetworkServer::Tick(float DeltaTime)
{
if (!IsActive()) return;

View File

@ -17,7 +17,7 @@ class REDNETWORK_API URedNetworkClient : public UObject, public FTickableGameObj
public:
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE(FLoginSignature, URedNetworkClient, OnLogin);
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE_OneParam(FRecvSignature, URedNetworkClient, OnRecv, const TArray<uint8>&, Data);
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE_TwoParams(FRecvSignature, URedNetworkClient, OnRecv, uint8, Channel, const TArray<uint8>&, Data);
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE(FUnloginSignature, URedNetworkClient, OnUnlogin);
public:
@ -46,7 +46,7 @@ public:
bool IsLogged() const { return ClientPass.IsValid(); }
UFUNCTION(BlueprintCallable, Category = "Red|Network")
bool Send(const TArray<uint8>& Data);
bool Send(uint8 Channel, const TArray<uint8>& Data);
public:
@ -78,9 +78,7 @@ private:
FDateTime LastRecvTime;
FDateTime LastHeartbeat;
TSharedPtr<FKCPWrap> KCPUnit;
void UDPSend(const uint8* Data, int32 Count);
TArray<TSharedPtr<FKCPWrap>> KCPUnits;
FDateTime NowTime;
@ -91,6 +89,8 @@ private:
void HandleKCPRecv();
void HandleTimeout();
void EnsureChannelCreated(uint8 Channel);
public:
//~ Begin FTickableGameObject Interface

View File

@ -18,7 +18,7 @@ class REDNETWORK_API URedNetworkServer : public UObject, public FTickableGameObj
public:
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE_OneParam(FLoginSignature, URedNetworkServer, OnLogin, int32, ClientID);
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE_TwoParams(FRecvSignature, URedNetworkServer, OnRecv, int32, ClientID, const TArray<uint8>&, Data);
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE_ThreeParams(FRecvSignature, URedNetworkServer, OnRecv, int32, ClientID, uint8, Channel, const TArray<uint8>&, Data);
DECLARE_DYNAMIC_MULTICAST_SPARSE_DELEGATE_OneParam(FUnloginSignature, URedNetworkServer, OnUnlogin, int32, ClientID);
public:
@ -44,7 +44,7 @@ public:
void Deactivate();
UFUNCTION(BlueprintCallable, Category = "Red|Network")
bool Send(int32 ClientID, const TArray<uint8>& Data);
bool Send(int32 ClientID, uint8 Channel, const TArray<uint8>& Data);
TSharedPtr<FInternetAddr> GetSocketAddr() const;
@ -90,13 +90,11 @@ private:
FDateTime RecvTime;
FDateTime Heartbeat;
TSharedPtr<FInternetAddr> Addr;
TSharedPtr<FKCPWrap> KCPUnit;
TArray<TSharedPtr<FKCPWrap>> KCPUnits;
};
TMap<int32, FConnectionInfo> Connections;
void UDPSend(int32 ClientID, const uint8* Data, int32 Count);
FDateTime NowTime;
void UpdateKCP();
@ -109,6 +107,8 @@ private:
void HandleExpiredReadyPass();
void HandleExpiredConnection();
void EnsureChannelCreated(int32 ClientID, uint8 Channel);
public:
//~ Begin FTickableGameObject Interface