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

Some strange breaking changes

Browse Source
This commit is contained in:
micialware
2025-05-02 19:17:11 +03:00
parent e6466b133b
commit 557a52f24c
10 changed files with 73 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mROA.Test/NextGenTest.cs
+2 -2
View File
@@ -33,7 +33,7 @@ namespace mROA.Test
Task.Run(() => Task.Run(() =>
{ {
_listener.Start(); _listener.Start();
_interactionModuleB.BaseStream = _listener.AcceptTcpClient().GetStream(); // _interactionModuleB.BaseStream = _listener.AcceptTcpClient().GetStream();
foreach (var guid in guids) foreach (var guid in guids)
{ {
@@ -43,7 +43,7 @@ namespace mROA.Test
var client = new TcpClient(); var client = new TcpClient();
client.Connect(IPAddress.Loopback, 4567); client.Connect(IPAddress.Loopback, 4567);
_interactionModuleA.BaseStream = client.GetStream(); // _interactionModuleA.BaseStream = client.GetStream();
var tasks = guids.Select(ReadStream); var tasks = guids.Select(ReadStream);
mROA/Abstract/IInteractionModule.cs → mROA/Abstract/IChannelInteractionModule.cs
+2 -4
View File
@@ -9,15 +9,13 @@ namespace mROA.Abstract
public interface IChannelInteractionModule : IInjectableModule, IDisposable public interface IChannelInteractionModule : IInjectableModule, IDisposable
{ {
int ConnectionId { get; set; } int ConnectionId { get; set; }
ChannelWriter<NetworkMessageHeader> ReceiveChanel { get; } Channel<NetworkMessageHeader> ReceiveChanel { get; }
ChannelReader<NetworkMessageHeader> TrustedPostChanel { get; } ChannelReader<NetworkMessageHeader> TrustedPostChanel { get; }
ChannelReader<NetworkMessageHeader> UntrustedPostChanel { get; } ChannelReader<NetworkMessageHeader> UntrustedPostChanel { get; }
Func<bool> IsConnected { get; set; } Func<bool> IsConnected { get; set; }
Task<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true); ValueTask<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true);
Task PostMessageAsync(NetworkMessageHeader messageHeader); Task PostMessageAsync(NetworkMessageHeader messageHeader);
Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader); Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader);
void HandleMessage(NetworkMessageHeader messageHeader);
NetworkMessageHeader? FirstByFilter(Predicate<NetworkMessageHeader> predicate);
event Action<int> OnDisconnected; event Action<int> OnDisconnected;
Task Restart(bool sendRecovery); Task Restart(bool sendRecovery);
} }
mROA/Abstract/IRepresentationModule.cs
+2 -1
View File
@@ -10,7 +10,8 @@ namespace mROA.Abstract
{ {
int Id { get; } int Id { get; }
Task<(object parced, EMessageType originalType)> GetSingle(Predicate<NetworkMessageHeader> rule, CancellationToken token, Task<(object? Deserialized, EMessageType MessageType)> GetSingle(Predicate<NetworkMessageHeader> rule,
CancellationToken token,
params Func<NetworkMessageHeader, Type?>[] converter); params Func<NetworkMessageHeader, Type?>[] converter);
IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate<NetworkMessageHeader> rule, CancellationToken token, IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate<NetworkMessageHeader> rule, CancellationToken token,
mROA/Implementation/Backend/NetworkGatewayModule.cs
+2 -4
View File
@@ -78,7 +78,7 @@ namespace mROA.Implementation.Backend
var streamExtractor = new StreamExtractor(client.GetStream(), _serialization); var streamExtractor = new StreamExtractor(client.GetStream(), _serialization);
interaction.IsConnected = () => streamExtractor.IsConnected; interaction.IsConnected = () => streamExtractor.IsConnected;
streamExtractor.MessageReceived += message => interaction.ReceiveChanel.WriteAsync(message); streamExtractor.MessageReceived = message => interaction.ReceiveChanel.Writer.WriteAsync(message);
streamExtractor.SingleReceive(); streamExtractor.SingleReceive();
var connectionRequest = interaction.GetNextMessageReceiving(false) var connectionRequest = interaction.GetNextMessageReceiving(false)
.GetAwaiter().GetResult()!; .GetAwaiter().GetResult()!;
@@ -98,10 +98,8 @@ 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);
streamExtractor = new StreamExtractor(client.GetStream(), _serialization); streamExtractor.MessageReceived = message => recoveryInteraction.ReceiveChanel.Writer.WriteAsync(message);
streamExtractor.MessageReceived += message => recoveryInteraction.ReceiveChanel.WriteAsync(message);
_ = streamExtractor.LoopedReceive();
recoveryInteraction.Restart(false); recoveryInteraction.Restart(false);
Console.WriteLine("Connection recovery for client {0} finished", recoveryRequest.Id); Console.WriteLine("Connection recovery for client {0} finished", recoveryRequest.Id);
break; break;
mROA/Implementation/ChannelInteractionModule.cs
+10 -10
View File
@@ -2,7 +2,6 @@
using System.Collections.Generic; using System.Collections.Generic;
using System.IO; using System.IO;
using System.Linq; using System.Linq;
using System.Threading;
using System.Threading.Channels; using System.Threading.Channels;
using System.Threading.Tasks; using System.Threading.Tasks;
using mROA.Abstract; using mROA.Abstract;
@@ -14,7 +13,6 @@ namespace mROA.Implementation
private readonly ChannelReader<NetworkMessageHeader> _receiveReader; private readonly ChannelReader<NetworkMessageHeader> _receiveReader;
private readonly ChannelWriter<NetworkMessageHeader> _trustedWriter; private readonly ChannelWriter<NetworkMessageHeader> _trustedWriter;
private readonly ChannelWriter<NetworkMessageHeader> _untrustedWriter; private readonly ChannelWriter<NetworkMessageHeader> _untrustedWriter;
private readonly Channel<NetworkMessageHeader> _inputChannel;
private readonly Channel<NetworkMessageHeader> _outputTrustedChannel; private readonly Channel<NetworkMessageHeader> _outputTrustedChannel;
private readonly Channel<NetworkMessageHeader> _outputUntrustedChannel; private readonly Channel<NetworkMessageHeader> _outputUntrustedChannel;
private readonly List<NetworkMessageHeader> _messageBuffer = new(128); private readonly List<NetworkMessageHeader> _messageBuffer = new(128);
@@ -26,13 +24,13 @@ namespace mROA.Implementation
public ChannelInteractionModule() public ChannelInteractionModule()
{ {
_inputChannel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions ReceiveChanel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
{ {
SingleReader = false, SingleReader = false,
SingleWriter = false, SingleWriter = false,
AllowSynchronousContinuations = true AllowSynchronousContinuations = true
}); });
_receiveReader = _inputChannel.Reader; _receiveReader = ReceiveChanel.Reader;
_outputTrustedChannel = Channel.CreateBounded<NetworkMessageHeader>(new BoundedChannelOptions(1) _outputTrustedChannel = Channel.CreateBounded<NetworkMessageHeader>(new BoundedChannelOptions(1)
{ {
SingleReader = true, SingleReader = true,
@@ -52,7 +50,8 @@ namespace mROA.Implementation
public int ConnectionId { get; set; } public int ConnectionId { get; set; }
public ChannelWriter<NetworkMessageHeader> ReceiveChanel => _inputChannel.Writer; public Channel<NetworkMessageHeader> ReceiveChanel { get; }
public ChannelReader<NetworkMessageHeader> TrustedPostChanel => _outputTrustedChannel.Reader; public ChannelReader<NetworkMessageHeader> TrustedPostChanel => _outputTrustedChannel.Reader;
public ChannelReader<NetworkMessageHeader> UntrustedPostChanel => _outputUntrustedChannel.Reader; public ChannelReader<NetworkMessageHeader> UntrustedPostChanel => _outputUntrustedChannel.Reader;
public Func<bool> IsConnected { get; set; } public Func<bool> IsConnected { get; set; }
@@ -71,12 +70,12 @@ namespace mROA.Implementation
} }
} }
public Task<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true) public ValueTask<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true)
{ {
if (!infinite) return _receiveReader.ReadAsync().AsTask(); return _receiveReader.ReadAsync();
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)
@@ -192,6 +191,7 @@ namespace mROA.Implementation
private async Task MakeRecovery(string source) private async Task MakeRecovery(string source)
{ {
//TODO переделать реконнект //TODO переделать реконнект
// Console.WriteLine("Staring recovery from {0}", source); // Console.WriteLine("Staring recovery from {0}", source);
// //
// lock (_reconnection) // lock (_reconnection)
mROA/Implementation/Frontend/NetworkFrontendBridge.cs
+1 -1
View File
@@ -78,7 +78,7 @@ namespace mROA.Implementation.Frontend
_ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel, _ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel,
_rawExtractorCancellation.Token); _rawExtractorCancellation.Token);
_currentExtractor.MessageReceived += message => _interactionModule.ReceiveChanel.WriteAsync(message); _currentExtractor.MessageReceived = message => _interactionModule.ReceiveChanel.Writer.WriteAsync(message);
} }
private async Task Reconnect() private async Task Reconnect()
mROA/Implementation/Frontend/RequestExtractor.cs
+14 -11
View File
@@ -47,6 +47,9 @@ namespace mROA.Implementation.Frontend
public async Task StartExtraction() public async Task StartExtraction()
{ {
ThrowIfNotInjected(); ThrowIfNotInjected();
TransmissionConfig.OwnershipRepository = new StaticOwnershipRepository(_representationModule.Id);
var multiClientOwnershipRepository = var multiClientOwnershipRepository =
TransmissionConfig.OwnershipRepository as MultiClientOwnershipRepository; TransmissionConfig.OwnershipRepository as MultiClientOwnershipRepository;
multiClientOwnershipRepository?.RegisterOwnership(_representationModule!.Id); multiClientOwnershipRepository?.RegisterOwnership(_representationModule!.Id);
@@ -54,11 +57,11 @@ namespace mROA.Implementation.Frontend
try try
{ {
#if TRACE #if TRACE
var sw = new Stopwatch(); var sw = new Stopwatch();
#endif #endif
var streamTokenSource = new CancellationTokenSource(); var streamTokenSource = new CancellationTokenSource();
var query = _representationModule!.GetStream(m => var query = _representationModule!.GetStream(m =>
m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest
or EMessageType.EventRequest or EMessageType.ClientDisconnect, streamTokenSource.Token, or EMessageType.EventRequest or EMessageType.ClientDisconnect, streamTokenSource.Token,
@@ -71,17 +74,18 @@ namespace mROA.Implementation.Frontend
await foreach (var command in query) await foreach (var command in query)
{ {
#if TRACE #if TRACE
Console.WriteLine("Waiting for request..."); Console.WriteLine("Waiting for request...");
if (sw.IsRunning) if (sw.IsRunning)
{ {
sw.Stop(); sw.Stop();
Console.WriteLine($"Request handling took {Math.Round(sw.Elapsed.TotalMilliseconds * 1000.0)} microseconds."); Console.WriteLine(
} $"Request handling took {Math.Round(sw.Elapsed.TotalMilliseconds * 1000.0)} microseconds.");
}
#endif #endif
#if TRACE #if TRACE
Console.WriteLine("Request received"); Console.WriteLine("Request received");
sw.Restart(); sw.Restart();
#endif #endif
switch (command.originalType) switch (command.originalType)
{ {
@@ -95,7 +99,6 @@ namespace mROA.Implementation.Frontend
break; break;
case EMessageType.CancelRequest: case EMessageType.CancelRequest:
HandleCancelRequest((command.parced as CancelRequest)!); HandleCancelRequest((command.parced as CancelRequest)!);
break; break;
default: default:
continue; continue;
mROA/Implementation/RemoteObjectBase.cs
+6 -6
View File
@@ -82,14 +82,14 @@ namespace mROA.Implementation
var response = await responseRequestTask; var response = await responseRequestTask;
if (response.parced is FinalCommandExecution<T> successResponse) if (response.Deserialized is FinalCommandExecution<T> successResponse)
{ {
localTokenSource.Cancel(); localTokenSource.Cancel();
return successResponse.Result!; return successResponse.Result!;
} }
localTokenSource.Cancel(); localTokenSource.Cancel();
throw (response.parced as ExceptionCommandExecution)!.GetException(); throw (response.Deserialized as ExceptionCommandExecution)!.GetException();
} }
protected async Task CallAsync(int methodId, object?[]? parameters = null, protected async Task CallAsync(int methodId, object?[]? parameters = null,
@@ -103,7 +103,7 @@ namespace mROA.Implementation
var localTokenSource = new CancellationTokenSource(); var localTokenSource = new CancellationTokenSource();
var responceRequestTask = _representationModule.GetSingle( var responseRequestTask = _representationModule.GetSingle(
m => m.MessageType is EMessageType.FinishedCommandExecution or EMessageType.ExceptionCommandExecution, m => m.MessageType is EMessageType.FinishedCommandExecution or EMessageType.ExceptionCommandExecution,
localTokenSource.Token, localTokenSource.Token,
m => m.MessageType is EMessageType.FinishedCommandExecution ? typeof(FinalCommandExecution) : null, m => m.MessageType is EMessageType.FinishedCommandExecution ? typeof(FinalCommandExecution) : null,
@@ -124,16 +124,16 @@ namespace mROA.Implementation
}).ContinueWith(_ => localTokenSource.Cancel()); }).ContinueWith(_ => localTokenSource.Cancel());
}); });
var responseRequest = await responceRequestTask; var responseRequest = await responseRequestTask;
#if TRACE #if TRACE
Console.WriteLine($"Handling message"); Console.WriteLine($"Handling message");
#endif #endif
switch (responseRequest.originalType) switch (responseRequest.MessageType)
{ {
case EMessageType.FinishedCommandExecution: case EMessageType.FinishedCommandExecution:
return; return;
case EMessageType.ExceptionCommandExecution: case EMessageType.ExceptionCommandExecution:
throw (responseRequest.parced as ExceptionCommandExecution)!.GetException(); throw (responseRequest.Deserialized as ExceptionCommandExecution)!.GetException();
} }
} }
mROA/Implementation/RepresentationModule.cs
+33 -55
View File
@@ -1,5 +1,7 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using mROA.Abstract; using mROA.Abstract;
@@ -24,72 +26,48 @@ namespace mROA.Implementation
} }
} }
public int Id => (_interaction ?? throw new NullReferenceException("Interaction is not initialized")) public int Id => (_interaction ?? throw new NullReferenceException("Interaction is not initialized"))
.ConnectionId; .ConnectionId;
public Task<(object parced, EMessageType originalType)> GetSingle(Predicate<NetworkMessageHeader> rule, CancellationToken token, params Func<NetworkMessageHeader, Type?>[] converter) public async Task<(object? Deserialized, EMessageType MessageType)> GetSingle(
Predicate<NetworkMessageHeader> rule,
CancellationToken token = default, params Func<NetworkMessageHeader, Type?>[] converter)
{ {
throw new NotImplementedException(); var writer = _interaction.ReceiveChanel.Writer;
} await foreach (var message in _interaction.ReceiveChanel.Reader.ReadAllAsync(token))
public IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate<NetworkMessageHeader> rule, CancellationToken token, params Func<NetworkMessageHeader, Type?>[] converter)
{
throw new NotImplementedException();
}
public async Task<T> GetMessageAsync<T>(Guid? requestId, EMessageType? messageType,
CancellationToken token = default)
{
if (_serialization == null)
throw new NullReferenceException("Serialization toolkit is not initialized");
var rawMessage = await GetRawMessage(requestId, messageType, token);
return _serialization.Deserialize<T>(rawMessage)!;
}
public T GetMessage<T>(Guid? requestId = null, EMessageType? messageType = null)
{
if (_serialization == null)
throw new NullReferenceException("Serialization toolkit is not initialized");
var rawMessage = GetRawMessage(requestId, messageType).GetAwaiter().GetResult();
return _serialization.Deserialize<T>(rawMessage)!;
}
public async Task<byte[]> GetRawMessage(Guid? requestId = null, EMessageType? messageType = null,
CancellationToken token = default)
{
if (_interaction == null)
throw new NullReferenceException("Interaction toolkit is not initialized");
var fromBuffer =
_interaction.FirstByFilter(message =>
(requestId is null || message.Id == requestId) &&
(messageType is null || message.MessageType == messageType));
if (fromBuffer == null)
{ {
while (token.IsCancellationRequested == false) if (!rule(message))
{ {
var message = await _interaction.GetNextMessageReceiving(); await writer.WriteAsync(message, token);
if ((requestId is not null && message.Id != requestId) || continue;
(messageType is not null && message.MessageType != messageType))
continue;
_interaction.HandleMessage(message);
return message.Data;
} }
var type = converter.Select(i => i(message)).First(i => i != null)!;
var deserialized = _serialization.Deserialize(message.Data, type);
return (deserialized, message.MessageType);
} }
if (fromBuffer == null) return (null, EMessageType.Unknown);
}
public async IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(
Predicate<NetworkMessageHeader> rule, [EnumeratorCancellation] CancellationToken token = default,
params Func<NetworkMessageHeader, Type?>[] converter)
{
var writer = _interaction?.ReceiveChanel.Writer;
await foreach (var message in _interaction.ReceiveChanel.Reader.ReadAllAsync(token))
{ {
return Array.Empty<byte>(); if (!rule(message))
} {
await writer.WriteAsync(message, token);
continue;
}
_interaction.HandleMessage(fromBuffer); var type = converter.Select(i => i(message)).First(i => i != null)!;
return fromBuffer.Data; var deserialized = _serialization.Deserialize(message.Data, type);
yield return (deserialized, message.MessageType)!;
}
} }
public async Task PostCallMessageAsync<T>(Guid id, EMessageType eMessageType, T payload) where T : notnull public async Task PostCallMessageAsync<T>(Guid id, EMessageType eMessageType, T payload) where T : notnull
mROA/Implementation/StreamExtractor.cs
+1 -1
View File
@@ -21,7 +21,7 @@ namespace mROA.Implementation
_serializationToolkit = serializationToolkit; _serializationToolkit = serializationToolkit;
} }
public event Action<NetworkMessageHeader> MessageReceived; public Action<NetworkMessageHeader> MessageReceived = _ => { };
private ushort ReadMessageLength() private ushort ReadMessageLength()
{ {