micialware/mROA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reconnection DONE!!!

Browse Source
This commit is contained in:
micialware
2025-04-08 22:42:00 +03:00
parent 24bd06b3d0
commit ab40cad3ee
5 changed files with 54 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mROA/Abstract/IInteractionModule.cs
+2 -2
View File
@@ -9,13 +9,13 @@ namespace mROA.Abstract
{ {
int ConnectionId { get; set; } int ConnectionId { get; set; }
public Stream? BaseStream { get; set; } public Stream? BaseStream { get; set; }
Task<NetworkMessageHeader> GetNextMessageReceiving(); public IntPtr StreamHandle { get; set; }
Task<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true);
Task PostMessageAsync(NetworkMessageHeader messageHeader); Task PostMessageAsync(NetworkMessageHeader messageHeader);
void HandleMessage(NetworkMessageHeader messageHeader); void HandleMessage(NetworkMessageHeader messageHeader);
NetworkMessageHeader[] UnhandledMessages { get; } NetworkMessageHeader[] UnhandledMessages { get; }
NetworkMessageHeader? FirstByFilter(Predicate<NetworkMessageHeader> predicate); NetworkMessageHeader? FirstByFilter(Predicate<NetworkMessageHeader> predicate);
event Action<int> OnDisconected; event Action<int> OnDisconected;
Task Restart(bool sendRecovery); Task Restart(bool sendRecovery);
} }
} }
mROA/Implementation/Backend/NetworkGatewayModule.cs
+6 -7
View File
@@ -1,10 +1,8 @@
using System; using System;
using System.Linq;
using System.Net; using System.Net;
using System.Net.Sockets; using System.Net.Sockets;
using System.Threading.Tasks; using System.Threading.Tasks;
using mROA.Abstract; using mROA.Abstract;
using static System.Byte;
namespace mROA.Implementation.Backend namespace mROA.Implementation.Backend
{ {
@@ -67,17 +65,18 @@ namespace mROA.Implementation.Backend
while (true) while (true)
{ {
var client = _tcpListener.AcceptTcpClient(); var client = _tcpListener.AcceptTcpClient();
Console.WriteLine($"Client connected from {client.Client.RemoteEndPoint}"); Console.WriteLine($"Client connected from {client.Client.RemoteEndPoint}");
var interaction = Activator.CreateInstance(_interactionModuleType!) as INextGenerationInteractionModule; var interaction = Activator.CreateInstance(_interactionModuleType!) as INextGenerationInteractionModule;
foreach (var injectableModule in _injectableModules!) foreach (var injectableModule in _injectableModules!)
interaction!.Inject(injectableModule); interaction!.Inject(injectableModule);
interaction!.Inject(_serialization); interaction!.Inject(_serialization);
interaction.BaseStream = client.GetStream(); interaction.BaseStream = client.GetStream();
interaction.StreamHandle = client.Client.Handle;
var connectionRequest = interaction.GetNextMessageReceiving().GetAwaiter().GetResult()!; var connectionRequest = interaction.GetNextMessageReceiving(false)
.GetAwaiter().GetResult()!;
switch (connectionRequest.MessageType) switch (connectionRequest.MessageType)
{ {
@@ -92,10 +91,10 @@ namespace mROA.Implementation.Backend
var recoveryRequest = _serialization!.Deserialize<ClientRecovery>(connectionRequest.Data)!; var recoveryRequest = _serialization!.Deserialize<ClientRecovery>(connectionRequest.Data)!;
var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id); var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id);
recoveryInteraction.BaseStream = client.GetStream(); recoveryInteraction.BaseStream = client.GetStream();
recoveryInteraction.StreamHandle = client.Client.Handle;
recoveryInteraction.Restart(false); recoveryInteraction.Restart(false);
Console.WriteLine($"Client {recoveryRequest.Id} reconnected"); Console.WriteLine("Connection recovery for client {0} finished", recoveryRequest.Id);
break; break;
} }
default: default:
mROA/Implementation/Frontend/NetworkFrontendBridge.cs
+2 -2
View File
@@ -71,8 +71,8 @@ namespace mROA.Implementation.Frontend
public void Obstacle() public void Obstacle()
{ {
_interactionModule!.BaseStream!.Close(); _interactionModule!.BaseStream!.Dispose();
_tcpClient.Close(); _tcpClient.Dispose();
} }
public void Disconnect() public void Disconnect()
mROA/Implementation/NetworkMessageHeader.cs
+1
View File
@@ -18,6 +18,7 @@ namespace mROA.Implementation
{ {
MessageType = networkMessage.MessageType; MessageType = networkMessage.MessageType;
Data = serializationToolkit.Serialize(networkMessage); Data = serializationToolkit.Serialize(networkMessage);
Id = Guid.NewGuid();
} }
public Guid Id { get; set; } public Guid Id { get; set; }
mROA/Implementation/NextGenerationInteractionModule.cs
+43 -64
View File
@@ -31,16 +31,11 @@ namespace mROA.Implementation
public Stream? BaseStream public Stream? BaseStream
{ {
get => _baseStream; get => _baseStream;
set set => _baseStream = value;
{
if (_baseStream is null)
{
}
_baseStream = value;
}
} }
public IntPtr StreamHandle { get; set; }
public void Inject<T>(T dependency) public void Inject<T>(T dependency)
{ {
@@ -55,25 +50,28 @@ namespace mROA.Implementation
} }
} }
public Task<NetworkMessageHeader> GetNextMessageReceiving() public Task<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true)
{ {
if (!infinite) return Receive().AsTask();
if (_currentReceiving != null) return _currentReceiving; if (_currentReceiving != null) return _currentReceiving;
_currentReceiving = Task.Run(async () => await GetNextMessage()); _currentReceiving = Task.Run(async () => await GetNextMessage());
return _currentReceiving; return _currentReceiving;
} }
#pragma warning disable CS8602 // Dereference of a possibly null reference. #pragma warning disable CS8602 // Dereference of a possibly null reference.
private async ValueTask<bool> PostMessageInternal(NetworkMessageHeader messageHeader) private async ValueTask<bool> PostMessageInternal(NetworkMessageHeader messageHeader)
{ {
#if TRACE #if TRACE
Console.WriteLine($"{DateTime.Now.TimeOfDay} Posting message: {messageHeader.Id} - {messageHeader.MessageType} to {ConnectionId}"); Console.WriteLine(
$"{DateTime.Now.TimeOfDay} Posting message to {StreamHandle}: {messageHeader.Id} - {messageHeader.MessageType} to {ConnectionId}");
#endif #endif
var rawMessage = _serialization.Serialize(messageHeader); var rawMessage = _serialization.Serialize(messageHeader);
var header = BitConverter.GetBytes((ushort)rawMessage.Length).AsMemory(0, sizeof(ushort)); var header = BitConverter.GetBytes((ushort)rawMessage.Length).AsMemory(0, sizeof(ushort));
if (!_baseStream.CanWrite) if (!_baseStream.CanWrite)
return false; return false;
await BaseStream.WriteAsync(header); await BaseStream.WriteAsync(header);
await BaseStream.WriteAsync(rawMessage); await BaseStream.WriteAsync(rawMessage);
return true; return true;
@@ -103,24 +101,8 @@ namespace mROA.Implementation
break; break;
_isConnected = false; _isConnected = false;
withError = true;
await MakeRecovery("OUT"); await MakeRecovery("OUT");
// Console.WriteLine("Try to get lock from post");
// lock (_reconection)
// {
// Console.WriteLine("Got lock from post");
// if (!_isRecovering)
// {
// _isRecovering = true;
// Console.WriteLine("Disconnect invoke for post");
// OnDisconected?.Invoke(ConnectionId);
// Console.WriteLine("Disconnect invoked for post");
// }
// }
//
// withError = true;
// Console.WriteLine("Start waiting for recovery from post");
// _ = await _reconection.Task;
// Console.WriteLine("Connection recovered from post");
} }
} }
@@ -152,42 +134,26 @@ namespace mROA.Implementation
{ {
if (withError) if (withError)
{ {
Console.WriteLine("Recieve again"); Console.WriteLine("Receive again");
} }
try try
{ {
return await Receive(); var message = await Receive();
_currentReceiving = Task.Run(async () => await GetNextMessage());
return message;
} }
catch (Exception ex) catch (Exception ex)
{ {
withError = true;
await MakeRecovery("IN"); await MakeRecovery("IN");
// Console.WriteLine("Try to get lock from receive");
// lock (_reconection)
// {
// Console.WriteLine("Got lock from receive");
//
// if (!_isRecovering)
// {
// _isRecovering = true;
// Console.WriteLine("Disconnect invoke");
// OnDisconected?.Invoke(ConnectionId);
// Console.WriteLine("Disconnect invoked for receive");
//
// }
// }
//
// withError = true;
// Console.WriteLine("Start waiting for recovery from receive");
// _ = await _reconection.Task;
// Console.WriteLine("Connection recovered");
} }
} }
} }
private ushort ReadMessageLength() private ushort ReadMessageLength()
{ {
var firstBit = BaseStream.ReadByte(); var firstBit = BaseStream.ReadByte();
if (firstBit == -1) if (firstBit == -1)
{ {
@@ -212,13 +178,11 @@ namespace mROA.Implementation
var message = _serialization.Deserialize<NetworkMessageHeader>(localSpan.Span); var message = _serialization.Deserialize<NetworkMessageHeader>(localSpan.Span);
#if TRACE #if TRACE
Console.WriteLine($"{DateTime.Now.TimeOfDay} Received Message {message.Id} - {message.MessageType}"); Console.WriteLine($"{DateTime.Now.TimeOfDay} Received Message from {StreamHandle} {message.Id} - {message.MessageType}");
TransmissionConfig.TotalTransmittedBytes += len; TransmissionConfig.TotalTransmittedBytes += len;
Console.WriteLine($"Total received bytes are {TransmissionConfig.TotalTransmittedBytes}"); Console.WriteLine($"Total received bytes are {TransmissionConfig.TotalTransmittedBytes}");
#endif #endif
_messageBuffer.Add(message); _messageBuffer.Add(message);
_currentReceiving = Task.Run(async () => await GetNextMessage());
return message; return message;
} }
@@ -228,32 +192,42 @@ namespace mROA.Implementation
{ {
await PostMessageAsync( await PostMessageAsync(
new NetworkMessageHeader(_serialization!, new ClientRecovery(Math.Abs(ConnectionId)))); new NetworkMessageHeader(_serialization!, new ClientRecovery(Math.Abs(ConnectionId))));
var iTest = _baseStream.ReadByte();
var confirmByte = BaseStream.ReadByte(); var bTest = (byte)iTest;
Console.WriteLine("Reconnection byte {0}", confirmByte); _baseStream.WriteByte(bTest);
} }
else else
{ {
BaseStream.WriteByte(byte.MaxValue); const byte confirmByte = 128;
_baseStream.WriteByte(confirmByte);
var iPong = _baseStream.ReadByte();
var bPong = (byte)iPong;
if (confirmByte != bPong)
{
Console.WriteLine("Incorrect byte");
}
} }
Console.WriteLine("Setting result for reconnection"); Console.WriteLine("Setting result for reconnection");
_reconnection.SetResult(BaseStream!); var setting = _reconnection.TrySetResult(BaseStream);
Console.WriteLine("Set result for reconnection successfull"); _isInReconnectionState = false;
_isConnected = true;
Console.WriteLine($"Set result for reconnection {setting}");
_reconnection = new TaskCompletionSource<Stream>(); _reconnection = new TaskCompletionSource<Stream>();
} }
private async Task MakeRecovery(string source) private async Task MakeRecovery(string source)
{ {
Console.WriteLine("Staring recovery from {0}", source); Console.WriteLine("Staring recovery from {0}", source);
lock (_reconnection) lock (_reconnection)
{ {
Console.WriteLine("Got lock from {0}", source); Console.WriteLine("Got lock from {0}", source);
if (_isConnected || _isInReconnectionState) if (_isConnected || _isInReconnectionState)
{ {
Console.WriteLine($"{_isConnected} {_isInReconnectionState} {!_baseStream.CanRead} {!_baseStream.CanWrite}"); Console.WriteLine(
$"{source} {_isConnected} {_isInReconnectionState} {!_baseStream.CanRead} {!_baseStream.CanWrite}");
return; return;
} }
@@ -261,9 +235,14 @@ namespace mROA.Implementation
_isInReconnectionState = true; _isInReconnectionState = true;
OnDisconected?.Invoke(ConnectionId); OnDisconected?.Invoke(ConnectionId);
} }
Console.WriteLine("Waiting for reconnect from {0}", source); Console.WriteLine("Waiting for reconnect from {0}", source);
await _reconnection.Task; if (!_reconnection.Task.IsCompleted && !_isConnected)
{
Console.WriteLine("Current connection state {0} from {1}", _isConnected, source);
await _reconnection.Task;
}
Console.WriteLine("Reconnect finished from {0}", source); Console.WriteLine("Reconnect finished from {0}", source);
lock (_reconnection) lock (_reconnection)
{ {