diff --git a/Example.Backend/Program.cs b/Example.Backend/Program.cs index 1e12d48..719d274 100644 --- a/Example.Backend/Program.cs +++ b/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()!); - builder.UseNetworkGateway(new IPEndPoint(IPAddress.Loopback, 4567), typeof(NextGenerationInteractionModule), + builder.UseNetworkGateway(new IPEndPoint(IPAddress.Loopback, 4567), typeof(ChannelInteractionModule), builder.GetModule()!); builder.Modules.Add(new ConnectionHub()); diff --git a/Example.Frontend/Program.cs b/Example.Frontend/Program.cs index 8ecc3f4..e87a449 100644 --- a/Example.Frontend/Program.cs +++ b/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); diff --git a/mROA.Test/NextGenTest.cs b/mROA.Test/NextGenTest.cs index 0b708a7..64c6d86 100644 --- a/mROA.Test/NextGenTest.cs +++ b/mROA.Test/NextGenTest.cs @@ -11,17 +11,17 @@ namespace mROA.Test public class NextGenTest { private TcpListener _listener; - private NextGenerationInteractionModule _interactionModuleA; - private NextGenerationInteractionModule _interactionModuleB; + private ChannelInteractionModule _interactionModuleA; + 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()); } diff --git a/mROA/Abstract/IConnectionHub.cs b/mROA/Abstract/IConnectionHub.cs index 3c6afd4..01f60b9 100644 --- a/mROA/Abstract/IConnectionHub.cs +++ b/mROA/Abstract/IConnectionHub.cs @@ -6,8 +6,8 @@ public interface IConnectionHub : IInjectableModule { - void RegisterInteraction(INextGenerationInteractionModule interaction); - INextGenerationInteractionModule GetInteraction(int id); + void RegisterInteraction(IChannelInteractionModule interaction); + IChannelInteractionModule GetInteraction(int id); event ConnectionHandler? OnConnected; event DisconnectionHandler? OnDisconnected; } diff --git a/mROA/Abstract/IInteractionModule.cs b/mROA/Abstract/IInteractionModule.cs index fae3768..cffade7 100644 --- a/mROA/Abstract/IInteractionModule.cs +++ b/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; } - ChannelReader UntrustedReceiveChanel { get; set; } - ChannelWriter<(int clientId, NetworkMessageHeader messageHeader)> UntrustedPostChanel { get; set; } - + ChannelWriter ReceiveChanel { get; } + ChannelReader TrustedPostChanel { get; } + ChannelReader UntrustedPostChanel { get; } + Action IsConnected { get; set; } Task GetNextMessageReceiving(bool infinite = true); Task PostMessageAsync(NetworkMessageHeader messageHeader); Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader); void HandleMessage(NetworkMessageHeader messageHeader); - // NetworkMessageHeader[] UnhandledMessages { get; } NetworkMessageHeader? FirstByFilter(Predicate predicate); event Action OnDisconnected; Task Restart(bool sendRecovery); diff --git a/mROA/Implementation/Backend/ConnectionHub.cs b/mROA/Implementation/Backend/ConnectionHub.cs index de302b0..22b7b6b 100644 --- a/mROA/Implementation/Backend/ConnectionHub.cs +++ b/mROA/Implementation/Backend/ConnectionHub.cs @@ -6,10 +6,10 @@ namespace mROA.Implementation.Backend { public class ConnectionHub : IConnectionHub { - private readonly Dictionary _connections = new(); + private readonly Dictionary _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"); } diff --git a/mROA/Implementation/Backend/NetworkGatewayModule.cs b/mROA/Implementation/Backend/NetworkGatewayModule.cs index f756920..4c3f22b 100644 --- a/mROA/Implementation/Backend/NetworkGatewayModule.cs +++ b/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(new UnboundedChannelOptions + var channel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleWriter = false, SingleReader = false, AllowSynchronousContinuations = true - }).Reader; + }); + interaction.UntrustedReceiveChanel = channel.Reader; + interaction.UntrustedReceiveChanelWriter = channel.Writer; var connectionRequest = interaction.GetNextMessageReceiving(false) .GetAwaiter().GetResult()!; diff --git a/mROA/Implementation/NextGenerationInteractionModule.cs b/mROA/Implementation/ChannelInteractionModule.cs similarity index 76% rename from mROA/Implementation/NextGenerationInteractionModule.cs rename to mROA/Implementation/ChannelInteractionModule.cs index 92d63a3..e5d51ba 100644 --- a/mROA/Implementation/NextGenerationInteractionModule.cs +++ b/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 _receiveReader; + private readonly Channel _inputChannel; + private readonly Channel _outputTrustedChannel; + private readonly Channel _outputUntrustedChannel; private const int BufferSize = ushort.MaxValue; private readonly Memory _buffer = new byte[BufferSize]; private readonly List _messageBuffer = new(128); private Task? _currentReceiving; private ISerializationToolkit? _serialization; - private Stream? _baseStream; private bool _isConnected = true; private bool _isInReconnectionState; private bool _isActive = true; private TaskCompletionSource _reconnection; - private ValueTask? _trustedReceive; - private TaskCompletionSource _untrustedReceive; - public NextGenerationInteractionModule() + public ChannelInteractionModule() { _reconnection = new TaskCompletionSource(); + _inputChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = false, + SingleWriter = false, + AllowSynchronousContinuations = true + }); + _receiveReader = _inputChannel.Reader; + _outputTrustedChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = true, + AllowSynchronousContinuations = true + }); + _outputUntrustedChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = true, + AllowSynchronousContinuations = true + }); } public int ConnectionId { get; set; } - public Stream? BaseStream - { - get => _baseStream; - set => _baseStream = value; - } - - public ChannelReader UntrustedReceiveChanel { get; set; } - public ChannelWriter<(int clientId, NetworkMessageHeader messageHeader)> UntrustedPostChanel { get; set; } + public ChannelWriter ReceiveChanel => _inputChannel.Writer; + public ChannelReader TrustedPostChanel => _outputTrustedChannel.Reader; + public ChannelReader UntrustedPostChanel => _outputUntrustedChannel.Reader; + public Action IsConnected { get; set; } public void Inject(T dependency) @@ -152,46 +168,10 @@ namespace mROA.Implementation try { - NetworkMessageHeader message; - - var wasNull = _untrustedReceive is null; - - _trustedReceive ??= Receive(); - _untrustedReceive = new TaskCompletionSource(); - - - if (!wasNull) - { - if (_trustedReceive.Value.IsCompleted) - { - _trustedReceive = Receive(); - } - - if (_untrustedReceive.Task.IsCompleted) - { - _untrustedReceive = new TaskCompletionSource(); - _ = 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()); - + var message = await _receiveReader.ReadAsync(); 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(); } } } \ No newline at end of file diff --git a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs index 27e8219..2a5ec69 100644 --- a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs +++ b/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(new UnboundedChannelOptions + var channel = Channel.CreateUnbounded(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(new UnboundedChannelOptions + var channel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleWriter = false, SingleReader = false, AllowSynchronousContinuations = true - }).Reader; + }); + _interactionModule.UntrustedReceiveChanel = channel.Reader; + _interactionModule.UntrustedReceiveChanelWriter = channel.Writer; await _interactionModule.Restart(true); } diff --git a/mROA/Implementation/RepresentationModule.cs b/mROA/Implementation/RepresentationModule.cs index 6ea6fcd..d758c6f 100644 --- a/mROA/Implementation/RepresentationModule.cs +++ b/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 dependency) @@ -17,7 +17,7 @@ namespace mROA.Implementation case ISerializationToolkit toolkit: _serialization = toolkit; break; - case INextGenerationInteractionModule interactionModule: + case IChannelInteractionModule interactionModule: _interaction = interactionModule; break; } diff --git a/mROA/Implementation/StreamExtractor.cs b/mROA/Implementation/StreamExtractor.cs new file mode 100644 index 0000000..f9f71c5 --- /dev/null +++ b/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 _buffer = new byte[BufferSize]; + private bool _manualConnectionState = true; + public StreamExtractor(Stream ioStream, ISerializationToolkit serializationToolkit) + { + _ioStream = ioStream; + _serializationToolkit = serializationToolkit; + } + + public event Action 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(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; + } +} \ No newline at end of file diff --git a/mROA/mROA.csproj b/mROA/mROA.csproj index 3c172de..e52150a 100644 --- a/mROA/mROA.csproj +++ b/mROA/mROA.csproj @@ -21,7 +21,7 @@ - + TRACE;