✨ It works!

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

ConnectorTest/Program.cs

@@ -24,7 +24,9 @@ async Task Receive()
     while (!result.CloseStatus.HasValue)
     {
         var reader = new CborReader(new ReadOnlyMemory<byte>(buffer, 0, result.Count));
-        Console.WriteLine(reader.BytesRemaining);
+        Console.WriteLine("Response length: " + reader.BytesRemaining);
+        reader.ReadStartArray();
+        Console.WriteLine("Result: " + reader.ReadTextString());
         
         result = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
     }
@@ -44,7 +46,7 @@ async Task Process()
     {
         var uri = Console.ReadLine();
         await webSocket.ConnectAsync(new Uri(uri!), CancellationToken.None);
-        Receive();
+        new Thread(() => Receive()).Start();
     }
     else if (operation == "message")
     {
@@ -80,6 +82,11 @@ async Task Process()
         {
             cborWriter.WriteTextString(valueStr!);
         }
+        else if (type == "byte")
+        {
+            var bytes = Convert.FromBase64String(valueStr);
+            cborWriter.WriteByteString(bytes);
+        }
         else if (type == "int")
         {
             var result = int.TryParse(valueStr, out var value);

ConnectorTest/TestCases/TestCase1.txt

@@ -0,0 +1,38 @@
+connect
+ws://localhost:33927
+
+message
+data
+string
+Authenticate
+data
+string
+qyl27IsHere!
+data
+string
+Client1
+send
+
+message
+data
+string
+Register
+data
+string
+crystalconnector:test
+data
+int
+3
+send
+
+message
+data
+string
+Publish
+data
+string
+crystalconnector:test
+data
+byte
+e2hlbGxvOiB3b3JsZCF9
+send

ConnectorTest/TestCases/TestCase2.txt

@@ -0,0 +1,39 @@
+connect
+ws://localhost:33927
+
+message
+data
+string
+Authenticate
+data
+string
+qyl27IsHere!
+data
+string
+Client2
+send
+
+message
+data
+string
+Register
+data
+string
+crystalconnector:test
+data
+int
+1
+send
+
+message
+data
+string
+Publish
+data
+string
+crystalconnector:test
+data
+byte
+e2hlbGxvOiB3b3JsZCF9
+send
+

CrystalConnector/Connector/Packet/S2C/PacketS2C.cs

@@ -6,9 +6,9 @@ public abstract class PacketS2C : IPacket
 {
     public void Write(CborWriter writer)
     {
-        writer.WriteStartMap(null);
+        writer.WriteStartArray(null);
         WriteData(writer);
-        writer.WriteEndMap();
+        writer.WriteEndArray();
     }
 
     protected abstract void WriteData(CborWriter writer);

CrystalConnector/Connector/Packet/S2C/S2CForwardMessagePacket.cs

@@ -1,4 +1,5 @@
 using System.Formats.Cbor;
+using CrystalConnector.Handlers;
 
 namespace CrystalConnector.Connector.Packet.S2C;
 
@@ -17,8 +18,9 @@ public class S2CForwardMessagePacket : PacketS2C
     
     protected override void WriteData(CborWriter writer)
     {
+        writer.WriteTextString(HandlerConstants.ResponseForward);
         writer.WriteTextString(Origin);
         writer.WriteTextString(ChannelId);
-        writer.WriteEncodedValue(Payload);
+        writer.WriteByteString(Payload);
     }
 }

CrystalConnector/Connector/Packet/S2C/S2CNameAuthenticatedPacket.cs

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

CrystalConnector/Connector/WebSocketConnectionManager.cs

@@ -8,6 +8,19 @@ public class WebSocketConnectionManager
 {
     internal static Dictionary<WebSocket, WebSocketConnectionInfo> Info { get; } = new();
 
+    public static bool HaveNameRegistered(string name)
+    {
+        foreach (var (webSocket, info) in Info)
+        {
+            if (info.Name == name)
+            {
+                return true;
+            }
+        }
+        
+        return false;
+    }
+
     public static void DisconnectAll()
     {
         foreach (var (webSocket, info) in Info)
@@ -23,8 +36,15 @@ public class WebSocketConnectionManager
         foreach (var (webSocket, info) in Info)
         {
             if (info.RegisteredChannels.TryGetValue(channelId, out var value) 
-                && value.Direction.HasFlag(MessageDirection.Incoming))
+                && value.Direction.HasFlag(MessageDirection.Incoming) 
+                && info.Name != origin)
             {
+                if (webSocket.State != WebSocketState.Open)
+                {
+                    webSocket.Purge();
+                    continue;
+                }
+                
                 await webSocket.Send(packet);
             }
         }

CrystalConnector/Handlers/HandlerConstants.cs

@@ -9,10 +9,12 @@ public class HandlerConstants
     // public static readonly string OperationExit = "Bye";
     
     public static readonly string ResponseSuccessful = "Successful";
+    public static readonly string ResponseForward = "Forward";
     
     public static readonly string ResponseErrorUnknownPacket = "Unknown";
     public static readonly string ResponseErrorMalformed = "Malformed";
     public static readonly string ResponseErrorAuthenticated = "Authenticated";
+    public static readonly string ResponseErrorNameAuthenticated = "NameAuthenticated";
     public static readonly string ResponseErrorUnauthenticated = "Unauthenticated";
     public static readonly string ResponseErrorUndefinedDirection = "UndefinedDirection";
     public static readonly string ResponseErrorUnregisteredDirection = "UnregisteredDirection";

CrystalConnector/Handlers/WebSocketHandler.cs

@@ -35,6 +35,12 @@ public class WebSocketHandler
                 var name = reader.ReadTextString();
                 if (secret == Config.GetSecretKey())
                 {
+                    if (WebSocketConnectionManager.HaveNameRegistered(name))
+                    {
+                        Logger.LogInformation("Client {Id} tried authorize with name {Name}, but it is already authorized!", webSocket.GetConnectionInfo().Id, name);
+                        return new S2CNameAuthenticatedPacket();
+                    }
+                    
                     Logger.LogInformation("Client {Name}({Id}) authorized", name, webSocket.GetConnectionInfo().Id);
                     webSocket.GetConnectionInfo().Authenticated = true;
                     webSocket.GetConnectionInfo().Name = name;
@@ -69,9 +75,9 @@ public class WebSocketHandler
             var direction = (MessageDirection)directionValue;
 
             var channels = webSocket.GetConnectionInfo().RegisteredChannels;
-            if (channelId.Contains(channelId))
+            if (channels.TryGetValue(channelId, out var channel))
             {
-                channels[channelId].Direction |= direction;
+                channel.Direction |= direction;
             }
             else
             {
@@ -89,26 +95,26 @@ public class WebSocketHandler
 
         if (type == HandlerConstants.OperationPublish)
         {
-            var origin = reader.ReadTextString();
             var channelId = reader.ReadTextString();
-            var payload = reader.ReadEncodedValue();
+            var payload = reader.ReadByteString();
 
             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})", 
+                Logger.LogWarning("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());
+                var name = webSocket.GetConnectionInfo().Name;
+                await WebSocketConnectionManager.Broadcast(name!, channelId, payload);
 
                 if (Config.IsDebug())
                 {
                     Logger.LogDebug("Client {Name}({Id}) send data in {Channel}: {Data}", 
-                        origin, webSocket.GetConnectionInfo().Id, channelId, Convert.ToBase64String(payload.Span));
+                        name, webSocket.GetConnectionInfo().Id, channelId, Convert.ToBase64String(payload));
                     return new S2CSuccessfulPacket();
                 }
             }

CrystalConnector/Startup.cs

@@ -36,7 +36,7 @@ public class Startup
             if (context.WebSockets.IsWebSocketRequest)
             {
                 var websocket = await context.WebSockets.AcceptWebSocketAsync();
-                Logger.LogInformation("Got WebSocket connection from {Ip}:{Port}", context.Connection.RemoteIpAddress?.ToString(), context.Connection.RemotePort);
+                Logger.LogInformation("Got WebSocket connection from {Ip}:{Port}({Id})", context.Connection.RemoteIpAddress?.ToString(), context.Connection.RemotePort, context.Connection.Id);
 
                 var taskCompletionSource = new TaskCompletionSource<WebSocketHandleResult>();
                 await WebSocketManager.StartHandle(context, websocket, taskCompletionSource);

CrystalConnector/WebSockets/WebSocketManager.cs

@@ -69,11 +69,19 @@ public class WebSocketManager : IWebSocketManager
                 Logger.LogWarning(ex, "Why do that?");
                 await webSocket.Send(new S2CMalformedPacket());
             }
-            
-            receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
-        }
 
-        webSocket.Purge();
+            try
+            {
+                receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
+            }
+            catch (Exception ex)
+            {
+                Logger.LogInformation("Client {Name}({Id}) disconnected!", 
+                    webSocket.GetConnectionInfo().Name, webSocket.GetConnectionInfo().Id);
+                webSocket.Purge();
+                break;
+            }
+        }
     }
 
     public void Disconnect(WebSocket webSocket)

README.md

@@ -5,6 +5,8 @@
 
 ## 传输协议
 
+版本：**V1.0（2023.9.17）**
+
 水晶互联采用 WebSocket 进行消息传输，使用 CBOR 进行消息编码。
 
 为了贴近 Guava 中 `ByteArrayDataInput` / `ByteArrayDataOutput` 的使用习惯，规定以单层基本类型的数组形式使用 CBOR 。（Todo：支持更复杂的结构。）
@@ -23,6 +25,8 @@ reader.ReadStartArray();
 reader.ReadEndArray();
 ```
 
+
+
 ### 文档约定
 
 文档中的代码块，以行首的 `>` 代表客户端向服务端传输的消息；以行首的 `<` 代表服务端向客户端传输的消息。其后的消息以纯文本形式发送。如需某一端进行操作，在尖括号后以 `!` 标志体现。
@@ -33,13 +37,100 @@ reader.ReadEndArray();
 
 文档中使用尖括号对 `<type:Name>` 代表必填参数，方括号对 `[type:Name]` 代表可选参数。 `type` 是参数类型，如 `string` 、 `int` 等。 `Name` 是参数名，不在具体传输过程中体现，是方便开发人员记忆的名称。
 
-### 客户端 to 服务端
+
+
+### 协议约定
+
+**服务端不会主动断开客户端的链接。**
+
+当客户端操作成功时，服务端必定响应：
+
+```
+< Successful
+```
+
+当操作失败时，服务端需要指出（供机器处理的）原因：
+
+```
+< <string:Reason>
+```
+
+
+
+#### 操作失败原因列表
+
+目前已定义的原因列表如下：
+
+| 原因                  | 说明                                                         |
+| --------------------- | ------------------------------------------------------------ |
+| Unknown               | 服务端无法处理客户端所发送的数据包，产生此错误。             |
+| Malformed             | 客户端发给服务端的信息是 CBOR 数据包，但无法按协议规则解析时，产生此错误。 |
+| Authenticated         | 当已认证客户端再次申请认证时，产生此错误。                   |
+| NameAuthenticated     | 当某个客户端尝试以一个已认证的客户端身份进行认证时，产生此错误。 |
+| Unauthenticated       | 当未认证客户端执行除认证外的其他操作时，产生此错误。         |
+| UndefinedDirection    | 当客户端尝试注册的频道信息方向超过了信息方向枚举允许的值时，产生此错误。 |
+| UnregisteredDirection | 当客户端未注册外发某个频道的信息，却想要发布对应频道信息时，产生此错误。 |
+
+
+
+#### 数据结构
+
+##### 消息方向（MessageDirection）
+
+```
+enum MessageDirection {
+	None = 0,
+	Incoming = 1,
+	Outgoing = 2,
+	All = Incoming | Outgoing	// 3
+}
+```
+
+
+
+##### ID 字符串（IdString）
+
+一个字符串，满足正则表达式 `/a-z0-9_/` 。
+
+例如，以下每一行都是合法的 ID 字符串：
+
+```
+crystalconnector
+cat_messenger
+qyl27
+```
+
+而以下每一行都是非法的的 ID 字符串：
+
+```
+CristalConnector
+@qyl28
+Hello, world!
+```
+
+
+
+##### 频道 ID （ChannelId）
+
+```
+<IdString:Namespace>:<IdString:ChannelName>
+```
+
+Namespace 可以是注册这一频道的项目名称，尽量避免冲突。
+
+ChannelName 是自定义的频道名。
+
+**目前版本的服务端对于频道ID没有严格规定，理论上是任意字符串都支持，但是推荐客户端遵守上述格式，以免被日后的更新破坏。**
+
+
+
+### 协议操作
 
 #### 认证客户端
 
 为了防止未授权的传输，水晶互联需要一个 SecretKey 来信任客户端，它由服务端配置文件指定。（这个设计没有任何安全性的保证，请保证不在公网中使用它。）
 
-认证请求：
+请求：
 
 ```
 > Authenticate
@@ -47,17 +138,42 @@ reader.ReadEndArray();
 > <string:ClientName>
 ```
 
-认证成功：
 
-```
-< Authenticated
-```
-
-认证失败：
-
-```
-<! Close
-```
 
 #### 注册频道
 
+为了降低处理压力，客户端需要在启动时向服务端注册频道及消息方向。否则将无法发送或接收消息。
+
+请求：
+
+```
+> Register
+> <ChannelId:ChannelId>
+> <int32:EnumDirection>
+```
+
+
+
+#### 发布消息
+
+请求：
+
+```
+> Publish
+> <ChannelId:ChannelId>
+> <byte[]:Payload>
+```
+
+Payload 是一个字节型数组，会被原样发给其他的客户端（不会发回给自身）。
+
+当服务端收到发布消息的请求时，所有注册了该频道的客户端都会收到如下信息：
+
+```
+< Forward
+< <string:PublisherName>
+< <ChannelId:ChannelId>
+< <byte[]:Payload>
+```
+
+PublisherName 是先前注册这一频道的客户端。
+