Refactor: initial state sync

Explorer/Assets/DCL/Multiplayer/Connections/DCL.Multiplayer.Connections.asmdef

@@ -16,7 +16,10 @@
     "GUID:e0eedfa2deb9406daf86fd8368728e39",
     "GUID:3640f3c0b42946b0b8794a1ed8e06ca5",
     "GUID:d414ef88f3b15f746a4b97636b50dfb4",
-    "GUID:ca4e81cdd6a34d1aa54c32ad41fc5b3b"
+    "GUID:ca4e81cdd6a34d1aa54c32ad41fc5b3b",
+    "GUID:1d2c76eb8b48e0b40940e8b31a679ce1",
+    "GUID:286980af24684da6acc1caa413039811",
+    "GUID:e25ef972de004615a22937e739de2def"
   ],
   "includePlatforms": [],
   "excludePlatforms": [],

Explorer/Assets/DCL/Multiplayer/Connections/Messaging/Hubs/MessagePipesHub.cs

@@ -11,8 +11,10 @@ public class MessagePipesHub : IMessagePipesHub
         private readonly IMessagePipe islandPipe;
 
         public MessagePipesHub(IRoomHub roomHub, IMultiPool sendingMultiPool, IMultiPool receivingMultiPool, IMemoryPool memoryPool) : this(
-            new MessagePipe(roomHub.SceneRoom().DataPipe, sendingMultiPool, receivingMultiPool, memoryPool).WithLog("Scene"),
-            new MessagePipe(roomHub.IslandRoom().DataPipe, sendingMultiPool, receivingMultiPool, memoryPool).WithLog("Island")
+            new MessagePipe(roomHub.SceneRoom().DataPipe, sendingMultiPool, receivingMultiPool, memoryPool)
+               .WithLog("Scene"),
+            new MessagePipe(roomHub.IslandRoom().DataPipe, sendingMultiPool, receivingMultiPool, memoryPool)
+               .WithLog("Island")
         ) { }
 
         public MessagePipesHub(IMessagePipe scenePipe, IMessagePipe islandPipe)

Explorer/Assets/DCL/Multiplayer/Connections/Messaging/Pipe/IMessagePipe.cs

@@ -1,5 +1,7 @@
 using Decentraland.Kernel.Comms.Rfc4;
+using ECS.SceneLifeCycle;
 using Google.Protobuf;
+using LiveKit.Rooms;
 using System;
 
 namespace DCL.Multiplayer.Connections.Messaging.Pipe

Explorer/Assets/Scripts/CrdtEcsBridge/JsModulesImplementation/Communications/CommunicationsControllerAPIImplementation.cs

@@ -1,12 +1,9 @@
 using CRDT.Memory;
-using CrdtEcsBridge.PoolsProviders;
 using SceneRunner.Scene;
 using SceneRuntime;
 using System;
 using System.Buffers;
-using System.Collections.Generic;
 using System.Text;
-using Utility;
 
 namespace CrdtEcsBridge.JsModulesImplementation.Communications
 {
@@ -16,7 +13,7 @@ public class CommunicationsControllerAPIImplementation : CommunicationsControlle
 
         public CommunicationsControllerAPIImplementation(
             ISceneData sceneData,
-            ICommunicationControllerHub messagePipesHub,
+            ISceneCommunicationPipe messagePipesHub,
             IJsOperations jsOperations,
             ICRDTMemoryAllocator crdtMemoryAllocator,
             ISceneStateProvider sceneStateProvider) : base(
@@ -28,18 +25,15 @@ public CommunicationsControllerAPIImplementation(
             this.crdtMemoryAllocator = crdtMemoryAllocator;
         }
 
-        protected override void OnMessageReceived(ICommunicationControllerHub.SceneMessage receivedMessage)
+        protected override void OnMessageReceived(MsgType messageType, ReadOnlySpan<byte> decodedMessage, string fromWalletId)
         {
-            ReadOnlySpan<byte> decodedMessage = receivedMessage.Data.Span;
-            MsgType msgType = DecodeMessage(ref decodedMessage);
-
-            if (msgType != MsgType.Uint8Array || decodedMessage.Length == 0)
+            if (messageType != MsgType.Uint8Array)
                 return;
 
             // Wallet Id
-            int walletBytesCount = Encoding.UTF8.GetByteCount(receivedMessage.WalletId);
+            int walletBytesCount = Encoding.UTF8.GetByteCount(fromWalletId);
             Span<byte> senderBytes = stackalloc byte[walletBytesCount];
-            Encoding.UTF8.GetBytes(receivedMessage.WalletId, senderBytes);
+            Encoding.UTF8.GetBytes(fromWalletId, senderBytes);
 
             int messageLength = senderBytes.Length + decodedMessage.Length + 1;
 

Explorer/Assets/Scripts/CrdtEcsBridge/JsModulesImplementation/Communications/CommunicationsControllerAPIImplementationBase.cs

@@ -1,6 +1,4 @@
 using CrdtEcsBridge.PoolsProviders;
-using DCL.Multiplayer.Connections.Messaging;
-using Decentraland.Kernel.Comms.Rfc4;
 using SceneRunner.Scene;
 using SceneRuntime;
 using SceneRuntime.Apis.Modules.CommunicationsControllerApi;
@@ -12,31 +10,32 @@
 
 namespace CrdtEcsBridge.JsModulesImplementation.Communications
 {
-    public class CommunicationsControllerAPIImplementationBase : ICommunicationsControllerAPI
+    public abstract class CommunicationsControllerAPIImplementationBase : ICommunicationsControllerAPI
     {
-        internal enum MsgType
+        public enum MsgType
         {
             String = 1, // SDK scenes MessageBus messages
             Uint8Array = 2,
         }
 
-        protected readonly CancellationTokenSource cancellationTokenSource = new ();
-        protected readonly ICommunicationControllerHub messagePipesHub;
-        protected readonly ISceneData sceneData;
-        protected readonly ISceneStateProvider sceneStateProvider;
-        protected readonly IJsOperations jsOperations;
-        protected readonly Action<ICommunicationControllerHub.SceneMessage> onMessageReceivedCached;
         protected readonly List<IMemoryOwner<byte>> eventsToProcess = new ();
+        private readonly CancellationTokenSource cancellationTokenSource = new ();
+        private readonly ISceneCommunicationPipe sceneCommunicationPipe;
+        private readonly ISceneData sceneData;
+        private readonly ISceneStateProvider sceneStateProvider;
+        private readonly IJsOperations jsOperations;
+        private readonly Action<ISceneCommunicationPipe.SceneMessage> onMessageReceivedCached;
+
         internal IReadOnlyList<IMemoryOwner<byte>> EventsToProcess => eventsToProcess;
 
-        public CommunicationsControllerAPIImplementationBase(
+        protected CommunicationsControllerAPIImplementationBase(
             ISceneData sceneData,
-            ICommunicationControllerHub messagePipesHub,
+            ISceneCommunicationPipe sceneCommunicationPipe,
             IJsOperations jsOperations,
             ISceneStateProvider sceneStateProvider)
         {
             this.sceneData = sceneData;
-            this.messagePipesHub = messagePipesHub;
+            this.sceneCommunicationPipe = sceneCommunicationPipe;
             this.jsOperations = jsOperations;
             this.sceneStateProvider = sceneStateProvider;
 
@@ -53,9 +52,9 @@ public void Dispose()
         public void OnSceneIsCurrentChanged(bool isCurrent)
         {
             if (isCurrent)
-                messagePipesHub.SetSceneMessageHandler(onMessageReceivedCached);
+                sceneCommunicationPipe.SetSceneMessageHandler(onMessageReceivedCached);
             else
-                messagePipesHub.RemoveSceneMessageHandler(onMessageReceivedCached);
+                sceneCommunicationPipe.RemoveSceneMessageHandler(onMessageReceivedCached);
         }
 
         public object SendBinary(IReadOnlyList<PoolableByteArray> data)
@@ -64,22 +63,8 @@ public object SendBinary(IReadOnlyList<PoolableByteArray> data)
                 return jsOperations.ConvertToScriptTypedArrays(Array.Empty<IMemoryOwner<byte>>());
 
             foreach (PoolableByteArray poolable in data)
-            {
-                if (poolable.Length == 0)
-                    continue;
-
-                Memory<byte> message = poolable.Memory;
-
-                EncodeAndSend();
-
-                void EncodeAndSend()
-                {
-                    Span<byte> encodedMessage = stackalloc byte[message.Length + 1];
-                    encodedMessage[0] = (byte)MsgType.Uint8Array;
-                    message.Span.CopyTo(encodedMessage[1..]);
-                    SendMessage(encodedMessage);
-                }
-            }
+                if (poolable.Length > 0)
+                    EncodeAndSendMessage(MsgType.Uint8Array, poolable.Memory.Span);
 
             lock (eventsToProcess)
             {
@@ -98,18 +83,32 @@ private void CleanUpReceivedMessages()
             eventsToProcess.Clear();
         }
 
-        private void SendMessage(ReadOnlySpan<byte> message)
+        protected void EncodeAndSendMessage(MsgType msgType, ReadOnlySpan<byte> message)
         {
-            messagePipesHub.SendMessage(message, sceneData.SceneEntityDefinition.id!, cancellationTokenSource.Token);
+            Span<byte> encodedMessage = stackalloc byte[message.Length + 1];
+            encodedMessage[0] = (byte)msgType;
+            message.CopyTo(encodedMessage[1..]);
+            sceneCommunicationPipe.SendMessage(encodedMessage, sceneData.SceneEntityDefinition.id!, cancellationTokenSource.Token);
         }
 
-        protected virtual void OnMessageReceived(ICommunicationControllerHub.SceneMessage receivedMessage) { }
-
-        internal static MsgType DecodeMessage(ref ReadOnlySpan<byte> value)
+        private static MsgType DecodeMessage(ref ReadOnlySpan<byte> value)
         {
             var msgType = (MsgType)value[0];
             value = value[1..];
             return msgType;
         }
+
+        private void OnMessageReceived(ISceneCommunicationPipe.SceneMessage receivedMessage)
+        {
+            ReadOnlySpan<byte> decodedMessage = receivedMessage.Data.Span;
+            MsgType msgType = DecodeMessage(ref decodedMessage);
+
+            if (decodedMessage.Length == 0)
+                return;
+
+            OnMessageReceived(msgType, decodedMessage, receivedMessage.FromWalletId);
+        }
+
+        protected abstract void OnMessageReceived(MsgType messageType, ReadOnlySpan<byte> decodedMessage, string fromWalletId);
     }
 }

Explorer/Assets/Scripts/CrdtEcsBridge/JsModulesImplementation/Communications/ICommunicationControllerHub.cs → Explorer/Assets/Scripts/CrdtEcsBridge/JsModulesImplementation/Communications/ISceneCommunicationPipe.cs

@@ -5,7 +5,7 @@
 
 namespace CrdtEcsBridge.JsModulesImplementation.Communications
 {
-    public interface ICommunicationControllerHub
+    public interface ISceneCommunicationPipe
     {
         void SetSceneMessageHandler(Action<SceneMessage> onSceneMessage);
 
@@ -17,13 +17,13 @@ readonly struct SceneMessage
         {
             public readonly ReadOnlyMemory<byte> Data;
             public readonly string SceneId;
-            public readonly string WalletId;
+            public readonly string FromWalletId;
 
             private SceneMessage(in ReceivedMessage<Scene> message)
             {
                 Data = message.Payload.Data.Memory;
                 SceneId = message.Payload.SceneId;
-                WalletId = message.FromWalletId;
+                FromWalletId = message.FromWalletId;
             }
 
             public static SceneMessage CopyFrom(in ReceivedMessage<Scene> message) =>

Explorer/Assets/Scripts/CrdtEcsBridge/JsModulesImplementation/Communications/SDKMessageBus/SDKMessageBusCommsAPIImplementation.cs

@@ -10,33 +10,31 @@ namespace CrdtEcsBridge.JsModulesImplementation.Communications.SDKMessageBus
 {
     public class SDKMessageBusCommsAPIImplementation : CommunicationsControllerAPIImplementationBase, ISDKMessageBusCommsControllerAPI
     {
-        public List<CommsPayload> SceneCommsMessages { get; } = new ();
+        private readonly List<CommsPayload> messages = new ();
 
-        public SDKMessageBusCommsAPIImplementation(ISceneData sceneData, ICommunicationControllerHub messagePipesHub, IJsOperations jsOperations, ISceneStateProvider sceneStateProvider) : base(sceneData, messagePipesHub, jsOperations, sceneStateProvider)
+        public IReadOnlyList<CommsPayload> SceneCommsMessages => messages;
+
+        public SDKMessageBusCommsAPIImplementation(ISceneData sceneData, ISceneCommunicationPipe sceneCommunicationPipe, IJsOperations jsOperations, ISceneStateProvider sceneStateProvider) : base(sceneData, sceneCommunicationPipe, jsOperations, sceneStateProvider) { }
+
+        public void ClearMessages()
         {
+            messages.Clear();
         }
 
         public void Send(string data)
         {
-            var dataBytes = Encoding.UTF8.GetBytes(data);
-            Span<byte> encodedMessage = stackalloc byte[dataBytes.Length + 1];
-            encodedMessage[0] = (byte)MsgType.String;
-            dataBytes.CopyTo(encodedMessage[1..]);
-
-            messagePipesHub.SendMessage(encodedMessage, sceneData.SceneEntityDefinition.id, cancellationTokenSource.Token);
+            byte[] dataBytes = Encoding.UTF8.GetBytes(data);
+            EncodeAndSendMessage(MsgType.String, dataBytes);
         }
 
-        protected override void OnMessageReceived(ICommunicationControllerHub.SceneMessage receivedMessage)
+        protected override void OnMessageReceived(MsgType messageType, ReadOnlySpan<byte> decodedMessage, string fromWalletId)
         {
-            ReadOnlySpan<byte> decodedMessage = receivedMessage.Data.Span;
-            MsgType msgType = DecodeMessage(ref decodedMessage);
-
-            if (msgType != MsgType.String || decodedMessage.Length == 0)
+            if (messageType != MsgType.String)
                 return;
 
-            SceneCommsMessages.Add(new CommsPayload
+            messages.Add(new CommsPayload
             {
-                sender = receivedMessage.WalletId,
+                sender = fromWalletId,
                 message = Encoding.UTF8.GetString(decodedMessage)
             });
         }

Explorer/Assets/Scripts/CrdtEcsBridge/JsModulesImplementation/Communications/CommunicationControllerHub.cs → Explorer/Assets/Scripts/CrdtEcsBridge/JsModulesImplementation/Communications/SceneCommunicationPipe.cs

@@ -13,12 +13,12 @@ namespace CrdtEcsBridge.JsModulesImplementation.Communications
     ///     We can't subscribe to the `Scene` message multiple times
     ///     so Hub handles the subscription and the API implementation handles the message processing
     /// </summary>
-    public class CommunicationControllerHub : ICommunicationControllerHub
+    public class SceneCommunicationPipe : ISceneCommunicationPipe
     {
-        private Action<ICommunicationControllerHub.SceneMessage>? onSceneMessage;
+        private Action<ISceneCommunicationPipe.SceneMessage>? onSceneMessage;
         private readonly IMessagePipe messagePipe;
 
-        public CommunicationControllerHub(IMessagePipesHub messagePipesHub)
+        public SceneCommunicationPipe(IMessagePipesHub messagePipesHub)
         {
             messagePipe = messagePipesHub.ScenePipe();
             messagePipe.Subscribe<Scene>(Packet.MessageOneofCase.Scene, InvokeCurrentHandler);
@@ -27,15 +27,15 @@ public CommunicationControllerHub(IMessagePipesHub messagePipesHub)
         private void InvokeCurrentHandler(ReceivedMessage<Scene> message)
         {
             using (message)
-                onSceneMessage?.Invoke(ICommunicationControllerHub.SceneMessage.CopyFrom(in message));
+                onSceneMessage?.Invoke(ISceneCommunicationPipe.SceneMessage.CopyFrom(in message));
         }
 
-        public void RemoveSceneMessageHandler(Action<ICommunicationControllerHub.SceneMessage> onSceneMessage)
+        public void RemoveSceneMessageHandler(Action<ISceneCommunicationPipe.SceneMessage> onSceneMessage)
         {
             lock (this) { this.onSceneMessage -= onSceneMessage; }
         }
 
-        public void SetSceneMessageHandler(Action<ICommunicationControllerHub.SceneMessage> onSceneMessage)
+        public void SetSceneMessageHandler(Action<ISceneCommunicationPipe.SceneMessage> onSceneMessage)
         {
             lock (this) { this.onSceneMessage += onSceneMessage; }
         }