New distribution model base

Example.Backend/Program.cs

@@ -40,6 +40,7 @@ class Program
             return repo;
         }));
 
+        builder.Services.AddSingleton<IMessageDistributorFactory, ChannelDistributorFactory>();
         builder.Services.AddSingleton<IMethodRepository>(p =>
         {
             var methodRepo = new CollectableMethodRepository();

mROA/Abstract/IPrimaryMessageDistributior.cs

@@ -0,0 +1,15 @@
+using System.Threading.Tasks;
+using mROA.Implementation;
+
+namespace mROA.Abstract
+{
+    public interface IPrimaryMessageDistributior
+    {
+        Task Distribute(NetworkMessageHeader message);
+    }
+
+    public interface IMessageDistributorFactory
+    {
+        IPrimaryMessageDistributior Produce(int clientId);
+    }
+}

mROA/Abstract/IRequestExtractor.cs

@@ -1,9 +1,14 @@
+using System;
 using System.Threading.Tasks;
+using mROA.Implementation;
 
 namespace mROA.Abstract
 {
     public interface IRequestExtractor
     {
         Task StartExtraction();
+        void PushMessage(object parced, EMessageType originalType);
+        Predicate<NetworkMessageHeader> Rule { get; }
+        Func<NetworkMessageHeader,Type?>[] Converters { get; }
     }
 }

mROA/Implementation/Backend/BasicExecutionModule.cs

@@ -23,7 +23,6 @@ namespace mROA.Implementation.Backend
         {
             try
             {
-                ThrowIfNotInjected(instanceRepository);
                 if (command is CancelRequest)
                 {
                     return CancelExecution(command);
@@ -96,18 +95,6 @@ namespace mROA.Implementation.Backend
             return castedParams;
         }
 
-        private void ThrowIfNotInjected(IInstanceRepository instanceRepository)
-        {
-            if (_cancellationRepo is null)
-                throw new NullReferenceException("Method repository was not defined");
-
-            if (_methodRepo is null)
-                throw new NullReferenceException("Method repository was not defined");
-
-            if (instanceRepository is null)
-                throw new NullReferenceException("Context repository was not defined");
-        }
-
         private FinalCommandExecution CancelExecution(ICallRequest command)
         {
             var cts = _cancellationRepo.GetCancellation(command.Id);

mROA/Implementation/Backend/HubRequestExtractor.cs

@@ -37,7 +37,7 @@ namespace mROA.Implementation.Backend
 
         private IRequestExtractor CreateExtractor(IRepresentationModule interaction)
         {
-            var extractor = new RequestExtractor(_executeModule, _methodRepository, interaction, _serializationToolkit, interaction.Context);
+            var extractor = new RequestExtractor(_executeModule, interaction, interaction.Context);
             var context = interaction.Context;
             if (_contextRepository is IContextRepositoryHub contextHub)
                 context.RealRepository = contextHub.GetRepository(interaction.Id);

mROA/Implementation/Backend/NetworkGatewayModule.cs

@@ -11,21 +11,21 @@ namespace mROA.Implementation.Backend
 {
     public class NetworkGatewayModule : IGatewayModule
     {
-        private readonly IServiceProvider _serviceProvider;
         private readonly TcpListener _tcpListener;
         private readonly IConnectionHub _hub;
         private readonly IContextualSerializationToolKit _serialization;
         private readonly Dictionary<int, CancellationTokenSource> _extractorsCTS = new();
         private ICallIndexProvider _callIndexProvider;
         private readonly IIdentityGenerator _identityGenerator;
-        public NetworkGatewayModule(IOptions<GatewayOptions> options, IServiceProvider service, IIdentityGenerator identityGenerator, IContextualSerializationToolKit serialization, ICallIndexProvider callIndexProvider, IConnectionHub hub)
+        private readonly IMessageDistributorFactory _distributorFactory;
+        public NetworkGatewayModule(IOptions<GatewayOptions> options, IIdentityGenerator identityGenerator, IContextualSerializationToolKit serialization, ICallIndexProvider callIndexProvider, IConnectionHub hub, IMessageDistributorFactory distributorFactory)
         {
             _tcpListener = new(options.Value.Endpoint);
-            _serviceProvider = service;
             _identityGenerator = identityGenerator;
             _serialization = serialization;
             _callIndexProvider = callIndexProvider;
             _hub = hub;
+            _distributorFactory = distributorFactory;
         }
 
         public void Run()
@@ -74,7 +74,7 @@ namespace mROA.Implementation.Backend
                         interaction.PostMessageAsync(new NetworkMessageHeader(_serialization!,
                             new IdAssignment { Id = interaction.ConnectionId }, null));
                         _extractorsCTS[interaction.ConnectionId] = cts;
-                        _hub!.RegisterInteraction(interaction);
+                        _hub.RegisterInteraction(interaction);
                         Console.WriteLine("Client registered");
                         break;
                     case EMessageType.ClientRecovery:

mROA/Implementation/ChannelDistributorFactory.cs

@@ -0,0 +1,35 @@
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using mROA.Abstract;
+
+namespace mROA.Implementation
+{
+    public class ChannelDistributorFactory : IMessageDistributorFactory
+    {
+        private readonly IConnectionHub _hub;
+
+        public ChannelDistributorFactory(IConnectionHub hub)
+        {
+            _hub = hub;
+        }
+        public IPrimaryMessageDistributior Produce(int clientId)
+        {
+            return new ChannelMessageDistributor(_hub.GetInteraction(clientId).ReceiveChanel.Writer);
+        }
+    }
+
+    public class ChannelMessageDistributor : IPrimaryMessageDistributior
+    {
+        private readonly ChannelWriter<NetworkMessageHeader> _writer;
+
+        public ChannelMessageDistributor(ChannelWriter<NetworkMessageHeader> writer)
+        {
+            _writer = writer;
+        }
+
+        public async Task Distribute(NetworkMessageHeader message)
+        {
+            await _writer.WriteAsync(message);
+        }
+    }
+}

mROA/Implementation/ChannelInteractionModule.cs

@@ -82,9 +82,6 @@ namespace mROA.Implementation
 
         public async Task PostMessageAsync(NetworkMessageHeader messageHeader)
         {
-            if (_serialization == null)
-                throw new NullReferenceException("Serialization toolkit is not initialized");
-
             while (true)
             {
                 if (await PostMessageInternal(messageHeader))

mROA/Implementation/CreativeRepresentationModuleProducer.cs

@@ -18,9 +18,6 @@ namespace mROA.Implementation
 
         public IRepresentationModule Produce(int id)
         {
-            if (_hub == null)
-                throw new NullReferenceException("Interaction module is null");
-
             var interaction = _hub.GetInteraction(id);
             
             var produced = new RepresentationModule(interaction, _serialization);

mROA/Implementation/Frontend/NetworkFrontendBridge.cs

@@ -31,11 +31,6 @@ namespace mROA.Implementation.Frontend
 
         public async Task Connect()
         {
-            if (_interactionModule is null)
-                throw new Exception("Interaction module was not injected");
-            if (_serialization == null)
-                throw new NullReferenceException("Serialization toolkit is not initialized");
-
             _tcpClient.Connect(_serverEndPoint);
             _tcpClient.NoDelay = true;
             PrepareExtractor();

mROA/Implementation/Frontend/RequestExtractor.cs

@@ -9,71 +9,64 @@ namespace mROA.Implementation.Frontend
 {
     public class RequestExtractor : IRequestExtractor
     {
+        
         private IExecuteModule _executeModule;
 
-        private IMethodRepository _methodRepository;
-
         private IRepresentationModule _representationModule;
-        private IContextualSerializationToolKit _serializationToolkit;
         private IEndPointContext _context;
 
-        public RequestExtractor(IExecuteModule executeModule, IMethodRepository methodRepository, IRepresentationModule representationModule, IContextualSerializationToolKit serializationToolkit, IEndPointContext context)
+        public RequestExtractor(IExecuteModule executeModule, IRepresentationModule representationModule, IEndPointContext context)
         {
             _executeModule = executeModule;
-            _methodRepository = methodRepository;
             _representationModule = representationModule;
-            _serializationToolkit = serializationToolkit;
             _context = context;
         }
 
         public async Task StartExtraction()
         {
-            ThrowIfNotInjected();
-
             var streamTokenSource = new CancellationTokenSource();
 
-            var query = _representationModule.GetStream(m =>
+            var query = _representationModule.GetStream(Rule, _context,
-                    m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest
-                        or EMessageType.EventRequest or EMessageType.ClientDisconnect, _context,
                 streamTokenSource.Token,
-                m => m.MessageType == EMessageType.CallRequest ? typeof(DefaultCallRequest) : null,
+                Converters);
-                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)
             {
-                switch (command.originalType)
+                PushMessage(command.parced, command.originalType);
+            }
+        }
+
+        public void PushMessage(object parced, EMessageType originalType)
+        {
+            switch (originalType)
             {
                 case EMessageType.CallRequest:
-                        HandleCallRequest((DefaultCallRequest)command.parced);
+                    HandleCallRequest((DefaultCallRequest)parced);
                     break;
                 case EMessageType.ClientDisconnect:
                     return;
                 case EMessageType.EventRequest:
-                        HandleEventRequest((DefaultCallRequest)command.parced);
+                    HandleEventRequest((DefaultCallRequest)parced);
                     break;
                 case EMessageType.CancelRequest:
-                        HandleCancelRequest((command.parced as CancelRequest)!);
+                    HandleCancelRequest((parced as CancelRequest)!);
                     break;
                 default:
-                        continue;
+                    return;
-                }
             }
         }
 
-        private void ThrowIfNotInjected()
+        public Predicate<NetworkMessageHeader> Rule { get; } = m =>
-        {
+            m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest
-            if (_serializationToolkit == null)
+                or EMessageType.EventRequest or EMessageType.ClientDisconnect;
-                throw new NullReferenceException("Serializing toolkit is null.");
+
-            if (_executeModule == null)
+        public Func<NetworkMessageHeader, Type?>[] Converters { get; } = {
-                throw new NullReferenceException("Execute module is null.");
+            m => m.MessageType == EMessageType.CallRequest ? typeof(DefaultCallRequest) : null,
-            if (_representationModule == null)
+            m => m.MessageType == EMessageType.CancelRequest ? typeof(CancelRequest) : null,
-                throw new NullReferenceException("Representation module is null.");
+            m => m.MessageType == EMessageType.EventRequest ? typeof(DefaultCallRequest) : null,
-            if (_methodRepository == null)
+            m => m.MessageType == EMessageType.ClientDisconnect ? typeof(ClientDisconnect) : null
-                throw new NullReferenceException("Method repository is null.");
+        };
-        }
 
         private void HandleCancelRequest(CancelRequest req)
         {

mROA/Implementation/RemoteInstanceRepository.cs

@@ -34,8 +34,6 @@ namespace mROA.Implementation
             var index = _producedProxys.Find(i => i.Identifier.Equals(id));
             if (index is not null)
                 return (T)(index as object);
-            if (_representationProducer == null)
-                throw new NullReferenceException("representation producer is not initialized");
 
             if (!_callIndexProvider.Activators.TryGetValue(typeof(T), out var remoteType))
                 throw new NotSupportedException();
@@ -56,9 +54,6 @@ namespace mROA.Implementation
 
         public object GetSingletonObject(Type type, IEndPointContext context)
         {
-            if (_representationProducer == null)
-                throw new NullReferenceException("representation producer is not initialized");
-
             var representationModule =
                 _representationProducer.Produce(context.OwnerId);
 

mROA/Implementation/RepresentationModule.cs

@@ -82,11 +82,6 @@ namespace mROA.Implementation
         public async Task PostCallMessageAsync<T>(Guid id, EMessageType eMessageType, T payload,
             IEndPointContext? context) where T : notnull
         {
-            if (_interaction == null)
-                throw new NullReferenceException("Interaction toolkit is not initialized");
-            if (_serialization == null)
-                throw new NullReferenceException("Serialization toolkit is not initialized");
-
             var serialized = _serialization.Serialize(payload, context);
             await _interaction.PostMessageAsync(new NetworkMessageHeader
                 { Id = id, MessageType = eMessageType, Data = serialized });