- Delay connections from going in game in N4E until initial NGO sync has finished

- Handle messages too large to be sent as a single RPC in N4E (using BatchedSendQueue and BatchedReceiveQueue)
- Handle RPCs being received out of order
- Switched connection flow so that connection is initialized by the transport (typical NGO flow) rather than using AutoConnectPort.

Author: ShadauxCat

com.unity.netcode.gameobjects/Runtime/Components/Helpers/NetworkObjectBridge.cs

@@ -41,7 +41,7 @@ public override bool Initialize(string defaultWorldName)
         {
             var networkManager = NetworkManager.Singleton;
             Instance = this;
-            AutoConnectPort = Port;
+            AutoConnectPort = 0;
             if (base.Initialize(defaultWorldName))
             {
                 UnityEngine.Debug.LogError($"[{nameof(UnifiedBootStrap)}] Auto-bootstrap is enabled!!! This will break the POC!");

com.unity.netcode.gameobjects/Runtime/Components/Helpers/UnifiedUpdateConnections.cs

@@ -14,6 +14,7 @@ public struct NetcodeConnection
         public int NetworkId;
 
         internal float ConnectedTime;
+        internal bool IsSynced;
 
         public bool IsServer => World.IsServer();
         public void GoInGame()
@@ -34,6 +35,15 @@ internal partial class UnifiedUpdateConnections : SystemBase
 
         private Dictionary<int, NetcodeConnection> m_NewConnections = new Dictionary<int, NetcodeConnection>();
 
+        public void MarkSync(int NetworkId)
+        {
+            if (m_NewConnections.TryGetValue(NetworkId, out var connection))
+            {
+                connection.IsSynced = true;
+                m_NewConnections[NetworkId] = connection;
+            }
+        }
+        
         protected override void OnUpdate()
         {
             var isServer = World.IsServer();
@@ -59,7 +69,7 @@ protected override void OnUpdate()
                 if (!m_NewConnections.ContainsKey(networkId.Value))
                 {
                     var delayTime = 0.0f;// isServer ? 0.2f : 0.1f;
-                    var newConnection = new NetcodeConnection { World = World, Entity = entity, NetworkId = networkId.Value, ConnectedTime = UnityEngine.Time.realtimeSinceStartup + delayTime};
+                    var newConnection = new NetcodeConnection { World = World, Entity = entity, NetworkId = networkId.Value, ConnectedTime = UnityEngine.Time.realtimeSinceStartup + delayTime, IsSynced = isServer};
                     m_NewConnections.Add(networkId.Value, newConnection);
                 }
             }
@@ -70,7 +80,7 @@ protected override void OnUpdate()
                 foreach (var entry in m_NewConnections)
                 {
                     // Check if the delay time has passed.
-                    if (entry.Value.ConnectedTime < UnityEngine.Time.realtimeSinceStartup)
+                    if (entry.Value.IsSynced && entry.Value.ConnectedTime < UnityEngine.Time.realtimeSinceStartup)
                     {
                         // Set the connection in-game
                         commandBuffer.AddComponent<NetworkStreamInGame>(entry.Value.Entity);

com.unity.netcode.gameobjects/Runtime/Messaging/Messages/SceneEventMessage.cs

@@ -1,4 +1,7 @@
 
+using Unity.Netcode.Components;
+using Unity.Netcode.Unified;
+
 namespace Unity.Netcode
 {
     // Todo: Would be lovely to get this one nicely formatted with all the data it sends in the struct
@@ -27,6 +30,11 @@ public void Handle(ref NetworkContext context)
         {
             var networkManager = (NetworkManager)context.SystemOwner;
             networkManager.SceneManager.HandleSceneEvent(context.SenderId, m_ReceivedData);
+            
+#if UNIFIED_NETCODE
+            var unifiedConnectionSystem = NetCode.Netcode.GetWorld(false).GetExistingSystemManaged<UnifiedUpdateConnections>();
+            unifiedConnectionSystem.MarkSync((int)context.SenderId);
+#endif
         }
     }
 }

com.unity.netcode.gameobjects/Runtime/Transports/Unified/UnifiedNetcodeTransport.cs

@@ -8,6 +8,7 @@
 using Unity.Collections.LowLevel.Unsafe;
 using Unity.Entities;
 using Unity.NetCode;
+using Unity.Netcode.Transports.UTP;
 using Unity.Networking.Transport;
 using UnityEngine;
 
@@ -16,37 +17,19 @@ namespace Unity.Netcode.Unified
     internal struct TransportRpc : IRpcCommand, IRpcCommandSerializer<TransportRpc>
     {
         public FixedList4096Bytes<byte> Buffer;
-        
-        internal static string ByteArrayToString(FixedList4096Bytes<byte> ba, int offset, int count)
-        {
-            var hex = new StringBuilder(ba.Length * 2);
-            for (int i = offset; i < offset + count; ++i)
-            {
-                hex.AppendFormat("{0:x2} ", ba[i]);
-            }
-
-            return hex.ToString();
-        }
-        internal static string ByteArrayToString(NativeArray<byte> ba, int offset, int count)
-        {
-            var hex = new StringBuilder(ba.Length * 2);
-            for (int i = offset; i < offset + count; ++i)
-            {
-                hex.AppendFormat("{0:x2} ", ba[i]);
-            }
-
-            return hex.ToString();
-        }
+        public ulong Order;
 
         public unsafe void Serialize(ref DataStreamWriter writer, in RpcSerializerState state, in TransportRpc data)
         {
+            writer.WriteULong(data.Order);
             writer.WriteInt(data.Buffer.Length);
             var span = new Span<byte>(data.Buffer.GetUnsafePtr(), data.Buffer.Length);
             writer.WriteBytes(span);
         }
 
         public unsafe void Deserialize(ref DataStreamReader reader, in RpcDeserializerState state, ref TransportRpc data)
         {
+            data.Order = reader.ReadULong();
             var length = reader.ReadInt();
             data.Buffer = new FixedList4096Bytes<byte>();
             data.Buffer.Length = length;
@@ -104,13 +87,13 @@ internal partial class UnifiedNetcodeUpdateSystem : SystemBase
     {
         public UnifiedNetcodeTransport Transport;
 
-        public List<Connection> DiscconedtQueue = new List<Connection>();
-
+        public List<Connection> DisconnectQueue = new List<Connection>();
+        
         public void Disconnect(Connection connection)
         {
-            DiscconedtQueue.Add(connection);
+            DisconnectQueue.Add(connection);
         }
-        
+
         protected override void OnUpdate()
         {
             using var commandBuffer = new EntityCommandBuffer(Allocator.Temp);
@@ -119,16 +102,24 @@ protected override void OnUpdate()
                 var connectionId = SystemAPI.GetComponent<NetworkId>(request.ValueRO.SourceConnection).Value;
 
                 var buffer = rpc.ValueRW.Buffer;
-                Transport.DispatchMessage(connectionId, buffer);
-                commandBuffer.DestroyEntity(entity);
+                try
+                {
+                    Transport.DispatchMessage(connectionId, buffer, rpc.ValueRO.Order);
+                }
+                finally
+                {
+                    commandBuffer.DestroyEntity(entity);
+                }
             }
 
-            foreach (var connection in DiscconedtQueue)
+            foreach (var connection in DisconnectQueue)
             {
                 commandBuffer.AddComponent<NetworkStreamRequestDisconnect>(connection.ConnectionEntity);
             }
+            DisconnectQueue.Clear();
+
             commandBuffer.Playback(EntityManager);
-            DiscconedtQueue.Clear();
+                
         }
     }
 
@@ -144,36 +135,90 @@ internal class UnifiedNetcodeTransport : NetworkTransport
 
         private IRealTimeProvider m_RealTimeProvider;
 
-        private Dictionary<int, Connection> m_Connections;
-
-        internal void DispatchMessage(int connectionId, FixedList4096Bytes<byte> buffer)
+        private class ConnectionInfo
         {
-            ArraySegment<byte> data = new ArraySegment<byte>(buffer.ToArray());
-            InvokeOnTransportEvent(NetworkEvent.Data, (ulong)connectionId, data, m_RealTimeProvider.RealTimeSinceStartup);
+            public BatchedSendQueue SendQueue;
+            public BatchedReceiveQueue ReceiveQueue;
+            public Connection Connection;
+            public ulong LastSent;
+            public ulong LastReceived;
+            public Dictionary<ulong, FixedList4096Bytes<byte>> DeferredMessages;
         }
+ 
+        private Dictionary<int, ConnectionInfo> m_Connections;
 
-        public override void Send(ulong clientId, ArraySegment<byte> payload, NetworkDelivery networkDelivery)
+        internal void DispatchMessage(int connectionId, FixedList4096Bytes<byte> buffer, ulong order)
         {
-            if (!m_Connections.TryGetValue((int)clientId, out Connection connection))
+            var connectionInfo = m_Connections[connectionId];
+
+            if (order != connectionInfo.LastReceived + 1)
             {
+                if (connectionInfo.DeferredMessages == null)
+                {
+                    connectionInfo.DeferredMessages = new Dictionary<ulong, FixedList4096Bytes<byte>>();
+                }
+
+                connectionInfo.DeferredMessages[order] = buffer;
                 return;
             }
 
-            var rpc = new TransportRpc
+            var reader = new DataStreamReader(buffer.ToNativeArray(Allocator.Temp));
+            if (connectionInfo.ReceiveQueue == null)
             {
-                Buffer = new FixedList4096Bytes<byte>(),
-            };
-            
-            unsafe
+                connectionInfo.ReceiveQueue = new BatchedReceiveQueue(reader);
+            }
+            else
+            {
+                connectionInfo.ReceiveQueue.PushReader(reader);
+            }
+
+            connectionInfo.LastReceived = order;
+            if (connectionInfo.DeferredMessages != null)
             {
-                rpc.Buffer.Length = payload.Count;
-                fixed (byte* data = payload.Array)
+                var next = order + 1;
+                while (connectionInfo.DeferredMessages.Remove(next, out var nextBuffer))
                 {
-                    UnsafeUtility.MemCpy(rpc.Buffer.GetUnsafePtr(), (void*)(data + payload.Offset), payload.Count);
+                    reader = new DataStreamReader(nextBuffer.ToNativeArray(Allocator.Temp));
+                    connectionInfo.ReceiveQueue.PushReader(reader);
+                    connectionInfo.LastReceived = next;
+                    ++next;
                 }
             }
 
-            connection.SendMessage(rpc);
+            var message = connectionInfo.ReceiveQueue.PopMessage();
+            while (message.Count != 0)
+            {
+                InvokeOnTransportEvent(NetworkEvent.Data, (ulong)connectionId, message,
+                    m_RealTimeProvider.RealTimeSinceStartup);
+                message = connectionInfo.ReceiveQueue.PopMessage();
+            }
+        }
+
+        public override unsafe void Send(ulong clientId, ArraySegment<byte> payload, NetworkDelivery networkDelivery)
+        {
+            if (!m_Connections.TryGetValue((int)clientId, out ConnectionInfo connectionInfo))
+            {
+                return;
+            }
+
+            connectionInfo.SendQueue.PushMessage(payload);
+
+            while (!connectionInfo.SendQueue.IsEmpty)
+            {
+                var rpc = new TransportRpc
+                {
+                    Buffer = new FixedList4096Bytes<byte>(),
+                };
+                var writer = new DataStreamWriter(rpc.Buffer.GetUnsafePtr(), 1024);
+
+                var amount = connectionInfo.SendQueue.FillWriterWithBytes(ref writer, 1024);
+                rpc.Buffer.Length = amount;
+                rpc.Order = ++connectionInfo.LastSent;
+
+                connectionInfo.Connection.SendMessage(rpc);
+                
+                connectionInfo.SendQueue.Consume(amount);
+            }
         }
 
         public override NetworkEvent PollEvent(out ulong clientId, out ArraySegment<byte> payload, out float receiveTime)
@@ -186,14 +231,24 @@ public override NetworkEvent PollEvent(out ulong clientId, out ArraySegment<byte
 
         private void OnClientConnectedToServer(Connection connection, NetCodeConnectionEvent connectionEvent)
         {
-            m_Connections[connection.NetworkId.Value] = connection;
+            m_Connections[connection.NetworkId.Value] = new ConnectionInfo
+            {
+                ReceiveQueue = null,
+                SendQueue = new BatchedSendQueue(BatchedSendQueue.MaximumMaximumCapacity),
+                Connection = connection
+            };
             m_ServerClientId = connection.NetworkId.Value;
             InvokeOnTransportEvent(NetworkEvent.Connect, (ulong)connection.NetworkId.Value, default,  m_RealTimeProvider.RealTimeSinceStartup);
         }
 
         private void OnServerNewClientConnection(Connection connection, NetCodeConnectionEvent connectionEvent)
         {
-            m_Connections[connection.NetworkId.Value] = connection;
+            m_Connections[connection.NetworkId.Value] = new ConnectionInfo
+            {
+                ReceiveQueue = null,
+                SendQueue = new BatchedSendQueue(BatchedSendQueue.MaximumMaximumCapacity),
+                Connection = connection
+            };;
             InvokeOnTransportEvent(NetworkEvent.Connect, (ulong)connection.NetworkId.Value, default,  m_RealTimeProvider.RealTimeSinceStartup);
         }
 
@@ -219,6 +274,9 @@ public override bool StartClient()
             NetCode.Netcode.Client.OnDisconnect = OnClientDisconnectFromServer;
             var updateSystem = NetCode.Netcode.GetWorld(false).GetExistingSystemManaged<UnifiedNetcodeUpdateSystem>();
             updateSystem.Transport = this;
+            using var drvQuery = updateSystem.EntityManager.CreateEntityQuery(ComponentType.ReadWrite<NetworkStreamDriver>());
+            var driver = drvQuery.GetSingletonRW<NetworkStreamDriver>();
+            driver.ValueRW.Connect(updateSystem.EntityManager, NetworkEndpoint.Parse("127.0.0.1", 7979));
             return true;
         }
 
@@ -241,20 +299,23 @@ public override bool StartServer()
             NetCode.Netcode.Server.OnDisconnect = OnServerClientDisconnected;
             var updateSystem = NetCode.Netcode.GetWorld(true).GetExistingSystemManaged<UnifiedNetcodeUpdateSystem>();
             updateSystem.Transport = this;
+            using var drvQuery = updateSystem.EntityManager.CreateEntityQuery(ComponentType.ReadWrite<NetworkStreamDriver>());
+            var driver = drvQuery.GetSingletonRW<NetworkStreamDriver>();
+            driver.ValueRW.Listen(NetworkEndpoint.Parse("127.0.0.1", 7979));
             return true;
         }
 
         public override void DisconnectRemoteClient(ulong clientId)
         {
             var updateSystem = NetCode.Netcode.GetWorld(true).GetExistingSystemManaged<UnifiedNetcodeUpdateSystem>();
-            updateSystem.Disconnect(m_Connections[(int)clientId]);
+            updateSystem.Disconnect(m_Connections[(int)clientId].Connection);
             m_Connections.Remove((int)clientId);
         }
 
         public override void DisconnectLocalClient()
         {
             var updateSystem = NetCode.Netcode.GetWorld(false).GetExistingSystemManaged<UnifiedNetcodeUpdateSystem>();
-            updateSystem.Disconnect(m_Connections[(int)ServerClientId]);
+            updateSystem.Disconnect(m_Connections[(int)ServerClientId].Connection);
             m_Connections.Remove((int)ServerClientId);
         }
 
@@ -279,7 +340,7 @@ public override void Shutdown()
 
         public override void Initialize(NetworkManager networkManager = null)
         {
-            m_Connections = new Dictionary<int, Connection>();
+            m_Connections = new Dictionary<int, ConnectionInfo>();
             m_RealTimeProvider = networkManager.RealTimeProvider;
         }
     }