Added support for LatentEvents

manups4e · Apr 5, 2024 · e1fe208 · e1fe208
1 parent 90a2e64
commit e1fe208
Show file tree

Hide file tree

Showing 5 changed files with 135 additions and 3 deletions.
diff --git a/src/FxEvents.Client/EventDispatcher.cs b/src/FxEvents.Client/EventDispatcher.cs
@@ -101,6 +101,17 @@ public static void Send(string endpoint, params object[] args)
             }
             Events.Send(endpoint, args);
         }
+
+        public static void SendLatent(string endpoint, int bytesPerSeconds, params object[] args)
+        {
+            if (!Initialized)
+            {
+                Logger.Error("Dispatcher not initialized, please initialize it and add the events strings");
+                return;
+            }
+            Events.SendLatent(endpoint, bytesPerSeconds, args);
+        }
+
         public static async Task<T> Get<T>(string endpoint, params object[] args)
         {
             if (!Initialized)

diff --git a/src/FxEvents.Client/EventSystem/ClientGateway.cs b/src/FxEvents.Client/EventSystem/ClientGateway.cs
@@ -81,6 +81,11 @@ public async void Send(string endpoint, params object[] args)
             await SendInternal(EventFlowType.Straight, new ServerId().Handle, endpoint, args);
         }
 
+        public async void SendLatent(string endpoint, int bytePerSecond, params object[] args)
+        {
+            await SendInternalLatent(EventFlowType.Straight, new ServerId().Handle, endpoint, bytePerSecond, args);
+        }
+
         public async Task<T> Get<T>(string endpoint, params object[] args)
         {
             return await GetInternal<T>(new ServerId().Handle, endpoint, args);

diff --git a/src/FxEvents.Server/EventDispatcher.cs b/src/FxEvents.Server/EventDispatcher.cs
@@ -133,7 +133,7 @@ public static void Send(IEnumerable<Player> players, string endpoint, params obj
             }
             Events.Send(players.Select(x => Convert.ToInt32(x.Handle)).ToList(), endpoint, args);
         }
-        
+
         public static void Send(string endpoint, params object[] args)
         {
             if (!Initialized)
@@ -142,10 +142,10 @@ public static void Send(string endpoint, params object[] args)
                 return;
             }
 
-            var playerList = Instance.GetPlayers;
+            PlayerList playerList = Instance.GetPlayers;
             Events.Send(playerList.Select(x => Convert.ToInt32(x.Handle)).ToList(), endpoint, args);
         }
-        
+
         public static void Send(IEnumerable<ISource> clients, string endpoint, params object[] args)
         {
             if (!Initialized)
@@ -155,6 +155,58 @@ public static void Send(IEnumerable<ISource> clients, string endpoint, params ob
             }
             Events.Send(clients.Select(x => x.Handle).ToList(), endpoint, args);
         }
+
+        public static void SendLatent(Player player, string endpoint, int bytesPerSeconds, params object[] args)
+        {
+            if (!Initialized)
+            {
+                Logger.Error("Dispatcher not initialized, please initialize it and add the events strings");
+                return;
+            }
+            Events.SendLatent(Convert.ToInt32(player.Handle), endpoint, bytesPerSeconds, args);
+        }
+
+        public static void SendLatent(ISource client, string endpoint, int bytesPerSeconds, params object[] args)
+        {
+            if (!Initialized)
+            {
+                Logger.Error("Dispatcher not initialized, please initialize it and add the events strings");
+                return;
+            }
+            Events.SendLatent(client.Handle, endpoint, bytesPerSeconds, args);
+        }
+
+        public static void SendLatent(IEnumerable<Player> players, string endpoint, int bytesPerSeconds, params object[] args)
+        {
+            if (!Initialized)
+            {
+                Logger.Error("Dispatcher not initialized, please initialize it and add the events strings");
+                return;
+            }
+            Events.SendLatent(players.Select(x => Convert.ToInt32(x.Handle)).ToList(), endpoint, bytesPerSeconds, args);
+        }
+
+        public static void SendLatent(string endpoint, int bytesPerSeconds, params object[] args)
+        {
+            if (!Initialized)
+            {
+                Logger.Error("Dispatcher not initialized, please initialize it and add the events strings");
+                return;
+            }
+            PlayerList playerList = Instance.GetPlayers;
+            Events.SendLatent(playerList.Select(x => Convert.ToInt32(x.Handle)).ToList(), endpoint, bytesPerSeconds, args);
+        }
+
+        public static void SendLatent(IEnumerable<ISource> clients, string endpoint, int bytesPerSeconds, params object[] args)
+        {
+            if (!Initialized)
+            {
+                Logger.Error("Dispatcher not initialized, please initialize it and add the events strings");
+                return;
+            }
+            Events.SendLatent(clients.Select(x => x.Handle).ToList(), endpoint, bytesPerSeconds, args);
+        }
+
         public static Task<T> Get<T>(Player player, string endpoint, params object[] args)
         {
             if (!Initialized)

diff --git a/src/FxEvents.Server/EventSystem/ServerGateway.cs b/src/FxEvents.Server/EventSystem/ServerGateway.cs
@@ -154,6 +154,27 @@ public async void Send(int target, string endpoint, params object[] args)
             await SendInternal(EventFlowType.Straight, target, endpoint, args);
         }
 
+        public void SendLatent(Player player, string endpoint, int bytesxSecond, params object[] args) => SendLatent(Convert.ToInt32(player.Handle), endpoint, bytesxSecond, args);
+        public void SendLatent(ISource client, string endpoint, int bytesxSecond, params object[] args) => SendLatent(client.Handle, endpoint, bytesxSecond, args);
+        public void SendLatent(List<Player> players, string endpoint, int bytesxSecond, params object[] args) => SendLatent(players.Select(x => Convert.ToInt32(x.Handle)).ToList(), endpoint, bytesxSecond, args);
+        public void SendLatent(List<ISource> clients, string endpoint, int bytesxSecond, params object[] args) => SendLatent(clients.Select(x => x.Handle).ToList(), endpoint, bytesxSecond, args);
+
+        public async void SendLatent(List<int> targets, string endpoint, int bytesxSecond, params object[] args)
+        {
+            int i = 0;
+            while (i < targets.Count)
+            {
+                await BaseScript.Delay(0);
+                SendLatent(targets[i], endpoint, bytesxSecond, args);
+                i++;
+            }
+        }
+
+        public async void SendLatent(int target, string endpoint, int bytesxSecond, params object[] args)
+        {
+            await SendInternalLatent(EventFlowType.Straight, target, endpoint, bytesxSecond, args);
+        }
+
         public Task<T> Get<T>(Player player, string endpoint, params object[] args) =>
             Get<T>(Convert.ToInt32(player.Handle), endpoint, args);
 

diff --git a/src/FxEvents.Shared/EventSubsystem/BaseGateway.cs b/src/FxEvents.Shared/EventSubsystem/BaseGateway.cs
@@ -19,6 +19,7 @@ namespace FxEvents.Shared.EventSubsystem
     public delegate Task EventDelayMethod(int ms = 0);
     public delegate Task EventMessagePreparation(string pipeline, int source, IMessage message);
     public delegate void EventMessagePush(string pipeline, int source, byte[] buffer);
+    public delegate void EventMessagePushLatent(string pipeline, int source, int bytePerSecond, byte[] buffer);
     public delegate ISource ConstructorCustomActivator<T>(int handle);
     public abstract class BaseGateway
     {
@@ -35,6 +36,7 @@ public abstract class BaseGateway
         public EventDelayMethod? DelayDelegate { get; set; }
         public EventMessagePreparation? PrepareDelegate { get; set; }
         public EventMessagePush? PushDelegate { get; set; }
+        public EventMessagePushLatent? PushDelegateLatent { get; set; }
 
         public async Task ProcessInboundAsync(int source, byte[] serialized)
         {
@@ -260,6 +262,47 @@ protected async Task<EventMessage> SendInternal(EventFlowType flow, int source,
             return message;
         }
 
+        protected async Task<EventMessage> SendInternalLatent(EventFlowType flow, int source, string endpoint, int bytePerSecond, params object[] args)
+        {
+            StopwatchUtil stopwatch = StopwatchUtil.StartNew();
+            List<EventParameter> parameters = [];
+
+            for (int idx = 0; idx < args.Length; idx++)
+            {
+                object argument = args[idx];
+                Type type = argument.GetType();
+                //Debug.WriteLine($"outbound latent {endpoint} - index: {idx}, type:{type.FullName}, value:{argument.ToJson()}");
+
+                using SerializationContext context = new(endpoint, $"(Send) Parameter Index '{idx}'", Serialization);
+
+                context.Serialize(type, argument);
+                parameters.Add(new EventParameter(context.GetData()));
+            }
+
+            EventMessage message = new(endpoint, flow, parameters);
+
+            if (PrepareDelegate != null)
+            {
+                stopwatch.Stop();
+
+                await PrepareDelegate(InboundPipeline, source, message);
+                stopwatch.Start();
+            }
+
+            byte[] data = message.EncryptObject(EventDispatcher.EncryptionKey);
+
+            PushDelegateLatent(InboundPipeline, source, bytePerSecond, data);
+            if (EventDispatcher.Debug)
+            {
+#if CLIENT
+                Logger.Debug($"[{endpoint} {flow}] Sent latent {data.Length} byte(s) to {(source == -1 ? "Server" : API.GetPlayerName(source))} in {stopwatch.Elapsed.TotalMilliseconds}ms");
+#elif SERVER
+                Logger.Debug($"[{endpoint} {flow}] Sent latent {data.Length} byte(s) to {(source == -1 ? "Server" : API.GetPlayerName("" + source))} in {stopwatch.Elapsed.TotalMilliseconds}ms");
+#endif
+            }
+            return message;
+        }
+
         protected async Task<T> GetInternal<T>(int source, string endpoint, params object[] args)
         {
             StopwatchUtil stopwatch = StopwatchUtil.StartNew();