Merge pull request #520 from starfi5h/pr-pause

Change send and receive behaviors

Author: sp00ktober

.github/scripts/thunderstore_bundle.js

@@ -124,6 +124,7 @@ function generateManifest() {
       `nebula-${apiPluginInfo.name}-${apiPluginInfo.version}`,
       "PhantomGamers-IlLine-1.0.0",
       "CommonAPI-CommonAPI-1.2.2",
+      "starfi5h-BulletTime-1.2.0",
     ],
     website_url: "https://github.com/hubastard/nebula"
   };

NebulaModel/Logger/Log.cs

@@ -5,6 +5,8 @@ namespace NebulaModel.Logger
 {
     public static class Log
     {
+        public static string MessageInfo { get; private set; }
+
         private static ILogger logger;
 
         public static void Init(ILogger logger)
@@ -27,6 +29,7 @@ public static void Debug(object message)
         public static void Info(string message)
         {
             logger.LogInfo(message);
+            MessageInfo = message;
         }
 
         public static void Info(object message)

NebulaModel/Networking/NebulaConnection.cs

@@ -2,6 +2,7 @@
 using NebulaModel.Logger;
 using NebulaModel.Networking.Serialization;
 using System;
+using System.Collections.Generic;
 using System.Net;
 using WebSocketSharp;
 
@@ -12,6 +13,8 @@ public class NebulaConnection : INebulaConnection
         private readonly EndPoint peerEndpoint;
         private readonly WebSocket peerSocket;
         private readonly NetPacketProcessor packetProcessor;
+        private readonly Queue<byte[]> pendingPackets = new Queue<byte[]>();
+        private bool enable = true;
 
         public bool IsAlive => peerSocket?.IsAlive ?? false;
 
@@ -24,25 +27,46 @@ public NebulaConnection(WebSocket peerSocket, EndPoint peerEndpoint, NetPacketPr
 
         public void SendPacket<T>(T packet) where T : class, new()
         {
-            if (peerSocket.ReadyState == WebSocketState.Open)
+            lock (pendingPackets)
             {
-                peerSocket.Send(packetProcessor.Write(packet));
-            }
-            else
-            {
-                Log.Warn($"Cannot send packet {packet?.GetType()} to a closed connection {peerEndpoint}");
-            }
+                byte[] rawData = packetProcessor.Write(packet);
+                pendingPackets.Enqueue(rawData);
+                ProcessPacketQueue();
+            }            
         }
 
         public void SendRawPacket(byte[] rawData)
         {
-            if (peerSocket.ReadyState == WebSocketState.Open)
+            lock (pendingPackets)
             {
-                peerSocket.Send(rawData);
+                pendingPackets.Enqueue(rawData);
+                ProcessPacketQueue();
+            }            
+        }
+
+        private void ProcessPacketQueue()
+        {
+            if (enable && pendingPackets.Count > 0)
+            {
+                byte[] packet = pendingPackets.Dequeue();
+                if (peerSocket.ReadyState == WebSocketState.Open)
+                {
+                    peerSocket.SendAsync(packet, OnSendCompleted);
+                    enable = false;
+                }
+                else
+                {
+                    Log.Warn($"Cannot send packet to a {peerSocket.ReadyState} connection {peerEndpoint.GetHashCode()}");
+                }
             }
-            else
+        }
+
+        private void OnSendCompleted(bool result)
+        {
+            lock (pendingPackets)
             {
-                Log.Warn($"Cannot send raw packet to a closed connection {peerSocket?.Url}");
+                enable = true;
+                ProcessPacketQueue();
             }
         }
 

NebulaModel/Networking/Serialization/NetPacketProcessor.cs

@@ -26,6 +26,7 @@ private static class HashCache<T>
         public bool SimulateLatency = false;
         public int SimulatedMinLatency = 20;
         public int SimulatedMaxLatency = 50;
+        public bool Enable { get; set; } = true;
 
         public NetPacketProcessor()
         {
@@ -70,7 +71,7 @@ public void ProcessPacketQueue()
             {
                 ProcessDelayedPackets();
 
-                while (pendingPackets.Count > 0)
+                while (pendingPackets.Count > 0 && Enable)
                 {
                     PendingPacket packet = pendingPackets.Dequeue();
                     ReadPacket(new NetDataReader(packet.Data), packet.UserData);

NebulaModel/Packets/Session/HandshakeRequest.cs

@@ -33,6 +33,17 @@ public HandshakeRequest(byte[] clientCert, string username, Float4[] mechaColors
                         writer.BinaryWriter.Write(mod.Version);
                         count++;
                     }
+                    else
+                    {
+                        foreach (BepInEx.BepInDependency dependency in pluginInfo.Value.Dependencies)
+                        {
+                            if (dependency.DependencyGUID == NebulaModAPI.API_GUID)
+                            {
+                                writer.BinaryWriter.Write(pluginInfo.Key);
+                                writer.BinaryWriter.Write(pluginInfo.Value.Metadata.Version.ToString());
+                            }
+                        }
+                    }
                 }
 
                 ModsVersion = writer.CloseAndGetBytes();

NebulaModel/Packets/Session/LobbyRequest.cs

@@ -28,6 +28,18 @@ public LobbyRequest(byte[] clientCert, string username, Float4[] mechaColors)
                         writer.BinaryWriter.Write(mod.Version);
                         count++;
                     }
+                    else
+                    {
+                        foreach (BepInEx.BepInDependency dependency in pluginInfo.Value.Dependencies)
+                        {
+                            if (dependency.DependencyGUID == NebulaModAPI.API_GUID)
+                            {
+                                writer.BinaryWriter.Write(pluginInfo.Key);
+                                writer.BinaryWriter.Write(pluginInfo.Value.Metadata.Version.ToString());
+                                count++;
+                            }
+                        }
+                    }
                 }
 
                 ModsVersion = writer.CloseAndGetBytes();

NebulaNetwork/Client.cs

@@ -140,7 +140,10 @@ public override void Update()
                 gameStateUpdateTimer += Time.deltaTime;
                 if (gameStateUpdateTimer >= GAME_STATE_UPDATE_INTERVAL)
                 {
-                    SendPacket(new PingPacket());
+                    if (!GameMain.isFullscreenPaused)
+                    {
+                        SendPacket(new PingPacket());
+                    }
                     gameStateUpdateTimer = 0f;
                 }
             }

NebulaNetwork/PacketProcessors/GameStates/GameStateUpdateProcessor.cs

@@ -1,11 +1,11 @@
 using NebulaAPI;
 using NebulaModel;
-using NebulaModel.DataStructures;
 using NebulaModel.Logger;
 using NebulaModel.Networking;
 using NebulaModel.Packets;
 using NebulaModel.Packets.GameStates;
 using NebulaWorld;
+using NebulaWorld.GameStates;
 using System;
 using UnityEngine;
 
@@ -14,9 +14,13 @@ namespace NebulaNetwork.PacketProcessors.GameStates
     [RegisterPacketProcessor]
     public class GameStateUpdateProcessor : PacketProcessor<GameStateUpdate>
     {
+        public float MaxUPS = 120f;
+        public float MinUPS = 30f;
+        public float BUFFERING_TIME = 30f;
+        public float BUFFERING_TICK = 60f;
+
         private int averageRTT;
         private float avaerageUPS = 60f;
-        private const float BUFFERING_TIME = 20f;
         private bool hasChanged;
 
         public override void ProcessPacket(GameStateUpdate packet, NebulaConnection conn)
@@ -64,23 +68,23 @@ public override void ProcessPacket(GameStateUpdate packet, NebulaConnection conn
             // Adjust client's UPS to match game tick with server, range 30~120 UPS
             float UPS = diff / 1f + avaerageUPS;
             long skipTick = 0;
-            if (UPS > 120f)
+            if (UPS > MaxUPS)
             {
                 // Try to disturbute game tick difference into BUFFERING_TIME (seconds)
-                if (diff / BUFFERING_TIME + avaerageUPS > 120f)
+                if (diff / BUFFERING_TIME + avaerageUPS > MaxUPS)
                 {
                     // The difference is too large, need to skip ticks to catch up
-                    skipTick = (long)(UPS - 120f);
+                    skipTick = (long)(UPS - MaxUPS);
                 }
-                UPS = 120f;
+                UPS = MaxUPS;
             }
-            else if (UPS < 30f)
+            else if (UPS < MinUPS)
             {
-                if (diff / BUFFERING_TIME + avaerageUPS < 30f)
+                if (diff + avaerageUPS - MinUPS < -BUFFERING_TICK)
                 {
-                    skipTick = (long)(UPS - 30f);
+                    skipTick = (long)(UPS - MinUPS);
                 }
-                UPS = 30f;
+                UPS = MinUPS;
             }
             if (skipTick != 0)
             {
@@ -89,6 +93,8 @@ public override void ProcessPacket(GameStateUpdate packet, NebulaConnection conn
             }
             FPSController.SetFixUPS(UPS);
             hasChanged = true;
+            // Tick difference in the next second. Expose for other mods
+            GameStatesManager.NotifyTickDifference((diff / 1f + avaerageUPS) - UPS);
         }
     }
 }

NebulaNetwork/PacketProcessors/Planet/FactoryDataProcessor.cs

@@ -1,4 +1,5 @@
 using NebulaAPI;
+using NebulaModel.Logger;
 using NebulaModel.Networking;
 using NebulaModel.Packets;
 using NebulaModel.Packets.Planet;
@@ -15,8 +16,13 @@ public override void ProcessPacket(FactoryData packet, NebulaConnection conn)
             {
                 return;
             }
+            // Stop packet processing until factory is imported
+            ((NebulaModel.NetworkProvider)Multiplayer.Session.Network).PacketProcessor.Enable = false;
+            Log.Info($"FactoryDataProcessor: Pause PacketProcessor");
 
+            PlanetData planet = GameMain.galaxy.PlanetById(packet.PlanetId);
             Multiplayer.Session.Planets.PendingFactories.Add(packet.PlanetId, packet.BinaryData);
+            Log.Info($"Parsing {packet.BinaryData.Length} bytes of data for factory {planet.name} (ID: {planet.id})");
 
             lock (PlanetModelingManager.fctPlanetReqList)
             {

NebulaNetwork/PacketProcessors/Planet/FactoryLoadRequestProcessor.cs

@@ -1,7 +1,9 @@
 using NebulaAPI;
+using NebulaModel.Logger;
 using NebulaModel.Networking;
 using NebulaModel.Packets;
 using NebulaModel.Packets.Planet;
+using NebulaWorld;
 
 namespace NebulaNetwork.PacketProcessors.Planet
 {
@@ -21,7 +23,22 @@ public override void ProcessPacket(FactoryLoadRequest packet, NebulaConnection c
             using (BinaryUtils.Writer writer = new BinaryUtils.Writer())
             {
                 factory.Export(writer.BinaryWriter);
-                conn.SendPacket(new FactoryData(packet.PlanetID, writer.CloseAndGetBytes()));
+                byte[] data = writer.CloseAndGetBytes();
+                Log.Info($"Sent {data.Length} bytes of data for PlanetFactory {planet.name} (ID: {planet.id})");
+                conn.SendPacket(new FactoryData(packet.PlanetID, data));
+            }
+
+            // Add requesting client to connected player, so he can receive following update
+            IPlayerManager playerManager = Multiplayer.Session.Network.PlayerManager;
+            INebulaPlayer player = playerManager.GetSyncingPlayer(conn);
+            if (player != null)
+            {
+                player.Data.LocalPlanetId = packet.PlanetID;
+                player.Data.LocalStarId = GameMain.galaxy.PlanetById(packet.PlanetID).star.id;
+                using (playerManager.GetConnectedPlayers(out System.Collections.Generic.Dictionary<INebulaConnection, INebulaPlayer> connectedPlayers))
+                {
+                    connectedPlayers.Add(player.Connection, player);
+                }
             }
         }
     }