✨ Connector server.

Signed-off-by: 秋雨落 <i@rain.cx>

CrystalConnector/Configs/ConfigAccessor.cs

@@ -13,7 +13,13 @@ public class ConfigAccessor
 
     public bool IsDebug()
     {
-        return Config.GetValue<bool>("Connector:Debug");
+        return HasDevEnvironmentVariable() || Config.GetValue<bool>("Connector:Debug");
+    }
+
+    private bool HasDevEnvironmentVariable()
+    {
+        var dev = Environment.GetEnvironmentVariable("DOTNET_ENVIRONMENT");
+        return dev != null && dev.Equals("Development", StringComparison.OrdinalIgnoreCase);
     }
 
     public string GetSecretKey()

CrystalConnector/Connector/MessageChannel.cs

@@ -2,5 +2,7 @@
 
 public class MessageChannel
 {
+    public required string Id { get; set; }
     
+    public required MessageDirection Direction { get; set; }
 }

CrystalConnector/Connector/MessageDirection.cs

@@ -0,0 +1,10 @@
+namespace CrystalConnector.Connector;
+
+[Flags]
+public enum MessageDirection
+{
+    None = 0,
+    Incoming = 1,
+    Outgoing = 2,
+    All = Incoming | Outgoing
+}

CrystalConnector/Connector/Packet/S2C/PacketS2C.cs

@@ -0,0 +1,15 @@
+using System.Formats.Cbor;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public abstract class PacketS2C : IPacket
+{
+    public void Write(CborWriter writer)
+    {
+        writer.WriteStartMap(null);
+        WriteData(writer);
+        writer.WriteEndMap();
+    }
+
+    protected abstract void WriteData(CborWriter writer);
+}

CrystalConnector/Connector/Packet/S2C/S2CAuthenticatedPacket.cs

@@ -0,0 +1,12 @@
+using System.Formats.Cbor;
+using CrystalConnector.Handlers;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public class S2CAuthenticatedPacket : PacketS2C
+{
+    protected override void WriteData(CborWriter writer)
+    {
+        writer.WriteTextString(HandlerConstants.ResponseAuthenticated);
+    }
+}

CrystalConnector/Connector/Packet/S2C/S2CForwardMessagePacket.cs

@@ -0,0 +1,24 @@
+using System.Formats.Cbor;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public class S2CForwardMessagePacket : PacketS2C
+{
+    private string Origin { get; }
+    private string ChannelId { get; }
+    private byte[] Payload { get; }
+    
+    public S2CForwardMessagePacket(string origin, string channelId, byte[] payload)
+    {
+        Origin = origin;
+        ChannelId = channelId;
+        Payload = payload;
+    }
+    
+    protected override void WriteData(CborWriter writer)
+    {
+        writer.WriteTextString(Origin);
+        writer.WriteTextString(ChannelId);
+        writer.WriteEncodedValue(Payload);
+    }
+}

CrystalConnector/Connector/Packet/S2C/S2CSuccessfulPacket.cs

@@ -0,0 +1,12 @@
+using System.Formats.Cbor;
+using CrystalConnector.Handlers;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public class S2CSuccessfulPacket : PacketS2C
+{
+    protected override void WriteData(CborWriter writer)
+    {
+        writer.WriteTextString(HandlerConstants.ResponseSuccessful);
+    }
+}

CrystalConnector/Connector/Packet/S2C/S2CUnauthenticatedPacket.cs

@@ -0,0 +1,12 @@
+using System.Formats.Cbor;
+using CrystalConnector.Handlers;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public class S2CUnauthenticatedPacket : PacketS2C
+{
+    protected override void WriteData(CborWriter writer)
+    {
+        writer.WriteTextString(HandlerConstants.ResponseErrorUnauthenticated);
+    }
+}

CrystalConnector/Connector/Packet/S2C/S2CUndefinedDirectionPacket.cs

@@ -0,0 +1,12 @@
+using System.Formats.Cbor;
+using CrystalConnector.Handlers;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public class S2CUndefinedDirectionPacket : PacketS2C
+{
+    protected override void WriteData(CborWriter writer)
+    {
+        writer.WriteTextString(HandlerConstants.ResponseErrorUndefinedDirection);
+    }
+}

CrystalConnector/Connector/Packet/S2C/S2CUnknownPacket.cs

@@ -0,0 +1,12 @@
+using System.Formats.Cbor;
+using CrystalConnector.Handlers;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public class S2CUnknownPacket : PacketS2C
+{
+    protected override void WriteData(CborWriter writer)
+    {
+        writer.WriteTextString(HandlerConstants.ResponseErrorUnknownPacket);
+    }
+}

CrystalConnector/Connector/Packet/S2C/S2CUnregisteredDirectionPacket.cs

@@ -0,0 +1,12 @@
+using System.Formats.Cbor;
+using CrystalConnector.Handlers;
+
+namespace CrystalConnector.Connector.Packet.S2C;
+
+public class S2CUnregisteredDirectionPacket : PacketS2C
+{
+    protected override void WriteData(CborWriter writer)
+    {
+        writer.WriteTextString(HandlerConstants.ResponseErrorUnregisteredDirection);
+    }
+}

CrystalConnector/Connector/Packet/S2CAuthenticatedPacket.cs

@@ -1,12 +0,0 @@
-using System.Formats.Cbor;
-using CrystalConnector.Handlers;
-
-namespace CrystalConnector.Connector.Packet;
-
-public class S2CAuthenticatedPacket : IPacket
-{
-    public void Write(CborWriter writer)
-    {
-        writer.WriteTextString(HandlerConstants.ResponseAuthenticated);
-    }
-}

CrystalConnector/Connector/WebSocketConnectionManager.cs

@@ -1,4 +1,5 @@
 using System.Net.WebSockets;
+using CrystalConnector.Connector.Packet.S2C;
 using CrystalConnector.Utilities;
 
 namespace CrystalConnector.Connector;
@@ -14,4 +15,18 @@ public class WebSocketConnectionManager
             webSocket.Disconnect();
         }
     }
+
+    public static async Task Broadcast(string origin, string channelId, byte[] payload, bool allowUnauthenticated = false)
+    {
+        var packet = new S2CForwardMessagePacket(origin, channelId, payload);
+        
+        foreach (var (webSocket, info) in Info)
+        {
+            if (info.RegisteredChannels.TryGetValue(channelId, out var value) 
+                && value.Direction.HasFlag(MessageDirection.Incoming))
+            {
+                await webSocket.Send(packet);
+            }
+        }
+    }
 }

CrystalConnector/CrystalConnector.csproj

@@ -32,4 +32,5 @@
     </ItemGroup>
 
 
+
 </Project>

CrystalConnector/Handlers/HandlerConstants.cs

@@ -5,6 +5,12 @@ public class HandlerConstants
     public static readonly string OperationAuthenticate = "Authenticate";
     public static readonly string OperationRegisterChannel = "Register";
     public static readonly string OperationPublish = "Publish";
-
+    
+    public static readonly string ResponseSuccessful = "Successful";
     public static readonly string ResponseAuthenticated = "Authenticated";
+
+    public static readonly string ResponseErrorUnknownPacket = "Unknown";
+    public static readonly string ResponseErrorUnauthenticated = "Unauthenticated";
+    public static readonly string ResponseErrorUndefinedDirection = "UndefinedDirection";
+    public static readonly string ResponseErrorUnregisteredDirection = "UnregisteredDirection";
 }

CrystalConnector/Handlers/WebSocketHandler.cs

@@ -1,7 +1,9 @@
 using System.Formats.Cbor;
 using System.Net.WebSockets;
 using CrystalConnector.Configs;
+using CrystalConnector.Connector;
 using CrystalConnector.Connector.Packet;
+using CrystalConnector.Connector.Packet.S2C;
 using CrystalConnector.Utilities;
 using Microsoft.Extensions.Logging;
 
@@ -19,10 +21,11 @@ public class WebSocketHandler
         Logger = logger;
     }
     
-    public async Task Handle(WebSocket webSocket, CborReader reader)
+    public async Task<PacketS2C> Handle(WebSocket webSocket, CborReader reader)
     {
         reader.ReadStartArray();
         
+        // Todo: qyl27: Use C2SPacketReader.
         var type = reader.ReadTextString();
         if (type == HandlerConstants.OperationAuthenticate)
         {
@@ -35,23 +38,79 @@ public class WebSocketHandler
                     Logger.LogInformation("Client {Name}({Id}) authorized", name, webSocket.GetConnectionInfo().Id);
                     webSocket.GetConnectionInfo().Authenticated = true;
                     webSocket.GetConnectionInfo().Name = name;
-                    await webSocket.Send(new S2CAuthenticatedPacket().Write);
+                    return new S2CSuccessfulPacket();
-                }
-                else
-                {
-                    throw new MessageException("Auth failed!");
                 }
+
+                Logger.LogWarning("Client {Name}({Id}) auth failed with wrong key!", name, webSocket.GetConnectionInfo().Id);
+                return new S2CUnauthenticatedPacket();
             }
-            else
+
-            {
+            return new S2CAuthenticatedPacket();
-                throw new MessageException("You are already authorized.");
-            }
         }
-        else if (type == "")
+
+        if (type == HandlerConstants.OperationRegisterChannel)
         {
+            if (!webSocket.GetConnectionInfo().Authenticated)
+            {
+                Logger.LogWarning("Client {Id} is unauthenticated!", webSocket.GetConnectionInfo().Id);
+                return new S2CUnauthenticatedPacket();
+            }
+
+            var channelId = reader.ReadTextString();
+            var directionValue = reader.ReadInt32();
+
+            if (!Enum.IsDefined(typeof(MessageDirection), directionValue))
+            {
+                Logger.LogWarning("Client {Name}({Id}) tried to register channel {Channel} with undefined direction {Direction}!", 
+                    webSocket.GetConnectionInfo().Name, webSocket.GetConnectionInfo().Id, channelId, directionValue);
+                return new S2CUndefinedDirectionPacket();
+            }
             
+            var direction = (MessageDirection)directionValue;
+                
+            webSocket.GetConnectionInfo().RegisteredChannels.Add(channelId, new MessageChannel
+            {
+                Id = channelId,
+                Direction = direction
+            });
+            Logger.LogInformation("Client {Name}({Id}) registered channel {Channel}(Direction: {Direction})", 
+                webSocket.GetConnectionInfo().Name, webSocket.GetConnectionInfo().Id, channelId, direction.ToString());
+            return new S2CSuccessfulPacket();
         }
-        
+
+        if (type == HandlerConstants.OperationPublish)
+        {
+            var origin = reader.ReadTextString();
+            var channelId = reader.ReadTextString();
+            var payload = reader.ReadEncodedValue();
+
+            if (!webSocket.GetConnectionInfo().RegisteredChannels.TryGetValue(channelId, out var value)
+                || !value.Direction.HasFlag(MessageDirection.Outgoing))
+            {
+                Logger.LogError("Channel {Channel}(Direction: {Direction}) have not been registered by Client {Name}({Id})", 
+                    channelId, MessageDirection.Outgoing, webSocket.GetConnectionInfo().Name, webSocket.GetConnectionInfo().Id);
+                return new S2CUnregisteredDirectionPacket();
+            }
+
+            try
+            {
+                await WebSocketConnectionManager.Broadcast(origin, channelId, payload.ToArray());
+
+                if (Config.IsDebug())
+                {
+                    Logger.LogDebug("Client {Name}({Id}) send data in {Channel}: {Data}", 
+                        origin, webSocket.GetConnectionInfo().Id, channelId, Convert.ToBase64String(payload.Span));
+                    return new S2CSuccessfulPacket();
+                }
+            }
+            catch (Exception ex)
+            {
+                Logger.LogError(ex, "Send error!");
+            }
+        }
+
         reader.ReadEndArray();
+        
+        return new S2CUnknownPacket();
     }
 }

CrystalConnector/Utilities/WebSocketConnectionInfo.cs

@@ -15,5 +15,5 @@ public class WebSocketConnectionInfo
     
     public bool Authenticated { get; set; } = false;
 
-    public List<MessageChannel> RegisteredChannels { get; } = new();
+    public Dictionary<string, MessageChannel> RegisteredChannels { get; } = new();
 }

CrystalConnector/Utilities/WebSocketExtension.cs

@@ -2,6 +2,7 @@
 using System.Net;
 using System.Net.WebSockets;
 using CrystalConnector.Connector;
+using CrystalConnector.Connector.Packet.S2C;
 
 namespace CrystalConnector.Utilities;
 
@@ -33,12 +34,10 @@ public static class WebSocketExtension
         WebSocketConnectionManager.Info.Remove(webSocket);
     }
 
-    public static async Task Send(this WebSocket webSocket, Action<CborWriter> write)
+    public static async Task Send(this WebSocket webSocket, PacketS2C packet)
     {
         var writer = new CborWriter();
-        writer.WriteStartArray(null);
+        packet.Write(writer);
-        write(writer);
-        writer.WriteEndArray();
         await webSocket.Send(writer.Encode());
     }
 

CrystalConnector/WebSockets/WebSocketManager.cs

@@ -42,7 +42,8 @@ public class WebSocketManager : IWebSocketManager
 
             try
             {
-                await Handler.Handle(webSocket, reader);
+                var resultPacket = await Handler.Handle(webSocket, reader);
+                await webSocket.Send(resultPacket);
             }
             catch (CborContentException ex)
             {