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

Refactoring and prepare for interaction module reconnection

Browse Source
This commit is contained in:
micialware
2025-04-08 11:48:46 +03:00
parent 0f5bb18787
commit 3a4456e362
13 changed files with 143 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Example.Backend/LoadTestImp.cs
+1
View File
@@ -40,6 +40,7 @@ namespace Example.Backend
Console.WriteLine("Waiting canceled");
return;
}
Console.WriteLine("Waiting...");
await Task.Delay(1000);
}
Example.Backend/Page.cs
+1
View File
@@ -6,6 +6,7 @@ namespace Example.Backend
public class Page : IPage
{
public string Text;
public byte[] GetData()
{
return Encoding.UTF8.GetBytes(Text);
Example.Frontend/Program.cs
+6 -2
View File
@@ -39,14 +39,14 @@ class Program
TransmissionConfig.RealContextRepository = builder.GetModule<ContextRepository>();
TransmissionConfig.RemoteEndpointContextRepository = builder.GetModule<RemoteContextRepository>();
builder.GetModule<IFrontendBridge>()!.Connect();
var frontendBridge = builder.GetModule<IFrontendBridge>()!;
frontendBridge.Connect();
_ = builder.GetModule<RequestExtractor>()!.StartExtraction();
Console.WriteLine(TransmissionConfig.OwnershipRepository.GetOwnershipId());
var context = builder.GetModule<RemoteContextRepository>();
var factory = context.GetSingleObject(typeof(IPrinterFactory), 0) as IPrinterFactory;
//правильный порядок команд 8-5-10-7
using (var disposingPrinter = factory.Create("Test"))
{
DemoCheck.CreatingPrinter = true;
@@ -58,6 +58,7 @@ class Program
Console.WriteLine("Printer created");
Thread.Sleep(100);
// frontendBridge.Obstacle();
var name = disposingPrinter.GetName();
DemoCheck.BasicNonParamsCall = true;
Console.WriteLine("Printer name : {0}", name);
@@ -114,6 +115,9 @@ class Program
cts.Cancel();
Console.WriteLine($"Token state {cts.Token.IsCancellationRequested}");
DemoCheck.TaskCancelation = true;
frontendBridge.Disconnect();
DemoCheck.Show();
Console.ReadKey();
//
mROA/Abstract/IContextRepositoryHub.cs
+1
View File
@@ -3,5 +3,6 @@ namespace mROA.Abstract
public interface IContextRepositoryHub
{
IContextRepository GetRepository(int clientId);
void FreeRepository(int clientId);
}
}
mROA/Abstract/IInteractionModule.cs
+1 -1
View File
@@ -15,7 +15,7 @@ namespace mROA.Abstract
NetworkMessageHeader[] UnhandledMessages { get; }
NetworkMessageHeader? FirstByFilter(Predicate<NetworkMessageHeader> predicate);
event Action<int> OnDisconected;
Task Restart();
Task Restart(bool sendRecovery);
}
}
mROA/Implementation/Backend/ContextRepository.cs
-4
View File
@@ -10,15 +10,11 @@ namespace mROA.Implementation.Backend
{
public class ContextRepository : IContextRepository
{
private const int StartupSize = 1024;
private const int GrowSize = 128;
public static object[] EventBinders = { };
private static int LastDebugId = -1;
private int _debugId = -1;
private Task<int> _lastIndexFinder = Task.FromResult(0);
private IRepresentationModuleProducer? _representationModuleProducer;
// [CanBeNull]
mROA/Implementation/Backend/HubRequestExtractor.cs
+7 -1
View File
@@ -49,7 +49,13 @@ namespace mROA.Implementation.Backend
private void HubOnOnConnected(IRepresentationModule interaction)
{
var extractor = CreateExtractor(interaction);
_ = extractor.StartExtraction();
extractor.StartExtraction().ContinueWith(t => OnDisconnected(interaction));
}
private void OnDisconnected(IRepresentationModule representationModule)
{
if (_contextRepository is IContextRepositoryHub contextHub)
contextHub.FreeRepository(representationModule.Id);
}
private IRequestExtractor CreateExtractor(IRepresentationModule interaction)
mROA/Implementation/Backend/MultiClientContextRepository.cs
+5
View File
@@ -56,6 +56,11 @@ namespace mROA.Implementation.Backend
return repository;
}
public void FreeRepository(int clientId)
{
_repositories.Remove(clientId);
}
private IContextRepository GetRepositoryByClientId(int clientId)
{
if (_repositories.TryGetValue(clientId, out var repository))
mROA/Implementation/Backend/NetworkGatewayModule.cs
+20 -12
View File
@@ -66,6 +66,7 @@ namespace mROA.Implementation.Backend
while (true)
{
var client = _tcpListener.AcceptTcpClient();
Console.WriteLine($"Client connected from {client.Client.RemoteEndPoint}");
var interaction = Activator.CreateInstance(_interactionModuleType!) as INextGenerationInteractionModule;
foreach (var injectableModule in _injectableModules!)
@@ -77,19 +78,26 @@ namespace mROA.Implementation.Backend
var connectionRequest = interaction.GetNextMessageReceiving().GetAwaiter().GetResult()!;
if (connectionRequest.MessageType == EMessageType.ClientConnect)
switch (connectionRequest.MessageType)
{
interaction.PostMessageAsync(new NetworkMessageHeader(_serialization!,
new IdAssignment { Id = -interaction.ConnectionId }));
_hub!.RegisterInteraction(interaction);
Console.WriteLine("Client registered");
}else if (connectionRequest.MessageType == EMessageType.ClientRecovery)
{
var recoveryRequest = _serialization!.Deserialize<ClientRecovery>(connectionRequest.Data)!;
var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id);
recoveryInteraction.BaseStream = client.GetStream();
recoveryInteraction.Restart();
Console.WriteLine($"Client {recoveryRequest.Id} reconnected");
case EMessageType.ClientConnect:
interaction.PostMessageAsync(new NetworkMessageHeader(_serialization!,
new IdAssignment { Id = -interaction.ConnectionId }));
_hub!.RegisterInteraction(interaction);
Console.WriteLine("Client registered");
break;
case EMessageType.ClientRecovery:
{
var recoveryRequest = _serialization!.Deserialize<ClientRecovery>(connectionRequest.Data)!;
var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id);
recoveryInteraction.BaseStream = client.GetStream();
recoveryInteraction.Restart(false);
Console.WriteLine($"Client {recoveryRequest.Id} reconnected");
break;
}
default:
client.Close();
break;
}
}
}
mROA/Implementation/Frontend/NetworkFrontendBridge.cs
+3 -3
View File
@@ -43,9 +43,9 @@ namespace mROA.Implementation.Frontend
_interactionModule.BaseStream = _tcpClient.GetStream();
_interactionModule.OnDisconected += async id =>
_interactionModule.OnDisconected += id =>
{
await Reconnect();
Reconnect();
};
_ = _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientConnect()));
var welcomeMessage = _interactionModule.GetNextMessageReceiving().GetAwaiter().GetResult();
@@ -66,7 +66,7 @@ namespace mROA.Implementation.Frontend
_tcpClient = new TcpClient();
_tcpClient.Connect(_serverEndPoint);
_interactionModule.BaseStream = _tcpClient.GetStream();
await _interactionModule.Restart();
await _interactionModule.Restart(true);
}
public void Obstacle()
mROA/Implementation/IdAssignment.cs
+4
View File
@@ -9,6 +9,10 @@ namespace mROA.Implementation
public class ClientRecovery : INetworkMessage
{
public ClientRecovery()
{
Id = 0;
}
public ClientRecovery(int id)
{
Id = id;
mROA/Implementation/NextGenerationInteractionModule.cs
+92 -12
View File
@@ -10,14 +10,23 @@ namespace mROA.Implementation
{
public class NextGenerationInteractionModule : INextGenerationInteractionModule
{
private int DebugId = new Random().Next();
private const int BufferSize = ushort.MaxValue;
private readonly Memory<byte> _buffer = new byte[BufferSize];
private readonly List<NetworkMessageHeader> _messageBuffer = new(128);
private Task<NetworkMessageHeader>? _currentReceiving;
private ISerializationToolkit? _serialization;
private Stream? _baseStream;
private bool _isRecovering;
private event Action OnReconnected;
private TaskCompletionSource<Stream> _reconection = new();
private TaskCompletionSource<Stream> _reconection;
public NextGenerationInteractionModule()
{
_reconection = new TaskCompletionSource<Stream>();
_reconection.SetResult(Stream.Null);
}
public int ConnectionId { get; set; }
@@ -54,6 +63,25 @@ namespace mROA.Implementation
_currentReceiving = Task.Run(async () => await GetNextMessage());
return _currentReceiving;
}
#pragma warning disable CS8602 // Dereference of a possibly null reference.
private async ValueTask<bool> PostMessageInternal(NetworkMessageHeader messageHeader)
{
#if TRACE
Console.WriteLine($"{DateTime.Now.TimeOfDay} Posting message: {messageHeader.Id} - {messageHeader.MessageType} to {ConnectionId}");
#endif
var rawMessage = _serialization.Serialize(messageHeader);
var header = BitConverter.GetBytes((ushort)rawMessage.Length).AsMemory(0, sizeof(ushort));
if (!_baseStream.CanWrite)
return false;
await BaseStream.WriteAsync(header);
await BaseStream.WriteAsync(rawMessage);
return true;
}
#pragma warning restore CS8602 // Dereference of a possibly null reference.
public async Task PostMessageAsync(NetworkMessageHeader messageHeader)
{
@@ -65,12 +93,35 @@ namespace mROA.Implementation
// Console.WriteLine("Sending {0}", JsonSerializer.Serialize(message));
bool withError = false;
while (true)
{
if (withError)
{
Console.WriteLine("Post again");
}
var rawMessage = _serialization.Serialize(messageHeader);
var header = BitConverter.GetBytes((ushort)rawMessage.Length).AsMemory(0, sizeof(ushort));
if (await PostMessageInternal(messageHeader))
break;
await BaseStream.WriteAsync(header);
await BaseStream.WriteAsync(rawMessage);
// 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");
}
}
public void HandleMessage(NetworkMessageHeader messageHeader)
@@ -95,16 +146,40 @@ namespace mROA.Implementation
if (_serialization == null)
throw new NullReferenceException("Serialization toolkit is null");
bool withError = false;
while (true)
{
if (withError)
{
Console.WriteLine("Recieve again");
}
try
{
return await Receive();
}
catch (Exception)
{
OnDisconected!.Invoke(ConnectionId);
_ = await _reconection.Task;
// 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");
}
}
}
@@ -119,7 +194,7 @@ namespace mROA.Implementation
return len;
}
private async Task<NetworkMessageHeader> Receive()
private async ValueTask<NetworkMessageHeader> Receive()
{
var len = ReadMessageLength();
var localSpan = _buffer[..len];
@@ -128,9 +203,9 @@ namespace mROA.Implementation
var message = _serialization.Deserialize<NetworkMessageHeader>(localSpan.Span);
#if TRACE
Console.WriteLine($"{DateTime.Now.TimeOfDay} Received Message {message.Id} - {message.SchemaId}");
Console.WriteLine($"{DateTime.Now.TimeOfDay} Received Message {message.Id} - {message.MessageType}");
TransmissionConfig.TotalTransmittedBytes += len;
Console.WriteLine($"Total recieced bytes are {TransmissionConfig.TotalTransmittedBytes}");
Console.WriteLine($"Total received bytes are {TransmissionConfig.TotalTransmittedBytes}");
#endif
_messageBuffer.Add(message);
_currentReceiving = Task.Run(async () => await GetNextMessage());
@@ -138,10 +213,15 @@ namespace mROA.Implementation
return message;
}
public async Task Restart()
public async Task Restart(bool sendRecovery)
{
await PostMessageAsync(new NetworkMessageHeader(_serialization!, new ClientRecovery(Math.Abs(ConnectionId))));
if (sendRecovery)
await PostMessageAsync(
new NetworkMessageHeader(_serialization!, new ClientRecovery(Math.Abs(ConnectionId))));
_isRecovering = false;
_reconection.SetResult(BaseStream!);
_reconection = new TaskCompletionSource<Stream>();
}
}
}
mROA/mROA.csproj
+1 -1
View File
@@ -21,7 +21,7 @@
</PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)' == 'Debug' ">
<DefineConstants></DefineConstants>
<DefineConstants>TRACE;</DefineConstants>
</PropertyGroup>
<ItemGroup>