diff --git a/mROA/Abstract/ISerialisationModule.cs b/mROA/Abstract/IRepresentationModule.cs similarity index 52% rename from mROA/Abstract/ISerialisationModule.cs rename to mROA/Abstract/IRepresentationModule.cs index 4cea03f..2d9fc06 100644 --- a/mROA/Abstract/ISerialisationModule.cs +++ b/mROA/Abstract/IRepresentationModule.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using mROA.Implementation; @@ -9,16 +10,13 @@ namespace mROA.Abstract { int Id { get; } - Task GetMessageAsync(Guid? requestId = null, EMessageType? messageType = null, - CancellationToken token = default); - - T GetMessage(Guid? requestId = null, EMessageType? messageType = null); - - Task GetRawMessage(Guid? requestId = null, EMessageType? messageType = null, - CancellationToken token = default); + Task<(object parced, EMessageType originalType)> GetSingle(Predicate rule, CancellationToken token, + params Func[] converter); + + IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate rule, CancellationToken token, + params Func[] converter); Task PostCallMessageAsync(Guid id, EMessageType eMessageType, T payload) where T : notnull; - Task PostCallMessageAsync(Guid id, EMessageType eMessageType, object payload, Type payloadType); void PostCallMessage(Guid id, EMessageType eMessageType, T payload) where T : notnull; void PostCallMessage(Guid id, EMessageType eMessageType, object payload, Type payloadType); } diff --git a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs index 5c48b40..4b98fab 100644 --- a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs +++ b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs @@ -1,4 +1,6 @@ using System; +using System.Collections.Generic; +using System.Linq; using System.Net; using System.Net.Sockets; using System.Threading; diff --git a/mROA/Implementation/Frontend/RequestExtractor.cs b/mROA/Implementation/Frontend/RequestExtractor.cs index fbe9539..5b348d9 100644 --- a/mROA/Implementation/Frontend/RequestExtractor.cs +++ b/mROA/Implementation/Frontend/RequestExtractor.cs @@ -4,7 +4,6 @@ using System.Threading; using System.Threading.Tasks; using mROA.Abstract; using mROA.Implementation.Backend; -using mROA.Implementation.CommandExecution; // ReSharper disable MethodHasAsyncOverload @@ -45,22 +44,32 @@ namespace mROA.Implementation.Frontend } } - public Task StartExtraction() + public async Task StartExtraction() { - return Task.Run(() => - { - ThrowIfNotInjected(); - var multiClientOwnershipRepository = - TransmissionConfig.OwnershipRepository as MultiClientOwnershipRepository; - multiClientOwnershipRepository?.RegisterOwnership(_representationModule.Id); + ThrowIfNotInjected(); + var multiClientOwnershipRepository = + TransmissionConfig.OwnershipRepository as MultiClientOwnershipRepository; + multiClientOwnershipRepository?.RegisterOwnership(_representationModule!.Id); - try - { + try + { #if TRACE var sw = new Stopwatch(); #endif - while (true) - { + + var streamTokenSource = new CancellationTokenSource(); + + var query = _representationModule!.GetStream(m => + m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest + or EMessageType.EventRequest or EMessageType.ClientDisconnect, streamTokenSource.Token, + m => m.MessageType == EMessageType.CallRequest ? typeof(DefaultCallRequest) : null, + m => m.MessageType == EMessageType.CancelRequest ? typeof(CancelRequest) : null, + m => m.MessageType == EMessageType.EventRequest ? typeof(DefaultCallRequest) : null, + m => m.MessageType == EMessageType.ClientDisconnect ? typeof(ClientDisconnect) : null); + + + await foreach (var command in query) + { #if TRACE Console.WriteLine("Waiting for request..."); if (sw.IsRunning) @@ -69,50 +78,34 @@ namespace mROA.Implementation.Frontend Console.WriteLine($"Request handling took {Math.Round(sw.Elapsed.TotalMilliseconds * 1000.0)} microseconds."); } #endif - var tokenSource = new CancellationTokenSource(); - var token = tokenSource.Token; - var defaultRequest = - _representationModule!.GetMessageAsync( - messageType: EMessageType.CallRequest, token: token); - var cancelRequest = - _representationModule!.GetMessageAsync( - messageType: EMessageType.CancelRequest, token: token); - var eventRequest = - _representationModule!.GetMessageAsync( - messageType: EMessageType.EventRequest, token: token); - var disconnectRequest = - _representationModule!.GetMessageAsync( - messageType: EMessageType.ClientDisconnect, token:token); - Task.WaitAny(defaultRequest, cancelRequest, eventRequest, disconnectRequest); + #if TRACE Console.WriteLine("Request received"); sw.Restart(); #endif - if (cancelRequest.IsCompleted) - { -#if TRACE - Console.WriteLine("Cancelling request"); -#endif - HandleCancelRequest(tokenSource, cancelRequest.Result); - } - else if (defaultRequest.IsCompleted) - { - HandleCallRequest(tokenSource, defaultRequest.Result); - } - else if(eventRequest.IsCompleted) - { - HandleEventRequest(tokenSource, eventRequest.Result); - }else if (disconnectRequest.IsCompleted) - { + switch (command.originalType) + { + case EMessageType.CallRequest: + HandleCallRequest((command.parced as DefaultCallRequest)!); break; - } + case EMessageType.ClientDisconnect: + return; + case EMessageType.EventRequest: + HandleEventRequest((command.parced as DefaultCallRequest)!); + break; + case EMessageType.CancelRequest: + HandleCancelRequest((command.parced as CancelRequest)!); + + break; + default: + continue; } } - catch - { - multiClientOwnershipRepository?.FreeOwnership(); - } - }); + } + catch + { + multiClientOwnershipRepository?.FreeOwnership(); + } } private void ThrowIfNotInjected() @@ -129,20 +122,17 @@ namespace mROA.Implementation.Frontend throw new NullReferenceException("Method repository is null."); } - private void HandleCancelRequest(CancellationTokenSource tokenSource, CancelRequest req) + private void HandleCancelRequest(CancelRequest req) { - tokenSource.Cancel(); _executeModule!.Execute(req, _realContextRepository!, _representationModule!); } - private void HandleCallRequest(CancellationTokenSource tokenSource, DefaultCallRequest request) + private void HandleCallRequest(DefaultCallRequest request) { - tokenSource.Cancel(); - var result = _executeModule!.Execute(request, _realContextRepository!, _representationModule!); var resultType = result.MessageType; - + if (resultType == EMessageType.Unknown) { return; @@ -151,9 +141,8 @@ namespace mROA.Implementation.Frontend _representationModule!.PostCallMessage(request.Id, resultType, result, result.GetType()); } - private void HandleEventRequest(CancellationTokenSource tokenSource, DefaultCallRequest request) + private void HandleEventRequest(DefaultCallRequest request) { - tokenSource.Cancel(); _executeModule!.Execute(request, _remoteContextRepository!, _representationModule!); } } diff --git a/mROA/Implementation/NetworkMessageHeader.cs b/mROA/Implementation/NetworkMessageHeader.cs index 8e6c643..61ecea2 100644 --- a/mROA/Implementation/NetworkMessageHeader.cs +++ b/mROA/Implementation/NetworkMessageHeader.cs @@ -1,6 +1,7 @@ using System; using System.Text.Json.Serialization; using mROA.Abstract; +using mROA.Implementation.Attributes; // ReSharper disable UnusedMember.Global @@ -45,5 +46,9 @@ namespace mROA.Implementation public EMessageType MessageType { get; set; } public byte[] Data { get; set; } + + [JsonIgnore] + [SerializationIgnore] + public object Parced { get; set; } } } \ No newline at end of file diff --git a/mROA/Implementation/RemoteObjectBase.cs b/mROA/Implementation/RemoteObjectBase.cs index 7f2be6f..941cb52 100644 --- a/mROA/Implementation/RemoteObjectBase.cs +++ b/mROA/Implementation/RemoteObjectBase.cs @@ -10,7 +10,7 @@ namespace mROA.Implementation { public abstract class RemoteObjectBase : IDisposable { - protected bool Equals(RemoteObjectBase other) + public bool Equals(RemoteObjectBase other) { return _identifier.Equals(other._identifier); } @@ -60,40 +60,36 @@ namespace mROA.Implementation var localTokenSource = new CancellationTokenSource(); - var successResponse = - _representationModule.GetMessageAsync>(request.Id, - EMessageType.FinishedCommandExecution, - localTokenSource.Token); - var errorResponse = - _representationModule.GetMessageAsync(requestId: request.Id, - EMessageType.ExceptionCommandExecution, localTokenSource.Token); + var responseRequestTask = _representationModule.GetSingle( + m => m.MessageType is EMessageType.FinishedCommandExecution or EMessageType.ExceptionCommandExecution, + localTokenSource.Token, + m => m.MessageType is EMessageType.FinishedCommandExecution ? typeof(FinalCommandExecution) : null, + m => m.MessageType is EMessageType.ExceptionCommandExecution + ? typeof(ExceptionCommandExecution) + : null); - cancellationToken.Register(async () => + cancellationToken.Register(() => { #if TRACE Console.WriteLine("Cancelling task"); #endif - await _representationModule.PostCallMessageAsync(request.Id, EMessageType.CancelRequest, + _representationModule.PostCallMessageAsync(request.Id, EMessageType.CancelRequest, new CancelRequest { Id = request.Id - }); - localTokenSource.Cancel(); + }).ContinueWith(_ => localTokenSource.Cancel()); }); - Task.WaitAny(new Task[] - { - successResponse, errorResponse - }, cancellationToken); + var response = await responseRequestTask; - if (successResponse.IsCompletedSuccessfully) + if (response.parced is FinalCommandExecution successResponse) { localTokenSource.Cancel(); - return successResponse.Result.Result!; + return successResponse.Result!; } localTokenSource.Cancel(); - throw errorResponse.Result.GetException(); + throw (response.parced as ExceptionCommandExecution)!.GetException(); } protected async Task CallAsync(int methodId, object?[]? parameters = null, @@ -107,40 +103,38 @@ namespace mROA.Implementation var localTokenSource = new CancellationTokenSource(); - var successResponse = - _representationModule.GetMessageAsync(request.Id, - EMessageType.FinishedCommandExecution, - localTokenSource.Token); - var errorResponse = - _representationModule.GetMessageAsync(requestId: request.Id, - EMessageType.ExceptionCommandExecution, localTokenSource.Token); + var responceRequestTask = _representationModule.GetSingle( + m => m.MessageType is EMessageType.FinishedCommandExecution or EMessageType.ExceptionCommandExecution, + localTokenSource.Token, + m => m.MessageType is EMessageType.FinishedCommandExecution ? typeof(FinalCommandExecution) : null, + m => m.MessageType is EMessageType.ExceptionCommandExecution + ? typeof(ExceptionCommandExecution) + : null); - cancellationToken.Register(async () => + + cancellationToken.Register(() => { #if TRACE Console.WriteLine("Cancelling task"); #endif - await _representationModule.PostCallMessageAsync(request.Id, EMessageType.CancelRequest, + _representationModule.PostCallMessageAsync(request.Id, EMessageType.CancelRequest, new CancelRequest { Id = request.Id - }); - localTokenSource.Cancel(); + }).ContinueWith(_ => localTokenSource.Cancel()); }); - Task.WaitAny(new Task[] - { - errorResponse, successResponse - }, cancellationToken); - + var responseRequest = await responceRequestTask; #if TRACE Console.WriteLine($"Handling message"); #endif - if (successResponse.IsCompletedSuccessfully) - return; - - if (errorResponse.IsCompletedSuccessfully) - throw errorResponse.Result.GetException(); + switch (responseRequest.originalType) + { + case EMessageType.FinishedCommandExecution: + return; + case EMessageType.ExceptionCommandExecution: + throw (responseRequest.parced as ExceptionCommandExecution)!.GetException(); + } } public override string ToString() diff --git a/mROA/Implementation/RepresentationModule.cs b/mROA/Implementation/RepresentationModule.cs index d758c6f..2e80308 100644 --- a/mROA/Implementation/RepresentationModule.cs +++ b/mROA/Implementation/RepresentationModule.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using mROA.Abstract; @@ -28,6 +29,16 @@ namespace mROA.Implementation public int Id => (_interaction ?? throw new NullReferenceException("Interaction is not initialized")) .ConnectionId; + public Task<(object parced, EMessageType originalType)> GetSingle(Predicate rule, CancellationToken token, params Func[] converter) + { + throw new NotImplementedException(); + } + + public IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate rule, CancellationToken token, params Func[] converter) + { + throw new NotImplementedException(); + } + public async Task GetMessageAsync(Guid? requestId, EMessageType? messageType, CancellationToken token = default) {