根据 fluffos/src/net/websocket.cc 及相关代码分析，FluffOS 支持的 WebSocket 子协议（模式）及其特性如下：
1. 子协议列表
通过 protocols 数组定义，共支持 4 个子协议名称，对应 3 种核心处理模式：
static struct lws_protocols protocols[] = {
{"http", lws_callback_http_dummy, 0, 0, WS_HTTP}, // HTTP 协议（非 WebSocket）
{"ascii", ws_ascii_callback, sizeof(struct ws_ascii_session), 4096, WS_ASCII}, // ASCII 模式
{"telnet", ws_telnet_callback, sizeof(struct ws_telnet_session), 4096, WS_TELNET}, // Telnet 模式
{"binary", ws_telnet_callback, sizeof(struct ws_telnet_session), 4096, WS_TELNET}, // 兼容旧版的 Binary 模式（复用 Telnet 逻辑）
{NULL, NULL, 0, 0} /* 终止符 */
};2. 各子协议详解
（1）ascii 子协议（WS_ASCII 模式）
- 核心处理：由
ws_ascii_callback回调函数处理，关联ws_ascii_session会话结构。 - 数据特性：
- 仅支持 UTF-8 编码的文本数据，会严格校验输入的 UTF-8 合法性（依赖
LWS_SERVER_OPTION_VALIDATE_UTF8配置）。 - 不处理 Telnet 控制指令，仅传输纯文本内容（如命令输出、JSON 消息等）。
- 仅支持 UTF-8 编码的文本数据，会严格校验输入的 UTF-8 合法性（依赖
- 发送逻辑：通过
ws_ascii_send函数发送，确保数据符合 UTF-8 规范。 - 适用场景：纯文本交互场景，需严格保证文本编码正确性。
（2）telnet 子协议（WS_TELNET 模式）
- 核心处理：由
ws_telnet_callback回调函数处理，关联ws_telnet_session会话结构。 - 数据特性：
- 支持 模拟 Telnet 终端交互，会解析和处理 Telnet 控制指令（如
IAC协商、终端选项等）。 - 兼容传统 Telnet 客户端功能（如回显、换行处理、终端尺寸协商等）。
- 支持 模拟 Telnet 终端交互，会解析和处理 Telnet 控制指令（如
- 发送逻辑：通过
ws_telnet_send函数发送，会对特殊控制字符进行转义处理。 - 适用场景：需要模拟 Telnet 终端的交互场景（如远程命令行操作）。
（3）binary 子协议（兼容模式）
- 核心处理：名称为
binary，但复用telnet模式的ws_telnet_callback回调和会话结构，本质属于WS_TELNET模式。 - 设计目的：为兼容 FluffOS 2.x 版本保留，确保旧版客户端使用
binary子协议时能正常工作。 - 特性：行为与
telnet模式一致，支持 Telnet 控制逻辑，并非原生二进制流传输（名称可能存在历史遗留命名差异）。
（4）http 子协议（非 WebSocket 模式）
- 核心处理：由
lws_callback_http_dummy处理，用于兼容普通 HTTP 请求（如静态文件访问）。 - 特性：不属于 WebSocket 协议，仅作为 HTTP 服务的默认回调，不参与 WebSocket 数据交互。
3. 协议选择与交互逻辑
- 客户端连接：客户端通过
new WebSocket(url, "子协议名称")指定模式（如telnet或ascii），服务端根据名称匹配对应处理逻辑。 - 数据发送：
websocket_send_text函数会根据当前连接的协议 ID（WS_TELNET或WS_ASCII）自动调用对应发送函数（ws_telnet_send或ws_ascii_send）。 - 兼容性：
binary子协议仅为兼容旧版本存在，新客户端建议优先使用telnet或ascii模式。
总结
FluffOS 核心支持 ascii（纯文本 UTF-8） 和 telnet（Telnet 终端兼容） 两种 WebSocket 子协议模式，同时通过 binary 子协议保持对旧版本的兼容。选择时需根据场景需求：纯文本交互用 ascii，终端模拟用 telnet。
在线websocket测试网站其实有很多的，但是因为MUD游戏的特殊性，无法直接用这些测试网站测试MUD的WS功能。所以自己动手撸了一个页面，欢迎使用。地址如下:
提示：你修改地址和端口为你本机地址可以直接连你的单机，比如：127.0.0.1，但记得改协议为ws
关于WebSocket这里不多做介绍，在fluffos v2019版中引入了websocket的支持，可以直接使用网页连接游戏，驱动自带了一个简单的服务端,使用xterm.js模拟终端。
这里简单分享一下MUD WebSocket的注意事项:
- 在使用WS时需要注意指定协议为"ascii",如:
ws = new WebSocket("ws://mud.ren:8888", "ascii"); ws.send()需要发送\n,如:ws.send("chat* hi\n");- 连接游戏后,服务器返回的数据是
blob格式,这个要做转码处理，可参考以下代码：ws.onmessage = function (event) { let data = event.data; let textarea = document.getElementById("mud"); if (data instanceof Blob) { let reader = new FileReader(); reader.readAsText(data, 'utf-8'); reader.onload = function (e) { textarea.innerHTML += reader.result; textarea.scrollTop = textarea.scrollHeight; } } };
如果还有问题，请直接参考项目页面源码，对比分析一下吧😘。
以下是Python版websocket客户端，直接保存为websocket_client.py，运行python websocket_client.py启动试试。
import asyncio
import websockets
import json
import threading
import sys
from datetime import datetime
class MUDClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
async def connect(self):
try:
self.websocket = await websockets.connect(
self.uri,
subprotocols=['ascii']
)
self.connected = True
print(f"✅ 已连接到MUD服务器: {self.uri}")
print("🎮 MUD游戏客户端已启动")
print("💡 输入 /help 查看可用命令")
print("-" * 50)
# 启动接收消息任务
receive_task = asyncio.create_task(self.receive_messages())
# 启动发送消息任务
send_task = asyncio.create_task(self.send_messages())
await asyncio.gather(receive_task, send_task)
except Exception as e:
print(f"❌ 连接错误: {e}")
self.connected = False
async def receive_messages(self):
"""接收服务器消息"""
try:
async for message in self.websocket:
timestamp = datetime.now().strftime("%H:%M:%S")
# 处理原始字节数据（Blob格式）
if isinstance(message, bytes):
try:
text = message.decode('utf-8')
self.display_message(text, timestamp)
except UnicodeDecodeError:
# 如果UTF-8解码失败，尝试其他编码
try:
text = message.decode('gb2312')
self.display_message(text, timestamp)
except:
text = message.decode('latin-1')
self.display_message(text, timestamp)
elif isinstance(message, str):
self.display_message(message, timestamp)
else:
print(f"📨 [{timestamp}] 未知数据类型: {type(message)}")
except websockets.exceptions.ConnectionClosed:
print("🔌 连接已关闭")
self.connected = False
except Exception as e:
print(f"❌ 接收消息错误: {e}")
self.connected = False
def display_message(self, text, timestamp):
"""显示消息"""
# 移除末尾的换行符
text = text.rstrip('\r\n')
# 如果是空消息，不显示
if not text.strip():
return
print(text)
async def send_messages(self):
"""发送用户输入到服务器"""
loop = asyncio.get_event_loop()
while self.connected:
try:
# 使用线程池读取输入，避免阻塞
user_input = await loop.run_in_executor(None, input)
if not self.connected:
break
if not user_input.strip():
# 空输入（直接按Enter）发送换行符，用于分页
await self.websocket.send('\n')
continue
# 处理特殊命令
if user_input.startswith('/'):
await self.handle_command(user_input)
else:
# 普通消息直接发送，确保以换行符结尾
message = user_input.strip()
if not message.endswith('\n'):
message += '\n'
await self.websocket.send(message)
except websockets.exceptions.ConnectionClosed:
print("🔌 连接已关闭")
break
except Exception as e:
print(f"❌ 发送消息错误: {e}")
break
async def handle_command(self, command):
"""处理特殊命令"""
parts = command.strip().split(' ', 1)
cmd = parts[0].lower()
args = parts[1] if len(parts) > 1 else ""
if cmd == '/help':
print("\n📋 可用命令:")
print(" /help - 显示此帮助")
print(" /quit - 退出客户端")
print(" /clear - 清屏")
print(" 其他命令直接输入即可发送")
print()
elif cmd == '/quit':
print("👋 正在退出...")
await self.websocket.close()
self.connected = False
elif cmd == '/clear':
os.system('cls' if os.name == 'nt' else 'clear')
else:
# 未知命令直接发送
message = command[1:].strip()
if not message.endswith('\n'):
message += '\n'
await self.websocket.send(message)
import os
async def main():
uri = "ws://mud.ren:8888"
client = MUDClient(uri)
await client.connect()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n👋 客户端已退出")