Новая система сетевого взаимодействия обложена тестами

mROA.Test/NextGenTest.cs

@@ -0,0 +1,71 @@
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using mROA.Implementation;
+
+namespace mROA.Test;
+
+public class NextGenTest
+{
+    private TcpListener _listener;
+    private NextGenerationInteractionModule _interactionModuleA;
+    private NextGenerationInteractionModule _interactionModuleB;
+    private Guid[] guids = [Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid()];
+
+    [SetUp]
+    public void Setup()
+    {
+        _listener = new TcpListener(IPAddress.Loopback, 4567);
+        _interactionModuleA = new NextGenerationInteractionModule();
+        _interactionModuleA.Inject(new JsonSerializationToolkit());
+        _interactionModuleB = new NextGenerationInteractionModule();
+        _interactionModuleB.Inject(new JsonSerializationToolkit());
+
+    }
+
+    [Test]
+    public void MultithreadedTest()
+    {
+        
+        Task.Run(() =>
+        {
+            _listener.Start();
+            _interactionModuleB.BaseStream = _listener.AcceptTcpClient().GetStream();
+
+            foreach (var guid in guids)
+            {
+                _interactionModuleB.PostMessage(new NetworkMessage { Id = guid, Data = "Hello user"u8.ToArray() });
+            }
+        });
+        
+        var client = new TcpClient();
+        client.Connect(IPAddress.Loopback, 4567);
+        _interactionModuleA.BaseStream = client.GetStream();
+
+        var tasks = guids.Select(ReadStream);
+
+        Task.WaitAll(tasks.ToArray());
+        Assert.Pass();
+    }
+
+    private async Task ReadStream(Guid current)
+    {
+        var msg = await _interactionModuleA.GetNextMessageReceiving();
+        Console.WriteLine(
+            $"{Environment.CurrentManagedThreadId} Received message: {Encoding.Default.GetString(new JsonSerializationToolkit().Serialize(msg))}");
+        while (msg.Id != current)
+        {
+            msg = await _interactionModuleA.GetNextMessageReceiving();
+            Console.WriteLine(
+                $"{Environment.CurrentManagedThreadId} Received message: {Encoding.Default.GetString(new JsonSerializationToolkit().Serialize(msg))}");
+        }
+
+        Console.WriteLine($"{Environment.CurrentManagedThreadId} Good message received");
+    }
+
+    [TearDown]
+    public void TearDown()
+    {
+        _listener.Dispose();
+    }
+}

mROA/Abstract/IInteractionModule.cs

@@ -1,3 +1,5 @@
+using mROA.Implementation;
+
 namespace mROA.Abstract;
 
 public interface IInteractionModule : IInjectableModule
@@ -11,4 +13,14 @@ public interface IInteractionModule : IInjectableModule
         public Task<byte[]> ReceiveMessage();
         public void PostMessage(byte[] message);
     }
+}
+
+public interface INextGenerationInteractionModule : IInjectableModule
+{
+    int ConntectionId { get; }
+    public Stream? BaseStream { get; set; }
+    Task<NetworkMessage> GetNextMessageReceiving();
+    void PostMessage(NetworkMessage message);
+    NetworkMessage[] UnhandledMessages { get; }
+    void HandleMessage(NetworkMessage msg);
 }

mROA/NextGenerationInteractionModule.cs

@@ -0,0 +1,54 @@
+using mROA.Abstract;
+using mROA.Implementation;
+
+namespace mROA;
+
+public class NextGenerationInteractionModule : INextGenerationInteractionModule
+{
+    private ISerializationToolkit _serialization;
+    public int ConntectionId { get; set; }
+    public Stream BaseStream { get; set; }
+    private Task<NetworkMessage>? _currentReceiving;
+    private const int BufferSize = ushort.MaxValue;
+    private byte[] _buffer = new byte[BufferSize];
+    private List<NetworkMessage> _unhandledMessages = new(8);
+
+    public void Inject<T>(T dependency)
+    {
+        if (dependency is ISerializationToolkit toolkit)
+            _serialization = toolkit;
+    }
+
+
+    public Task<NetworkMessage> GetNextMessageReceiving()
+    {
+        if (_currentReceiving is { IsCompleted: false })
+            return _currentReceiving;
+
+        _currentReceiving = GetNextMessage();
+        return _currentReceiving;
+    }
+
+    public void PostMessage(NetworkMessage message)
+    {
+        var rawMessage = _serialization.Serialize(message);
+        BaseStream.Write(BitConverter.GetBytes((ushort)rawMessage.Length), 0, sizeof(ushort));
+        BaseStream.Write(rawMessage, 0, rawMessage.Length);
+    }
+
+    public NetworkMessage[] UnhandledMessages => _unhandledMessages.ToArray();
+
+    public void HandleMessage(NetworkMessage msg)
+    {
+        _unhandledMessages.Remove(msg);
+    }
+
+    private async Task<NetworkMessage> GetNextMessage()
+    {
+        await BaseStream.ReadExactlyAsync(_buffer, 0, 2);
+        var len = BitConverter.ToUInt16(_buffer, 0);
+        await BaseStream.ReadExactlyAsync(_buffer, 0, len);
+        
+        return _serialization.Deserialize<NetworkMessage>(_buffer[..len])!;
+    }
+}