diff --git a/Example.Frontend/Program.cs b/Example.Frontend/Program.cs index fdbd77d..1f73baa 100644 --- a/Example.Frontend/Program.cs +++ b/Example.Frontend/Program.cs @@ -71,6 +71,7 @@ var x = 0; for (int i = 0; i < iterations; i++) { x = loadSingleton.Next(x); + // Console.WriteLine(x); } timer.Stop(); diff --git a/mROA.Test/NextGenTest.cs b/mROA.Test/NextGenTest.cs index 7293877..99025f3 100644 --- a/mROA.Test/NextGenTest.cs +++ b/mROA.Test/NextGenTest.cs @@ -31,7 +31,6 @@ public class NextGenTest { _listener.Start(); _interactionModuleB.BaseStream = _listener.AcceptTcpClient().GetStream(); - foreach (var guid in guids) { _interactionModuleB.PostMessage(new NetworkMessage { Id = guid, Data = "Hello user"u8.ToArray() }); diff --git a/mROA/Abstract/IInteractionModule.cs b/mROA/Abstract/IInteractionModule.cs index 83f6ee5..9068ba1 100644 --- a/mROA/Abstract/IInteractionModule.cs +++ b/mROA/Abstract/IInteractionModule.cs @@ -5,7 +5,7 @@ namespace mROA.Abstract; public interface INextGenerationInteractionModule : IInjectableModule { int ConnectionId { get; } - public Stream? BaseStream { get; set; } + public Stream BaseStream { get; set; } NetworkMessage[] UnhandledMessages { get; } NetworkMessage LastMessage { get; } EventWaitHandle CurrentReceivingHandle { get; } diff --git a/mROA/Abstract/ISerialisationModule.cs b/mROA/Abstract/ISerialisationModule.cs index 14ea0ef..d18e78e 100644 --- a/mROA/Abstract/ISerialisationModule.cs +++ b/mROA/Abstract/ISerialisationModule.cs @@ -1,3 +1,4 @@ +using System.Windows.Input; using mROA.Implementation; using mROA.Implementation.CommandExecution; @@ -20,9 +21,10 @@ public interface ISerialisationModule : IInjectableModule public interface IRepresentationModule : IInjectableModule { int Id { get; } - Task GetMessageAsync(Guid? requestId = null, MessageType? messageType = null); + Task GetMessageAsync(Guid? requestId = null, MessageType? messageType = null, CancellationToken token = default); T GetMessage(Guid? requestId = null, MessageType? messageType = null); - Task GetRawMessage(Guid? requestId = null, MessageType? messageType = null); + T GetMessage(Predicate filter); + Task GetRawMessage(Predicate filter, CancellationToken token = default); Task PostCallMessageAsync(Guid id, MessageType messageType, T payload) where T : notnull; Task PostCallMessageAsync(Guid id, MessageType messageType, object payload, Type payloadType); diff --git a/mROA/Implementation/Backend/NetworkGatewayModule.cs b/mROA/Implementation/Backend/NetworkGatewayModule.cs index c62cfc1..5ae5206 100644 --- a/mROA/Implementation/Backend/NetworkGatewayModule.cs +++ b/mROA/Implementation/Backend/NetworkGatewayModule.cs @@ -68,7 +68,7 @@ public class NetworkGatewayModule : IGatewayModule interaction!.Inject(_serialization); interaction.BaseStream = client.GetStream(); - + interaction.StartInfiniteReceiving(); interaction.PostMessage(new NetworkMessage { Id = Guid.NewGuid(), SchemaId = MessageType.IdAssigning, diff --git a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs index a5f4a16..5c1920b 100644 --- a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs +++ b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs @@ -32,12 +32,16 @@ public class NetworkFrontendBridge(IPEndPoint ipEndPoint) : IFrontendBridge _tcpClient.Connect(ipEndPoint); _interactionModule.BaseStream = _tcpClient.GetStream(); - var welcomeMessage = _interactionModule.GetNextMessageReceiving().GetAwaiter().GetResult(); + _interactionModule.StartInfiniteReceiving(); + + var handle = _interactionModule.CurrentReceivingHandle; + handle.WaitOne(); + var welcomeMessage = _interactionModule.LastMessage; if (welcomeMessage.SchemaId != MessageType.IdAssigning) { throw new Exception($"Incorrect message type. Must be IdAssigning, current : {welcomeMessage.SchemaId.ToString()}"); } - + _interactionModule.HandleMessage(welcomeMessage); TransmissionConfig.OwnershipRepository = new StaticOwnershipRepository(_serialization.Deserialize(welcomeMessage.Data)!.Id); } } \ No newline at end of file diff --git a/mROA/Implementation/Frontend/RequestExtractor.cs b/mROA/Implementation/Frontend/RequestExtractor.cs index 486a61b..7da79bd 100644 --- a/mROA/Implementation/Frontend/RequestExtractor.cs +++ b/mROA/Implementation/Frontend/RequestExtractor.cs @@ -56,13 +56,13 @@ public class RequestExtractor : IRequestExtractor try { + var lastCommandId = Guid.Empty; while (true) { - var request = - _representationModule!.GetMessage(messageType: MessageType.CallRequest); - - // Console.WriteLine("Executing {0}", request.Id); + var request = + _representationModule!.GetMessage(m => m.Id != lastCommandId && m.SchemaId == MessageType.CallRequest); + lastCommandId = request.Id; if (request.Parameter is not null) { var parameterType = _methodRepository!.GetMethod(request.CommandId).GetParameters().First() diff --git a/mROA/Implementation/NetworkMessage.cs b/mROA/Implementation/NetworkMessage.cs index 11f8035..5571251 100644 --- a/mROA/Implementation/NetworkMessage.cs +++ b/mROA/Implementation/NetworkMessage.cs @@ -1,19 +1,32 @@ using System.Text.Json.Serialization; + // ReSharper disable UnusedMember.Global namespace mROA.Implementation; public class NetworkMessage { - public static readonly NetworkMessage Null = new() {SchemaId = MessageType.Unknown, Id = Guid.Empty, Data = []}; + public static readonly NetworkMessage Null = new() { SchemaId = MessageType.Unknown, Id = Guid.Empty, Data = [] }; public Guid Id { get; init; } + [JsonConverter(typeof(JsonStringEnumConverter))] public MessageType SchemaId { get; init; } public required byte[] Data { get; init; } + + public bool IsValidMessage(Guid? requestId = null, MessageType? messageType = null) + { + return (requestId is null || Id == requestId) && + (messageType is null || SchemaId == messageType); + } } public enum MessageType { - Unknown, FinishedCommandExecution, ExceptionCommandExecution, AsyncCancelCommandExecution, CallRequest, IdAssigning + Unknown, + FinishedCommandExecution, + ExceptionCommandExecution, + AsyncCancelCommandExecution, + CallRequest, + IdAssigning } \ No newline at end of file diff --git a/mROA/Implementation/NextGenerationInteractionModule.cs b/mROA/Implementation/NextGenerationInteractionModule.cs index e96ed15..a02183c 100644 --- a/mROA/Implementation/NextGenerationInteractionModule.cs +++ b/mROA/Implementation/NextGenerationInteractionModule.cs @@ -6,14 +6,14 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule { private ISerializationToolkit? _serialization; public int ConnectionId { get; private set; } - public Stream? BaseStream { get; set; } + public Stream BaseStream { get; set; } = Stream.Null; private Task? _currentReceiving; private const int BufferSize = ushort.MaxValue; private readonly Memory _buffer = new byte[BufferSize]; private readonly List _messageBuffer = new (128); public NetworkMessage[] UnhandledMessages => _messageBuffer.ToArray(); public NetworkMessage LastMessage { get; private set; } = NetworkMessage.Null; - public EventWaitHandle CurrentReceivingHandle { get; private set; } = new(false, EventResetMode.ManualReset); + public EventWaitHandle CurrentReceivingHandle { get; private set; } = new(true, EventResetMode.ManualReset); public void Inject(T dependency) { @@ -31,12 +31,23 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule public async void StartInfiniteReceiving() { - while (true) + try { - var message = ReceiveMessage(); - LastMessage = message; - CurrentReceivingHandle.Set(); CurrentReceivingHandle = new EventWaitHandle(false, EventResetMode.ManualReset); + + await Task.Yield(); + while (true) + { + var message = ReceiveMessage(); + LastMessage = message; + _messageBuffer.Add(message); + CurrentReceivingHandle.Set(); + CurrentReceivingHandle = new EventWaitHandle(false, EventResetMode.ManualReset); + } + } + catch (Exception) + { + // ignored } } @@ -71,9 +82,9 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule _messageBuffer.Remove(message); } - public NetworkMessage? FirstByFilter(Predicate predicate) + public NetworkMessage FirstByFilter(Predicate predicate) { - return _messageBuffer.FirstOrDefault(m => predicate(m)); + return _messageBuffer.FirstOrDefault(m => predicate(m)) ?? NetworkMessage.Null; } private NetworkMessage GetNextMessage() @@ -83,19 +94,18 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule if (_serialization == null) throw new NullReferenceException("Serialization toolkit is null"); - - + var message = ReceiveMessage(); _messageBuffer.Add(message); _currentReceiving = Task.Run(GetNextMessage); - return message!; + return message; } private NetworkMessage ReceiveMessage() { - var len = BitConverter.ToUInt16([(byte)BaseStream!.ReadByte(), (byte)BaseStream.ReadByte()]); + var len = BitConverter.ToUInt16([(byte)BaseStream.ReadByte(), (byte)BaseStream.ReadByte()]); var localSpan = _buffer.Span.Slice(0, len); BaseStream.ReadExactly(localSpan); diff --git a/mROA/Implementation/RemoteObjectBase.cs b/mROA/Implementation/RemoteObjectBase.cs index 6a813f2..3794a0e 100644 --- a/mROA/Implementation/RemoteObjectBase.cs +++ b/mROA/Implementation/RemoteObjectBase.cs @@ -15,18 +15,27 @@ public abstract class RemoteObjectBase(int id, IRepresentationModule representat var request = new DefaultCallRequest { CommandId = methodId, ObjectId = id, Parameter = parameter, ParameterType = parameter?.GetType() }; await representationModule.PostCallMessageAsync(request.Id, MessageType.CallRequest, request); - + + var localTokenSource = new CancellationTokenSource(); + var successResponse = representationModule.GetMessageAsync>( - messageType: MessageType.FinishedCommandExecution, requestId: request.Id); + messageType: MessageType.FinishedCommandExecution, requestId: request.Id, token: localTokenSource.Token); var errorResponse = representationModule.GetMessageAsync( - messageType: MessageType.ExceptionCommandExecution, requestId: request.Id); + messageType: MessageType.ExceptionCommandExecution, requestId: request.Id, token: localTokenSource.Token); + Task.WaitAny(successResponse, errorResponse); + + if (successResponse.IsCompletedSuccessfully) + { + await localTokenSource.CancelAsync(); return successResponse.Result.Result!; - + } + + await localTokenSource.CancelAsync(); throw errorResponse.Result.GetException(); } diff --git a/mROA/Implementation/RepresentationModule.cs b/mROA/Implementation/RepresentationModule.cs index 4a119b9..038a9ac 100644 --- a/mROA/Implementation/RepresentationModule.cs +++ b/mROA/Implementation/RepresentationModule.cs @@ -22,47 +22,74 @@ public class RepresentationModule : IRepresentationModule public int Id => (_interaction ?? throw new NullReferenceException("Interaction is not initialized")).ConnectionId; - public async Task GetMessageAsync(Guid? requestId, MessageType? messageType) + public async Task GetMessageAsync(Guid? requestId, MessageType? messageType, + CancellationToken token = default) { if (_serialization == null) throw new NullReferenceException("Serialization toolkit is not initialized"); - - return _serialization.Deserialize(await GetRawMessage(requestId, messageType))!; + + return _serialization.Deserialize(await GetRawMessage(m => m.IsValidMessage(requestId, messageType), token))!; } public T GetMessage(Guid? requestId = null, MessageType? messageType = null) { - if (_serialization == null) - throw new NullReferenceException("Serialization toolkit is not initialized"); - - return _serialization.Deserialize(GetRawMessage(requestId, messageType).GetAwaiter().GetResult())!; + return GetMessage(m => m.IsValidMessage(requestId, messageType)); } - public async Task GetRawMessage(Guid? requestId = null, MessageType? messageType = null) + public T GetMessage(Predicate filter) { + if (_serialization == null) + throw new NullReferenceException("Serialization toolkit is not initialized"); + + return _serialization.Deserialize(GetRawMessage(filter).GetAwaiter().GetResult())!; + } + + public async Task GetRawMessage(Predicate filter, CancellationToken token = default) + { + await Task.Yield(); + 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.SchemaId == messageType)); - - if (fromBuffer == null) + + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Reading"); + + if (filter(_interaction.LastMessage)) { - while (true) - { - var message = await _interaction.GetNextMessageReceiving(); - if ((requestId is not null && message.Id != requestId) || - (messageType is not null && message.SchemaId != messageType)) continue; - - _interaction.HandleMessage(message); - return message.Data; - } + _interaction.HandleMessage(_interaction.LastMessage); + return _interaction.LastMessage.Data; } - _interaction.HandleMessage(fromBuffer); - return fromBuffer.Data; + var message = _interaction.FirstByFilter(filter); + + if (message != NetworkMessage.Null) + { + _interaction.HandleMessage(message); + return message.Data; + } + + + while (!token.IsCancellationRequested) + { + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Receiving message..."); + var handle = _interaction.CurrentReceivingHandle; + handle.WaitOne(); + + message = _interaction.LastMessage; + + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Message received {message.SchemaId} - {message.Id}"); + if (!filter(message)) continue; + + message = _interaction.LastMessage; + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Message received Successfully {message.SchemaId} - {message.Id}"); + _interaction.HandleMessage(message); + return message.Data; + } + + return []; } public async Task PostCallMessageAsync(Guid id, MessageType messageType, T payload) where T : notnull @@ -76,7 +103,7 @@ public class RepresentationModule : IRepresentationModule throw new NullReferenceException("Interaction toolkit is not initialized"); if (_serialization == null) throw new NullReferenceException("Serialization toolkit is not initialized"); - + await _interaction.PostMessage(new NetworkMessage { Id = id, SchemaId = messageType, Data = _serialization.Serialize(payload, payloadType) }); }