Abstraction of representation module rewrite

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<T> GetMessageAsync<T>(Guid? requestId = null, EMessageType? messageType = null,
+        Task<(object parced, EMessageType originalType)> GetSingle(Predicate<NetworkMessageHeader> rule, CancellationToken token,
-            CancellationToken token = default);
+            params Func<NetworkMessageHeader, Type?>[] converter);
         
-        T GetMessage<T>(Guid? requestId = null, EMessageType? messageType = null);
+        IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate<NetworkMessageHeader> rule, CancellationToken token,
-
+            params Func<NetworkMessageHeader, Type?>[] converter);
-        Task<byte[]> GetRawMessage(Guid? requestId = null, EMessageType? messageType = null,
-            CancellationToken token = default);
 
         Task PostCallMessageAsync<T>(Guid id, EMessageType eMessageType, T payload) where T : notnull;
-        Task PostCallMessageAsync(Guid id, EMessageType eMessageType, object payload, Type payloadType);
         void PostCallMessage<T>(Guid id, EMessageType eMessageType, T payload) where T : notnull;
         void PostCallMessage(Guid id, EMessageType eMessageType, object payload, Type payloadType);
     }

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;

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,21 +44,31 @@ 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);
+            multiClientOwnershipRepository?.RegisterOwnership(_representationModule!.Id);
 
             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...");
@@ -69,42 +78,27 @@ 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<DefaultCallRequest>(
-                                messageType: EMessageType.CallRequest, token: token);
-                        var cancelRequest =
-                            _representationModule!.GetMessageAsync<CancelRequest>(
-                                messageType: EMessageType.CancelRequest, token: token);
-                        var eventRequest =
-                            _representationModule!.GetMessageAsync<DefaultCallRequest>(
-                                messageType: EMessageType.EventRequest, token: token);
-                        var disconnectRequest =
-                            _representationModule!.GetMessageAsync<ClientDisconnect>(
-                                messageType: EMessageType.ClientDisconnect, token:token);
-                        Task.WaitAny(defaultRequest, cancelRequest, eventRequest, disconnectRequest);
 #if TRACE
                         Console.WriteLine("Request received");
                         sw.Restart();
 #endif
-                        if (cancelRequest.IsCompleted)
+                    switch (command.originalType)
-                        {
-#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)
                     {
+                        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;
                     }
                 }
             }
@@ -112,7 +106,6 @@ namespace mROA.Implementation.Frontend
             {
                 multiClientOwnershipRepository?.FreeOwnership();
             }
-            });
         }
 
         private void ThrowIfNotInjected()
@@ -129,16 +122,13 @@ 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;
@@ -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!);
         }
     }

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; }
     }
 }

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 =
+            var responseRequestTask = _representationModule.GetSingle(
-                _representationModule.GetMessageAsync<FinalCommandExecution<T>>(request.Id,
+                m => m.MessageType is EMessageType.FinishedCommandExecution or EMessageType.ExceptionCommandExecution,
-                    EMessageType.FinishedCommandExecution,
+                localTokenSource.Token,
-                    localTokenSource.Token);
+                m => m.MessageType is EMessageType.FinishedCommandExecution ? typeof(FinalCommandExecution<T>) : null,
-            var errorResponse =
+                m => m.MessageType is EMessageType.ExceptionCommandExecution
-                _representationModule.GetMessageAsync<ExceptionCommandExecution>(requestId: request.Id,
+                    ? typeof(ExceptionCommandExecution)
-                    EMessageType.ExceptionCommandExecution, localTokenSource.Token);
+                    : 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
-                    });
+                    }).ContinueWith(_ => localTokenSource.Cancel());
-                localTokenSource.Cancel();
             });
 
-            Task.WaitAny(new Task[]
+            var response = await responseRequestTask;
-            {
-                successResponse, errorResponse
-            }, cancellationToken);
 
-            if (successResponse.IsCompletedSuccessfully)
+            if (response.parced is FinalCommandExecution<T> 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 =
+            var responceRequestTask = _representationModule.GetSingle(
-                _representationModule.GetMessageAsync<FinalCommandExecution>(request.Id,
+                m => m.MessageType is EMessageType.FinishedCommandExecution or EMessageType.ExceptionCommandExecution,
-                    EMessageType.FinishedCommandExecution,
+                localTokenSource.Token,
-                    localTokenSource.Token);
+                m => m.MessageType is EMessageType.FinishedCommandExecution ? typeof(FinalCommandExecution) : null,
-            var errorResponse =
+                m => m.MessageType is EMessageType.ExceptionCommandExecution
-                _representationModule.GetMessageAsync<ExceptionCommandExecution>(requestId: request.Id,
+                    ? typeof(ExceptionCommandExecution)
-                    EMessageType.ExceptionCommandExecution, localTokenSource.Token);
+                    : 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
-                    });
+                    }).ContinueWith(_ => localTokenSource.Cancel());
-                localTokenSource.Cancel();
             });
 
-            Task.WaitAny(new Task[]
+            var responseRequest = await responceRequestTask;
-            {
-                errorResponse, successResponse
-            }, cancellationToken);
-
 #if TRACE
             Console.WriteLine($"Handling message");
 #endif
-            if (successResponse.IsCompletedSuccessfully)
+            switch (responseRequest.originalType)
+            {
+                case EMessageType.FinishedCommandExecution:
                     return;
-
+                case EMessageType.ExceptionCommandExecution:
-            if (errorResponse.IsCompletedSuccessfully)
+                    throw (responseRequest.parced as ExceptionCommandExecution)!.GetException();
-                throw errorResponse.Result.GetException();
+            }
         }
 
         public override string ToString()

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<NetworkMessageHeader> rule, CancellationToken token, params Func<NetworkMessageHeader, Type?>[] converter)
+        {
+            throw new NotImplementedException();
+        }
+
+        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)
         {