Merge upstream/main and resolve conflicts

Author: xingxing2008

.github/workflows/ci.yml

@@ -0,0 +1,36 @@
+name: Publish ConnectX.Client NuGet Package
+
+on:
+  push:
+    branches:
+      - main
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/setup-dotnet@v4
+        with:
+          dotnet-version: 9.0
+
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          submodules: recursive
+
+      - name: Get current datetime
+        id: currentdatetime
+        uses: Kaven-Universe/github-action-current-date-time@v1
+        with:
+          format: "YYYY.MMDD.HHmmssSSS"
+
+      - name: Restore packages
+        run: dotnet restore ./ConnectX.Client/ConnectX.Client.csproj
+
+      - name: Build solution
+        run: dotnet build ./ConnectX.Client/ConnectX.Client.csproj -c Release --no-restore
+
+      - name: Pack solution
+        run: dotnet pack ./ConnectX.Client/ConnectX.Client.csproj -c Release -p:PackageVersion="${{ steps.currentdatetime.outputs.time }}" --no-build --no-restore -o .
+
+      - name: Publish to NuGet
+        run: dotnet nuget push *.nupkg -k ${{secrets.NUGET_KEY}} -s https://api.nuget.org/v3/index.json

ConnectX.Client/ConnectX.Client.csproj

@@ -10,6 +10,7 @@
     <PackageReference Include="CommunityToolkit.HighPerformance" Version="8.4.0" />
     <PackageReference Condition="'$(Configuration)' == 'Debug'" Include="ConsoleTables" Version="2.6.2" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.4" />
+    <PackageReference Include="Snappier" Version="1.2.0" />
     <PackageReference Include="ZeroTier.Sockets" Version="1.8.4" />
   </ItemGroup>
 
@@ -25,4 +26,13 @@
     <ProjectReference Include="..\Hive.Framework\Hive.Network.Udp\Hive.Network.Udp.csproj" />
   </ItemGroup>
 
+  <PropertyGroup>
+    <IncludeSymbols>true</IncludeSymbols>
+    <SymbolPackageFormat>snupkg</SymbolPackageFormat>
+    <PackageLicenseExpression>MIT</PackageLicenseExpression>
+    <PackageProjectUrl>https://github.com/Corona-Studio/ConnectX</PackageProjectUrl>
+    <Title>$(AssemblyName)</Title>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
 </Project>

ConnectX.Client/Interfaces/IZeroTierNodeLinkHolder.cs

@@ -9,7 +9,9 @@ public interface IZeroTierNodeLinkHolder : IHostedService
     public const ushort RandomPortLower = 30000;
     public const ushort RandomPortUpper = 40000;
 
+    bool IsZeroTierInitialized { get; }
     Node? Node { get; }
+
     Task<bool> JoinNetworkAsync(ulong networkId, CancellationToken cancellationToken);
     Task LeaveNetworkAsync(CancellationToken cancellationToken);
 

ConnectX.Client/Managers/GenericProxyManager.cs

@@ -1,7 +1,11 @@
-using System.Net;
+using System.Collections.Concurrent;
+using System.Net;
 using System.Net.Sockets;
 using ConnectX.Client.Messages.Proxy;
+using ConnectX.Client.Models;
 using ConnectX.Client.Proxy;
+using ConnectX.Client.Route;
+using ConnectX.Client.Transmission;
 using ConnectX.Shared.Helpers;
 using ConnectX.Shared.Interfaces;
 using Hive.Both.General.Dispatchers;
@@ -14,15 +18,17 @@ namespace ConnectX.Client.Managers;
 public abstract class GenericProxyManager : BackgroundService
 {
     //使用一个二元组确定一个连接
-    private readonly Dictionary<TunnelIdentifier, GenericProxyPair> _proxies = [];
-    private readonly Dictionary<TunnelIdentifier, Socket> _acceptedSockets = [];
-    private readonly Dictionary<ValueTuple<Guid, ushort>, GenericProxyAcceptor> _acceptors = [];
+    private readonly ConcurrentDictionary<TunnelIdentifier, GenericProxyPair> _proxies = [];
+    private readonly ConcurrentDictionary<TunnelIdentifier, Socket> _acceptedSockets = [];
+    private readonly ConcurrentDictionary<ValueTuple<Guid, ushort>, GenericProxyAcceptor> _acceptors = [];
 
     private readonly IHostApplicationLifetime _lifetime;
     private readonly IServiceProvider _serviceProvider;
     protected readonly ILogger Logger;
 
     protected GenericProxyManager(
+        RouterPacketDispatcher packetDispatcher,
+        RelayPacketDispatcher relayPacketDispatcher,
         IHostApplicationLifetime lifetime,
         IServiceProvider serviceProvider,
         ILogger logger)
@@ -31,6 +37,9 @@ protected GenericProxyManager(
         _serviceProvider = serviceProvider;
 
         Logger = logger;
+
+        packetDispatcher.OnReceive<ProxyDisconnectReq>(RecycleClientProxy);
+        relayPacketDispatcher.OnReceive<ProxyDisconnectReq>(RecycleClientProxy);
     }
 
     public override void Dispose()
@@ -45,6 +54,20 @@ public override void Dispose()
         GC.SuppressFinalize(this);
     }
 
+    private void RecycleClientProxy(ProxyDisconnectReq req, PacketContext ctx)
+    {
+        var key = new TunnelIdentifier(req.ClientId, req.ServerRealPort, req.ClientRealPort);
+
+        if (_proxies.TryRemove(key, out var pair))
+        {
+            Logger.LogClientProxyDisconnectedBecauseRequested(key);
+            pair.Dispose();
+            return;
+        }
+
+        Logger.LogFailedToShutdownClientProxyBecauseNotFound(key);
+    }
+
     public void ReceivedProxyConnectReq(
         MessageContext<ProxyConnectReq> ctx,
         ISender sender)
@@ -133,7 +156,11 @@ public virtual GenericProxyPair CreateClientProxy(
             dispatcher,
             sender);
 
-        _proxies.Add(key, pair);
+        _proxies.AddOrUpdate(key, _ => pair, (_, old) =>
+        {
+            old.Dispose();
+            return pair;
+        });
 
         Logger.LogCreateProxy(key);
 
@@ -164,7 +191,9 @@ public virtual GenericProxyPair CreateClientProxy(
             socket,
             key,
             _lifetime.ApplicationStopping);
+
         proxy.OnRealServerDisconnected += OnProxyDisconnected;
+        proxy.OnRealServerDisconnected += NotifyServerProxyDisconnect;
 
         var pair = new GenericProxyPair(
             partnerId,
@@ -174,10 +203,29 @@ public virtual GenericProxyPair CreateClientProxy(
             dispatcher,
             sender);
 
-        _proxies.Add(key, pair);
+        _proxies.AddOrUpdate(key, _ => pair, (_, old) =>
+        {
+            old.Dispose();
+            return pair;
+        });
         proxy.Start();
 
         return proxy;
+
+        void NotifyServerProxyDisconnect(TunnelIdentifier id, GenericProxyBase proxy)
+        {
+            proxy.OnRealServerDisconnected -= NotifyServerProxyDisconnect;
+
+            // 真客户端离开了房间或掉线，通知房主回收对应的 proxy
+            sender.SendData(new ProxyDisconnectReq
+            {
+                ClientId = id.PartnerId,
+                ClientRealPort = id.LocalRealPort,
+                ServerRealPort = id.RemoteRealPort
+            });
+
+            Logger.LogNotifyServerProxyNeedToDisconnect(id);
+        }
     }
 
     private void OnProxyDisconnected(TunnelIdentifier id, GenericProxyBase proxy)
@@ -206,11 +254,11 @@ public virtual GenericProxyAcceptor CreateAcceptor(
 
         var key = (partnerId, remoteRealServerPort);
 
-        if (_acceptors.Remove(key, out var oldAcceptor))
+        if (_acceptors.TryRemove(key, out var oldAcceptor))
         {
             if (oldAcceptor.IsRunning)
             {
-                _acceptors.Add(key, oldAcceptor);
+                _acceptors.AddOrUpdate((partnerId, remoteRealServerPort), _ => oldAcceptor, (_, _) => oldAcceptor);
                 throw new InvalidOperationException($"There has been a acceptor with same key: {partnerId}-{remoteRealServerPort}");
             }
 
@@ -224,7 +272,7 @@ public virtual GenericProxyAcceptor CreateAcceptor(
             localMapPort,
             _lifetime.ApplicationStopping);
 
-        _acceptors.Add((partnerId, remoteRealServerPort), acceptor);
+        _acceptors.AddOrUpdate((partnerId, remoteRealServerPort), _ => acceptor, (_, _) => acceptor);
 
         // 当有真客户端连接的时候，发送一个McConnectReq，当收到回复后启动Proxy（启动的逻辑在OnReceive里）
         acceptor.OnRealClientConnected += (_, socket) =>
@@ -240,7 +288,7 @@ public virtual GenericProxyAcceptor CreateAcceptor(
                 (ushort)remoteEndPoint.Port,
                 remoteRealServerPort);
 
-            _acceptedSockets.Add(id, socket);
+            _acceptedSockets.AddOrUpdate(id, _ => socket, (_, _) => socket);
 
             sender.SendData(new ProxyConnectReq
             {
@@ -351,4 +399,13 @@ public static partial void LogErrorProxyPairWithSameKey(this ILogger logger, Gui
 
     [LoggerMessage(LogLevel.Information, "[GEN_PROXY_MANAGER] Create acceptor {Key}")]
     public static partial void LogCreateAcceptor(this ILogger logger, (Guid, ushort) key);
+
+    [LoggerMessage(LogLevel.Information, "[GEN_PROXY_MANAGER] Notify server proxy need to disconnect {Id}")]
+    public static partial void LogNotifyServerProxyNeedToDisconnect(this ILogger logger, TunnelIdentifier id);
+
+    [LoggerMessage(LogLevel.Error, "[GEN_PROXY_MANAGER] Failed to shutdown client proxy because not found {key}")]
+    public static partial void LogFailedToShutdownClientProxyBecauseNotFound(this ILogger logger, TunnelIdentifier key);
+
+    [LoggerMessage(LogLevel.Information, "[GEN_PROXY_MANAGER] Client proxy disconnected because requested {key}")]
+    public static partial void LogClientProxyDisconnectedBecauseRequested(this ILogger logger, TunnelIdentifier key);
 }

ConnectX.Client/Managers/ProxyManager.cs

@@ -1,7 +1,7 @@
 using System.Collections.Concurrent;
-using ConnectX.Client.Interfaces;
 using ConnectX.Client.Messages.Proxy;
 using ConnectX.Client.Proxy;
+using ConnectX.Client.Route;
 using ConnectX.Client.Transmission;
 using ConnectX.Shared.Helpers;
 using Hive.Both.General.Dispatchers;
@@ -16,11 +16,13 @@ public sealed class ProxyManager : GenericProxyManager
     private readonly ConcurrentBag<(IDispatcher, HandlerId)> _registeredHandlers = [];
 
     public ProxyManager(
+        RouterPacketDispatcher packetDispatcher,
+        RelayPacketDispatcher relayPacketDispatcher,
         PartnerManager partnerManager,
         IHostApplicationLifetime lifetime,
         IServiceProvider serviceProvider,
         ILogger<ProxyManager> logger)
-        : base(lifetime, serviceProvider, logger)
+        : base(packetDispatcher, relayPacketDispatcher, lifetime, serviceProvider, logger)
     {
         _partnerManager = partnerManager;
     }

ConnectX.Client/Managers/RoomInfoManager.cs

@@ -46,7 +46,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         {
             if (CurrentGroupInfo == null)
             {
-                // Reset flags to original state
+                // Reset flags to the original state
                 needToRefreshRoomInfo = false;
                 lastRefreshTime = DateTime.MinValue;
 

ConnectX.Client/Messages/Proxy/ProxyDisconnectReq.cs

@@ -0,0 +1,13 @@
+using Hive.Codec.Shared;
+using MemoryPack;
+
+namespace ConnectX.Client.Messages.Proxy;
+
+[MessageDefine]
+[MemoryPackable]
+public partial class ProxyDisconnectReq
+{
+    public required Guid ClientId { get; init; }
+    public required ushort ClientRealPort { get; init; }
+    public required ushort ServerRealPort { get; init; }
+}

ConnectX.Client/Proxy/FakeServerMultiCaster.cs

@@ -1,4 +1,5 @@
-using System.Net;
+using System.Collections.Frozen;
+using System.Net;
 using System.Net.Sockets;
 using System.Text;
 using ConnectX.Client.Interfaces;
@@ -47,12 +48,24 @@ public FakeServerMultiCaster(
     }
 
     public event Action<string, int>? OnListenedLanServer;
+
+    private static readonly FrozenSet<string> VirtualKeywords = FrozenSet.Create(
+        "virtual", "vmware", "loopback",
+        "pseudo", "tunneling", "tap",
+        "container", "hyper-v", "bluetooth",
+        "docker");
 
     private static IPAddress GetLocalIpAddress()
     {
+        var candidates = new List<IPAddress>();
+
         var networkInterfaces = System.Net.NetworkInformation.NetworkInterface.GetAllNetworkInterfaces()
-            .Where(ni => ni.OperationalStatus == System.Net.NetworkInformation.OperationalStatus.Up 
-                         && ni.NetworkInterfaceType != System.Net.NetworkInformation.NetworkInterfaceType.Loopback);
+            .Where(ni =>
+                ni.OperationalStatus == System.Net.NetworkInformation.OperationalStatus.Up &&
+                ni.NetworkInterfaceType != System.Net.NetworkInformation.NetworkInterfaceType.Loopback &&
+                !VirtualKeywords.Contains(ni.Description.ToLowerInvariant()) &&
+                !VirtualKeywords.Contains(ni.Name.ToLowerInvariant())
+            );
 
         foreach (var ni in networkInterfaces)
         {
@@ -61,12 +74,19 @@ private static IPAddress GetLocalIpAddress()
             {
                 if (addr.Address.AddressFamily == AddressFamily.InterNetwork)
                 {
-                    return addr.Address;
+                    var ip = addr.Address;
+                    if (ip.ToString().StartsWith("192.168."))
+                        return ip;
+
+                    candidates.Add(ip);
                 }
             }
         }
 
-        throw new Exception("No network adapters with an IPv4 address in the system!");
+        if (candidates.Count > 0)
+            return candidates[0];
+
+        throw new Exception("No suitable IPv4 address found.");
     }
 
     private void OnReceiveMcMulticastMessage(McMulticastMessage message, PacketContext context)
@@ -88,6 +108,8 @@ private void OnReceiveMcMulticastMessage(McMulticastMessage message, PacketConte
 
             var multicastOption = new MulticastOption(MulticastAddress, IPAddress.Any);
             socket.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.AddMembership, multicastOption);
+            
+            _logger.LogSocketSetupForWindows(MulticastAddress.ToString());
         }
 
         if (OperatingSystem.IsMacOS() || OperatingSystem.IsLinux())
@@ -99,16 +121,21 @@ private void OnReceiveMcMulticastMessage(McMulticastMessage message, PacketConte
             socket.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.MulticastInterface, localIp.GetAddressBytes());
             socket.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.MulticastTimeToLive, 255);
             socket.Bind(new IPEndPoint(localIp, 0));
+            
+            _logger.LogSocketSetupForLinuxMacOS(localIp.ToString());
         }
 
         ArgumentNullException.ThrowIfNull(socket);
 
         using var multicastSocket = socket;
 
         var mess = $"[MOTD]{message.Name}[/MOTD][AD]{proxy.LocalMappingPort}[/AD]";
-        var buf = Encoding.Default.GetBytes(mess);
-
-        multicastSocket.SendTo(buf, MulticastIpe);
+        var buf = Encoding.Default.GetBytes(mess).AsSpan();
+        var sentLen = 0;
+        
+        while (sentLen < buf.Length)
+            sentLen += multicastSocket.SendTo(buf[sentLen..], MulticastIpe);
+        
         _logger.LogMulticastMessageToClient(proxy.LocalMappingPort, message.Name);
     }
 
@@ -235,4 +262,10 @@ public static partial void LogSendLanServerToPartner(this ILogger logger, string
 
     [LoggerMessage(LogLevel.Error, "Failed to receive multicast message.")]
     public static partial void LogFailedToReceiveMulticastMessage(this ILogger logger, Exception exception);
+    
+    [LoggerMessage(LogLevel.Debug, "Socket setup for Windows, multicast address is {MulticastAddress}")]
+    public static partial void LogSocketSetupForWindows(this ILogger logger, string multicastAddress);
+    
+    [LoggerMessage(LogLevel.Debug, "Socket setup for Linux/MacOS, local IP is {LocalIp}")]
+    public static partial void LogSocketSetupForLinuxMacOS(this ILogger logger, string localIp);
 }

ConnectX.Client/Proxy/GenericProxyAcceptor.cs

@@ -62,7 +62,7 @@ public Task StartAcceptAsync()
                     var tmp = await _acceptSocket.AcceptAsync(_cancellationToken);
 
                     tmp.NoDelay = true;
-                    tmp.LingerState = new LingerOption(true, 3);
+                    //tmp.LingerState = new LingerOption(true, 3);
 
                     if (tmp.RemoteEndPoint is not IPEndPoint remoteEndPoint) continue;