From 5aa1afb45208b6a3ee70107824025a95f0cc23a6 Mon Sep 17 00:00:00 2001 From: James Frowen Date: Fri, 7 Jun 2024 15:18:53 +0100 Subject: [PATCH] feat: starting idea to send large message via NACK --- .../Serialization/LargeMessageSender.cs | 119 ++++++++++++++++++ .../Serialization/LargeMessageSender.cs.meta | 11 ++ 2 files changed, 130 insertions(+) create mode 100644 Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs create mode 100644 Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs.meta diff --git a/Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs b/Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs new file mode 100644 index 0000000000..9303b17507 --- /dev/null +++ b/Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs @@ -0,0 +1,119 @@ +using System.Collections.Generic; +using Cysharp.Threading.Tasks; +using Mirage.SocketLayer; + +namespace Mirage.Serialization +{ + public class LargeMessageSender + { + private readonly int _maxPacketSize; + private readonly IMessageReceiver _messageHandler; + private uint _largeMessageId; + + public LargeMessageSender(int maxPacketSize, IMessageReceiver messageHandler) + { + _maxPacketSize = maxPacketSize; + _messageHandler = messageHandler; + } + + public async UniTask SendLargeMessage(IConnection connection, byte[] data, int offset, int length) + { + var notAcked = new List(); + SendInternal(connection, notAcked, data, offset, length); + + // todo wait for NACK message, + // if empty that means all parts have been acked + // other peer should be sending NACK message short delay after receiving first packet, and repeat + await UniTask.Yield(); + } + + private void SendInternal(IConnection connection, List notAcked, byte[] data, int offset, int length) + { + // send type, Id (if this message), current fragment, max fragment. + //const int header = 1 + 4 + 4 + 4; + var sizePerFragment = 1000;//_maxPacketSize - header; + var fragments = length / sizePerFragment; + var extra = length % sizePerFragment; + _largeMessageId++; + using (var writer = NetworkWriterPool.GetWriter()) + { + { + // send meta data + var meta = new LargeMessageFragmentMeta + { + Id = _largeMessageId, + Fragments = checked((uint)fragments), + Extra = checked((uint)extra) + }; + MessagePacker.Pack(meta, writer); + connection.SendReliable(writer.ToArraySegment()); + writer.Reset(); + } + + // send real data + for (var i = 0; i < fragments; i++) + { + var msg = new LargeMessageFragment + { + Id = _largeMessageId, + Fragment = checked((uint)i) + }; + notAcked.Add(checked((uint)i)); + // write head + MessagePacker.Pack(msg, writer); + // then write bytes + writer.WriteBytes(data, offset + (i * sizePerFragment), sizePerFragment); + connection.SendReliable(writer.ToArraySegment()); + writer.Reset(); + } + + if (extra != 0) + { + var msg = new LargeMessageFragment + { + Id = _largeMessageId, + Fragment = checked((uint)fragments) + }; + notAcked.Add(checked((uint)fragments)); + // write head + MessagePacker.Pack(msg, writer); + // then write bytes + writer.WriteBytes(data, offset + (fragments * sizePerFragment), extra); + connection.SendReliable(writer.ToArraySegment()); + writer.Reset(); + } + } + } + + public struct LargeMessageFragmentMeta + { + public uint Id; + public uint Fragments; + public uint Extra; + } + public struct LargeMessageFragment + { + public uint Id; + public uint Fragment; + public NetworkReader Reader; + } + public struct LargeMessageFragmentNotAck + { + public uint Id; + public List Fragment; + } + } + + public static class E + { + public static void WriteNetworkReader(this NetworkWriter writer, NetworkReader reader) + { + // nothing + } + + public static NetworkReader ReadNetworkReader(this NetworkReader reader) + { + return reader; + } + } +} diff --git a/Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs.meta b/Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs.meta new file mode 100644 index 0000000000..02cff3b321 --- /dev/null +++ b/Assets/Mirage/Runtime/Serialization/LargeMessageSender.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 822717abf633011438e1968f80106c41 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: