Channel interaction module refresh base

Example.Backend/Program.cs

@@ -19,7 +19,7 @@ class Program
         builder.Modules.Add(new BackendIdentityGenerator());
         // builder.UseNetworkGateway(new IPEndPoint(IPAddress.Loopback, 4567), typeof(NextGenerationInteractionModule),
         //     builder.GetModule<IIdentityGenerator>()!);
-        builder.UseNetworkGateway(new IPEndPoint(IPAddress.Loopback, 4567), typeof(NextGenerationInteractionModule),
+        builder.UseNetworkGateway(new IPEndPoint(IPAddress.Loopback, 4567), typeof(ChannelInteractionModule),
             builder.GetModule<IIdentityGenerator>()!);
 
         builder.Modules.Add(new ConnectionHub());

Example.Frontend/Program.cs

@@ -23,7 +23,7 @@ class Program
         builder.Modules.Add(new CborSerializationToolkit());
 
         builder.Modules.Add(new RemoteContextRepository());
-        builder.Modules.Add(new NextGenerationInteractionModule());
+        builder.Modules.Add(new ChannelInteractionModule());
         builder.Modules.Add(new RepresentationModule());
         builder.Modules.Add(new NetworkFrontendBridge(new IPEndPoint(IPAddress.Loopback, 4567)));
         builder.Modules.Add(new StaticRepresentationModuleProducer());
@@ -58,7 +58,7 @@ class Program
             Console.WriteLine("Printer created");
             Thread.Sleep(100);
 
-            frontendBridge.Obstacle();
+            // frontendBridge.Obstacle();
             var name = disposingPrinter.GetName();
             DemoCheck.BasicNonParamsCall = true;
             Console.WriteLine("Printer name : {0}", name);

mROA.Test/NextGenTest.cs

@@ -11,17 +11,17 @@ namespace mROA.Test
     public class NextGenTest
     {
         private TcpListener _listener;
-        private NextGenerationInteractionModule _interactionModuleA;
+        private ChannelInteractionModule _interactionModuleA;
-        private NextGenerationInteractionModule _interactionModuleB;
+        private ChannelInteractionModule _interactionModuleB;
         private Guid[] guids = [Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid()];
 
         [SetUp]
         public void Setup()
         {
             _listener = new TcpListener(IPAddress.Loopback, 4567);
-            _interactionModuleA = new NextGenerationInteractionModule();
+            _interactionModuleA = new ChannelInteractionModule();
             _interactionModuleA.Inject(new JsonSerializationToolkit());
-            _interactionModuleB = new NextGenerationInteractionModule();
+            _interactionModuleB = new ChannelInteractionModule();
             _interactionModuleB.Inject(new JsonSerializationToolkit());
 
         }

mROA/Abstract/IConnectionHub.cs

@@ -6,8 +6,8 @@
 
     public interface IConnectionHub : IInjectableModule
     {
-        void RegisterInteraction(INextGenerationInteractionModule interaction);
+        void RegisterInteraction(IChannelInteractionModule interaction);
-        INextGenerationInteractionModule GetInteraction(int id);
+        IChannelInteractionModule GetInteraction(int id);
         event ConnectionHandler? OnConnected;
         event DisconnectionHandler? OnDisconnected;
     }

mROA/Abstract/IInteractionModule.cs

@@ -6,18 +6,17 @@ using mROA.Implementation;
 
 namespace mROA.Abstract
 {
-    public interface INextGenerationInteractionModule : IInjectableModule, IDisposable
+    public interface IChannelInteractionModule : IInjectableModule, IDisposable
     {
         int ConnectionId { get; set; }
-        Stream? BaseStream { get; set; }
+        ChannelWriter<NetworkMessageHeader> ReceiveChanel { get; }
-        ChannelReader<NetworkMessageHeader> UntrustedReceiveChanel { get; set; }
+        ChannelReader<NetworkMessageHeader> TrustedPostChanel { get; }
-        ChannelWriter<(int clientId, NetworkMessageHeader messageHeader)> UntrustedPostChanel { get; set; }
+        ChannelReader<NetworkMessageHeader> UntrustedPostChanel { get; }
-        
+        Action<bool> IsConnected { get; set; }
         Task<NetworkMessageHeader> GetNextMessageReceiving(bool infinite = true);
         Task PostMessageAsync(NetworkMessageHeader messageHeader);
         Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader);
         void HandleMessage(NetworkMessageHeader messageHeader);
-        // NetworkMessageHeader[] UnhandledMessages { get; }
         NetworkMessageHeader? FirstByFilter(Predicate<NetworkMessageHeader> predicate);
         event Action<int> OnDisconnected;
         Task Restart(bool sendRecovery);

mROA/Implementation/Backend/ConnectionHub.cs

@@ -6,10 +6,10 @@ namespace mROA.Implementation.Backend
 {
     public class ConnectionHub : IConnectionHub
     {
-        private readonly Dictionary<int, INextGenerationInteractionModule> _connections = new();
+        private readonly Dictionary<int, IChannelInteractionModule> _connections = new();
         private ISerializationToolkit? _serializationToolkit;
 
-        public void RegisterInteraction(INextGenerationInteractionModule interaction)
+        public void RegisterInteraction(IChannelInteractionModule interaction)
         {
             if (_serializationToolkit is null)
                 throw new NullReferenceException("Serialization toolkit is null");
@@ -21,7 +21,7 @@ namespace mROA.Implementation.Backend
             OnConnected?.Invoke(module);
         }
 
-        public INextGenerationInteractionModule GetInteraction(int id)
+        public IChannelInteractionModule GetInteraction(int id)
         {
             return _connections!.GetValueOrDefault(id, null) ?? throw new Exception("No connection found");
         }

mROA/Implementation/Backend/NetworkGatewayModule.cs

@@ -67,19 +67,21 @@ namespace mROA.Implementation.Backend
             {
                 var client = _tcpListener.AcceptTcpClient();
                 Console.WriteLine($"Client connected from {client.Client.RemoteEndPoint}");
-                var interaction = Activator.CreateInstance(_interactionModuleType!) as INextGenerationInteractionModule;
+                var interaction = Activator.CreateInstance(_interactionModuleType!) as IChannelInteractionModule;
 
                 foreach (var injectableModule in _injectableModules!)
                     interaction!.Inject(injectableModule);
 
                 interaction!.Inject(_serialization);
                 interaction.BaseStream = client.GetStream();
-                interaction.UntrustedReceiveChanel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
+                var channel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
                 {
                     SingleWriter = false,
                     SingleReader = false,
                     AllowSynchronousContinuations = true
-                }).Reader;
+                }); 
+                interaction.UntrustedReceiveChanel = channel.Reader;
+                interaction.UntrustedReceiveChanelWriter = channel.Writer;
                 var connectionRequest = interaction.GetNextMessageReceiving(false)
                     .GetAwaiter().GetResult()!;
 

mROA/Implementation/ChannelInteractionModule.cs

@@ -2,43 +2,59 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using mROA.Abstract;
 
 namespace mROA.Implementation
 {
-    public class NextGenerationInteractionModule : INextGenerationInteractionModule
+    public class ChannelInteractionModule : IChannelInteractionModule
     {
-        private int DebugId = new Random().Next();
+        private readonly ChannelReader<NetworkMessageHeader> _receiveReader;
+        private readonly Channel<NetworkMessageHeader> _inputChannel;
+        private readonly Channel<NetworkMessageHeader> _outputTrustedChannel;
+        private readonly Channel<NetworkMessageHeader> _outputUntrustedChannel;
         private const int BufferSize = ushort.MaxValue;
         private readonly Memory<byte> _buffer = new byte[BufferSize];
         private readonly List<NetworkMessageHeader> _messageBuffer = new(128);
         private Task<NetworkMessageHeader>? _currentReceiving;
         private ISerializationToolkit? _serialization;
-        private Stream? _baseStream;
         private bool _isConnected = true;
         private bool _isInReconnectionState;
         private bool _isActive = true;
         private TaskCompletionSource<Stream> _reconnection;
-        private ValueTask<NetworkMessageHeader>? _trustedReceive;
-        private TaskCompletionSource<NetworkMessageHeader> _untrustedReceive;
 
-        public NextGenerationInteractionModule()
+        public ChannelInteractionModule()
         {
             _reconnection = new TaskCompletionSource<Stream>();
+            _inputChannel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
+            {
+                SingleReader = false,
+                SingleWriter = false,
+                AllowSynchronousContinuations = true
+            });
+            _receiveReader = _inputChannel.Reader;
+            _outputTrustedChannel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
+            {
+                SingleReader = true,
+                SingleWriter = true,
+                AllowSynchronousContinuations = true
+            });
+            _outputUntrustedChannel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
+            {
+                SingleReader = true,
+                SingleWriter = true,
+                AllowSynchronousContinuations = true
+            });
         }
 
         public int ConnectionId { get; set; }
 
-        public Stream? BaseStream
+        public ChannelWriter<NetworkMessageHeader> ReceiveChanel => _inputChannel.Writer;
-        {
+        public ChannelReader<NetworkMessageHeader> TrustedPostChanel => _outputTrustedChannel.Reader;
-            get => _baseStream;
+        public ChannelReader<NetworkMessageHeader> UntrustedPostChanel => _outputUntrustedChannel.Reader;
-            set => _baseStream = value;
+        public Action<bool> IsConnected { get; set; }
-        }
-
-        public ChannelReader<NetworkMessageHeader> UntrustedReceiveChanel { get; set; }
-        public ChannelWriter<(int clientId, NetworkMessageHeader messageHeader)> UntrustedPostChanel { get; set; }
 
 
         public void Inject<T>(T dependency)
@@ -152,46 +168,10 @@ namespace mROA.Implementation
 
                 try
                 {
-                    NetworkMessageHeader message;
+                    var message = await _receiveReader.ReadAsync();
-
-                    var wasNull = _untrustedReceive is null;
-
-                    _trustedReceive ??= Receive();
-                    _untrustedReceive = new TaskCompletionSource<NetworkMessageHeader>();
-
-
-                    if (!wasNull)
-                    {
-                        if (_trustedReceive.Value.IsCompleted)
-                        {
-                            _trustedReceive = Receive();
-                        }
-
-                        if (_untrustedReceive.Task.IsCompleted)
-                        {
-                            _untrustedReceive = new TaskCompletionSource<NetworkMessageHeader>();
-                            _ = UntrustedReceiveChanel.ReadAsync().AsTask()
-                                .ContinueWith(task => _untrustedReceive.SetResult(task.Result));
-                        }
-                    }
-                    else
-                    {
-                        _ = UntrustedReceiveChanel.ReadAsync().AsTask()
-                            .ContinueWith(task => _untrustedReceive.SetResult(task.Result));
-                    }
-
-
-                    await Task.WhenAny(_trustedReceive.Value.AsTask() , _untrustedReceive.Task);
-
-                    message = _trustedReceive.Value.IsCompleted
-                        ? _trustedReceive.Value.Result
-                        : _untrustedReceive.Task.Result;
-
-                    _currentReceiving = Task.Run(async () => await GetNextMessage());
-
                     return message;
                 }
-                catch (Exception ex)
+                catch (Exception)
                 {
                     if (!_isActive)
                     {
@@ -276,12 +256,6 @@ namespace mROA.Implementation
             lock (_reconnection)
             {
                 Console.WriteLine("Got lock from {0}", source);
-                if (_isConnected || _isInReconnectionState)
-                {
-                    Console.WriteLine(
-                        $"{source} {_isConnected} {_isInReconnectionState} {!_baseStream.CanRead} {!_baseStream.CanWrite}");
-                    return;
-                }
 
                 Console.WriteLine("Call OnDisconnected from {0}", source);
                 _isInReconnectionState = true;
@@ -310,8 +284,6 @@ namespace mROA.Implementation
             {
                 _currentReceiving?.Dispose();
             }
-
-            _baseStream?.Dispose();
         }
     }
 }

mROA/Implementation/Frontend/NetworkFrontendBridge.cs

@@ -12,7 +12,7 @@ namespace mROA.Implementation.Frontend
     {
         private readonly IPEndPoint _serverEndPoint;
         private TcpClient _tcpClient = new();
-        private INextGenerationInteractionModule? _interactionModule;
+        private IChannelInteractionModule? _interactionModule;
         private ISerializationToolkit? _serialization;
 
         public NetworkFrontendBridge(IPEndPoint serverEndPoint)
@@ -24,7 +24,7 @@ namespace mROA.Implementation.Frontend
         {
             switch (dependency)
             {
-                case NextGenerationInteractionModule interactionModule:
+                case ChannelInteractionModule interactionModule:
                     _interactionModule = interactionModule;
                     break;
                 case ISerializationToolkit toolkit:
@@ -43,12 +43,14 @@ namespace mROA.Implementation.Frontend
             _tcpClient.Connect(_serverEndPoint);
 
             _interactionModule.BaseStream = _tcpClient.GetStream();
-            _interactionModule.UntrustedReceiveChanel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
+            var channel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
             {
                 SingleWriter = false,
                 SingleReader = false,
                 AllowSynchronousContinuations = true
-            }).Reader;
+            });
+            _interactionModule.UntrustedReceiveChanel = channel.Reader;
+            _interactionModule.UntrustedReceiveChanelWriter = channel.Writer;
             _interactionModule.OnDisconnected += id => { Reconnect(); };
 
             _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientConnect())).Wait();
@@ -70,12 +72,14 @@ namespace mROA.Implementation.Frontend
             _tcpClient = new TcpClient();
             _tcpClient.Connect(_serverEndPoint);
             _interactionModule.BaseStream = _tcpClient.GetStream();
-            _interactionModule.UntrustedReceiveChanel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
+            var channel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions
             {
                 SingleWriter = false,
                 SingleReader = false,
                 AllowSynchronousContinuations = true
-            }).Reader;
+            });
+            _interactionModule.UntrustedReceiveChanel = channel.Reader;
+            _interactionModule.UntrustedReceiveChanelWriter = channel.Writer;
             await _interactionModule.Restart(true);
         }
 

mROA/Implementation/RepresentationModule.cs

@@ -7,7 +7,7 @@ namespace mROA.Implementation
 {
     public class RepresentationModule : IRepresentationModule
     {
-        private INextGenerationInteractionModule? _interaction;
+        private IChannelInteractionModule? _interaction;
         private ISerializationToolkit? _serialization;
 
         public void Inject<T>(T dependency)
@@ -17,7 +17,7 @@ namespace mROA.Implementation
                 case ISerializationToolkit toolkit:
                     _serialization = toolkit;
                     break;
-                case INextGenerationInteractionModule interactionModule:
+                case IChannelInteractionModule interactionModule:
                     _interaction = interactionModule;
                     break;
             }

mROA/Implementation/StreamExtractor.cs

@@ -0,0 +1,76 @@
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using mROA.Abstract;
+
+namespace mROA.Implementation
+{
+    public class StreamExtractor
+    {
+        private readonly Stream _ioStream;
+        private readonly ISerializationToolkit _serializationToolkit;
+        private const int BufferSize = ushort.MaxValue;
+        private readonly Memory<byte> _buffer = new byte[BufferSize];
+        private bool _manualConnectionState = true;
+        public StreamExtractor(Stream ioStream, ISerializationToolkit serializationToolkit)
+        {
+            _ioStream = ioStream;
+            _serializationToolkit = serializationToolkit;
+        }
+
+        public event Action<NetworkMessageHeader> MessageReceived;
+
+        private ushort ReadMessageLength()
+        {
+            var firstBit = _ioStream.ReadByte();
+            if (firstBit == -1)
+            {
+                _manualConnectionState = false;
+                throw new EndOfStreamException();
+            }
+
+            _manualConnectionState = true;
+            var secondBit = (byte)_ioStream.ReadByte();
+
+            var len = BitConverter.ToUInt16(new[] { (byte)firstBit, secondBit });
+
+            return len;
+        }
+
+        public async Task SingleReceive()
+        {
+            var len = ReadMessageLength();
+            var localSpan = _buffer[..len];
+
+            await _ioStream.ReadExactlyAsync(localSpan);
+
+            var message = _serializationToolkit.Deserialize<NetworkMessageHeader>(localSpan.Span);
+#if TRACE
+            Console.WriteLine($"{DateTime.Now.TimeOfDay} Received Message {message.Id} - {message.MessageType}");
+            TransmissionConfig.TotalTransmittedBytes += len;
+            Console.WriteLine($"Total received bytes are {TransmissionConfig.TotalTransmittedBytes}");
+#endif
+            MessageReceived(message);
+        }
+
+        public async Task InfiniteReceive(CancellationToken token)
+        {
+            while (token.IsCancellationRequested == false)
+            {
+                await SingleReceive();
+            }
+        }
+
+        public async Task Send(NetworkMessageHeader message)
+        {
+            var rawMessage = _serializationToolkit.Serialize(message);
+            var header = BitConverter.GetBytes((ushort)rawMessage.Length).AsMemory(0, sizeof(ushort));
+
+            await _ioStream.WriteAsync(header);
+            await _ioStream.WriteAsync(rawMessage);
+        }
+
+        public bool IsConnected => _ioStream is { CanRead: true, CanWrite: true } && _manualConnectionState;
+    }
+}

mROA/mROA.csproj

@@ -21,7 +21,7 @@
     </PropertyGroup>
 
     <PropertyGroup Condition=" '$(Configuration)' == 'Debug' ">
-        <DefineConstants></DefineConstants>
+        <DefineConstants>TRACE;</DefineConstants>
     </PropertyGroup>
 
     <ItemGroup>