CocosPlugin/esengine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server): add Schema validation system and binary encoding optimization (#421)

Browse Source 
* feat(server): add distributed room support

- Add DistributedRoomManager for multi-server room management
- Add MemoryAdapter for testing and standalone mode
- Add RedisAdapter for production multi-server deployments
- Add LoadBalancedRouter with 5 load balancing strategies
- Add distributed config option to createServer
- Add $redirect message for cross-server player redirection
- Add failover mechanism for automatic room recovery
- Add room:migrated and server:draining event types
- Update documentation (zh/en)

* feat(server): add Schema validation system and binary encoding optimization

## Schema Validation System
- Add lightweight schema validation system (s.object, s.string, s.number, etc.)
- Support auto type inference with Infer<> generic
- Integrate schema validation into API/message handlers
- Add defineApiWithSchema and defineMsgWithSchema helpers

## Binary Encoding Optimization
- Add native WebSocket binary frame support via sendBinary()
- Add PacketType.Binary for efficient binary data transmission
- Optimize ECSRoom.broadcastBinary() to use native binary

## Architecture Improvements
- Extract BaseValidator to separate file to eliminate code duplication
- Add ECSRoom export to main index.ts for better discoverability
- Add Core.worldManager initialization check in ECSRoom constructor
- Remove deprecated validate field from ApiDefinition (use schema instead)

## Documentation
- Add Schema validation documentation in Chinese and English

* fix(rpc): resolve ESLint warnings with proper types

- Replace `any` with proper WebSocket type in connection.ts
- Add IncomingMessage type for request handling in index.ts
- Use Record<string, Handler> pattern instead of `any` casting
- Replace `any` with `unknown` in ProtocolDef and type inference
This commit is contained in:
YHH
2026-01-02 17:18:13 +08:00
committed by GitHub
parent 69bb6bd946
commit f333b81298
44 changed files with 8405 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/framework/rpc/package.json
View File

@@ -32,6 +32,8 @@
"build": "tsup && tsc --emitDeclarationOnly",
"build:watch": "tsup --watch",
"type-check": "tsc --noEmit",
"lint": "eslint src --max-warnings 0",
"lint:fix": "eslint src --fix",
"clean": "rimraf dist"
},
"dependencies": {},

232
packages/framework/rpc/src/client/index.ts
View File

@@ -11,11 +11,11 @@ import type {
ApiOutput,
MsgData,
Packet,
ConnectionStatus,
} from '../types'
import { RpcError, ErrorCode } from '../types'
import { json } from '../codec/json'
import type { Codec } from '../codec/types'
ConnectionStatus
} from '../types';
import { RpcError, ErrorCode } from '../types';
import { json } from '../codec/json';
import type { Codec } from '../codec/types';
// ============================================================================
// Re-exports | 类型重导出
@@ -29,9 +29,9 @@ export type {
ApiOutput,
MsgData,
ConnectionStatus,
Codec,
}
export { RpcError, ErrorCode }
Codec
};
export { RpcError, ErrorCode };
// ============================================================================
// Types | 类型定义
@@ -133,11 +133,11 @@ const PacketType = {
ApiResponse: 1,
ApiError: 2,
Message: 3,
Heartbeat: 9,
} as const
Heartbeat: 9
} as const;
const defaultWebSocketFactory: WebSocketFactory = (url) =>
new WebSocket(url) as unknown as WebSocketAdapter
new WebSocket(url) as unknown as WebSocketAdapter;
// ============================================================================
// RpcClient Class | RPC 客户端类
@@ -164,34 +164,34 @@ interface PendingCall {
* ```
*/
export class RpcClient<P extends ProtocolDef> {
private readonly _url: string
private readonly _codec: Codec
private readonly _timeout: number
private readonly _reconnectInterval: number
private readonly _wsFactory: WebSocketFactory
private readonly _options: RpcClientOptions
private readonly _url: string;
private readonly _codec: Codec;
private readonly _timeout: number;
private readonly _reconnectInterval: number;
private readonly _wsFactory: WebSocketFactory;
private readonly _options: RpcClientOptions;
private _ws: WebSocketAdapter | null = null
private _status: ConnectionStatus = 'closed'
private _callIdCounter = 0
private _shouldReconnect: boolean
private _reconnectTimer: ReturnType<typeof setTimeout> | null = null
private _ws: WebSocketAdapter | null = null;
private _status: ConnectionStatus = 'closed';
private _callIdCounter = 0;
private _shouldReconnect: boolean;
private _reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private readonly _pendingCalls = new Map<number, PendingCall>()
private readonly _msgHandlers = new Map<string, Set<(data: unknown) => void>>()
private readonly _pendingCalls = new Map<number, PendingCall>();
private readonly _msgHandlers = new Map<string, Set<(data: unknown) => void>>();
constructor(
_protocol: P,
url: string,
options: RpcClientOptions = {}
) {
this._url = url
this._options = options
this._codec = options.codec ?? json()
this._timeout = options.timeout ?? 30000
this._shouldReconnect = options.autoReconnect ?? true
this._reconnectInterval = options.reconnectInterval ?? 3000
this._wsFactory = options.webSocketFactory ?? defaultWebSocketFactory
this._url = url;
this._options = options;
this._codec = options.codec ?? json();
this._timeout = options.timeout ?? 30000;
this._shouldReconnect = options.autoReconnect ?? true;
this._reconnectInterval = options.reconnectInterval ?? 3000;
this._wsFactory = options.webSocketFactory ?? defaultWebSocketFactory;
}
/**
@@ -199,7 +199,7 @@ export class RpcClient<P extends ProtocolDef> {
* @en Connection status
*/
get status(): ConnectionStatus {
return this._status
return this._status;
}
/**
@@ -207,7 +207,7 @@ export class RpcClient<P extends ProtocolDef> {
* @en Whether connected
*/
get isConnected(): boolean {
return this._status === 'open'
return this._status === 'open';
}
/**
@@ -217,38 +217,38 @@ export class RpcClient<P extends ProtocolDef> {
connect(): Promise<this> {
return new Promise((resolve, reject) => {
if (this._status === 'open' || this._status === 'connecting') {
resolve(this)
return
resolve(this);
return;
}
this._status = 'connecting'
this._ws = this._wsFactory(this._url)
this._status = 'connecting';
this._ws = this._wsFactory(this._url);
this._ws.onopen = () => {
this._status = 'open'
this._options.onConnect?.()
resolve(this)
}
this._status = 'open';
this._options.onConnect?.();
resolve(this);
};
this._ws.onclose = (e) => {
this._status = 'closed'
this._rejectAllPending()
this._options.onDisconnect?.(e.reason)
this._scheduleReconnect()
}
this._status = 'closed';
this._rejectAllPending();
this._options.onDisconnect?.(e.reason);
this._scheduleReconnect();
};
this._ws.onerror = () => {
const err = new Error('WebSocket error')
this._options.onError?.(err)
const err = new Error('WebSocket error');
this._options.onError?.(err);
if (this._status === 'connecting') {
reject(err)
reject(err);
}
}
};
this._ws.onmessage = (e) => {
this._handleMessage(e.data as string | ArrayBuffer)
}
})
this._handleMessage(e.data as string | ArrayBuffer);
};
});
}
/**
@@ -256,12 +256,12 @@ export class RpcClient<P extends ProtocolDef> {
* @en Disconnect
*/
disconnect(): void {
this._shouldReconnect = false
this._clearReconnectTimer()
this._shouldReconnect = false;
this._clearReconnectTimer();
if (this._ws) {
this._status = 'closing'
this._ws.close()
this._ws = null
this._status = 'closing';
this._ws.close();
this._ws = null;
}
}
@@ -275,25 +275,25 @@ export class RpcClient<P extends ProtocolDef> {
): Promise<ApiOutput<P['api'][K]>> {
return new Promise((resolve, reject) => {
if (this._status !== 'open') {
reject(new RpcError(ErrorCode.CONNECTION_CLOSED, 'Not connected'))
return
reject(new RpcError(ErrorCode.CONNECTION_CLOSED, 'Not connected'));
return;
}
const id = ++this._callIdCounter
const id = ++this._callIdCounter;
const timer = setTimeout(() => {
this._pendingCalls.delete(id)
reject(new RpcError(ErrorCode.TIMEOUT, 'Request timeout'))
}, this._timeout)
this._pendingCalls.delete(id);
reject(new RpcError(ErrorCode.TIMEOUT, 'Request timeout'));
}, this._timeout);
this._pendingCalls.set(id, {
resolve: resolve as (v: unknown) => void,
reject,
timer,
})
timer
});
const packet: Packet = [PacketType.ApiRequest, id, name as string, input]
this._ws!.send(this._codec.encode(packet) as string | ArrayBuffer)
})
const packet: Packet = [PacketType.ApiRequest, id, name as string, input];
this._ws!.send(this._codec.encode(packet) as string | ArrayBuffer);
});
}
/**
@@ -301,9 +301,9 @@ export class RpcClient<P extends ProtocolDef> {
* @en Send message
*/
send<K extends MsgNames<P>>(name: K, data: MsgData<P['msg'][K]>): void {
if (this._status !== 'open') return
const packet: Packet = [PacketType.Message, name as string, data]
this._ws!.send(this._codec.encode(packet) as string | ArrayBuffer)
if (this._status !== 'open') return;
const packet: Packet = [PacketType.Message, name as string, data];
this._ws!.send(this._codec.encode(packet) as string | ArrayBuffer);
}
/**
@@ -314,14 +314,14 @@ export class RpcClient<P extends ProtocolDef> {
name: K,
handler: (data: MsgData<P['msg'][K]>) => void
): this {
const key = name as string
let handlers = this._msgHandlers.get(key)
const key = name as string;
let handlers = this._msgHandlers.get(key);
if (!handlers) {
handlers = new Set()
this._msgHandlers.set(key, handlers)
handlers = new Set();
this._msgHandlers.set(key, handlers);
}
handlers.add(handler as (data: unknown) => void)
return this
handlers.add(handler as (data: unknown) => void);
return this;
}
/**
@@ -332,13 +332,13 @@ export class RpcClient<P extends ProtocolDef> {
name: K,
handler?: (data: MsgData<P['msg'][K]>) => void
): this {
const key = name as string
const key = name as string;
if (handler) {
this._msgHandlers.get(key)?.delete(handler as (data: unknown) => void)
this._msgHandlers.get(key)?.delete(handler as (data: unknown) => void);
} else {
this._msgHandlers.delete(key)
this._msgHandlers.delete(key);
}
return this
return this;
}
/**
@@ -350,10 +350,10 @@ export class RpcClient<P extends ProtocolDef> {
handler: (data: MsgData<P['msg'][K]>) => void
): this {
const wrapper = (data: MsgData<P['msg'][K]>) => {
this.off(name, wrapper)
handler(data)
}
return this.on(name, wrapper)
this.off(name, wrapper);
handler(data);
};
return this.on(name, wrapper);
}
// ========================================================================
@@ -362,52 +362,52 @@ export class RpcClient<P extends ProtocolDef> {
private _handleMessage(raw: string | ArrayBuffer): void {
try {
const data = typeof raw === 'string' ? raw : new Uint8Array(raw)
const packet = this._codec.decode(data)
const type = packet[0]
const data = typeof raw === 'string' ? raw : new Uint8Array(raw);
const packet = this._codec.decode(data);
const type = packet[0];
switch (type) {
case PacketType.ApiResponse:
this._handleApiResponse(packet as [number, number, unknown])
break
this._handleApiResponse(packet as [number, number, unknown]);
break;
case PacketType.ApiError:
this._handleApiError(packet as [number, number, string, string])
break
this._handleApiError(packet as [number, number, string, string]);
break;
case PacketType.Message:
this._handleMsg(packet as [number, string, unknown])
break
this._handleMsg(packet as [number, string, unknown]);
break;
}
} catch (err) {
this._options.onError?.(err as Error)
this._options.onError?.(err as Error);
}
}
private _handleApiResponse([, id, result]: [number, number, unknown]): void {
const pending = this._pendingCalls.get(id)
const pending = this._pendingCalls.get(id);
if (pending) {
clearTimeout(pending.timer)
this._pendingCalls.delete(id)
pending.resolve(result)
clearTimeout(pending.timer);
this._pendingCalls.delete(id);
pending.resolve(result);
}
}
private _handleApiError([, id, code, message]: [number, number, string, string]): void {
const pending = this._pendingCalls.get(id)
const pending = this._pendingCalls.get(id);
if (pending) {
clearTimeout(pending.timer)
this._pendingCalls.delete(id)
pending.reject(new RpcError(code, message))
clearTimeout(pending.timer);
this._pendingCalls.delete(id);
pending.reject(new RpcError(code, message));
}
}
private _handleMsg([, path, data]: [number, string, unknown]): void {
const handlers = this._msgHandlers.get(path)
const handlers = this._msgHandlers.get(path);
if (handlers) {
for (const handler of handlers) {
try {
handler(data)
handler(data);
} catch (err) {
this._options.onError?.(err as Error)
this._options.onError?.(err as Error);
}
}
}
@@ -415,25 +415,25 @@ export class RpcClient<P extends ProtocolDef> {
private _rejectAllPending(): void {
for (const [, pending] of this._pendingCalls) {
clearTimeout(pending.timer)
pending.reject(new RpcError(ErrorCode.CONNECTION_CLOSED, 'Connection closed'))
clearTimeout(pending.timer);
pending.reject(new RpcError(ErrorCode.CONNECTION_CLOSED, 'Connection closed'));
}
this._pendingCalls.clear()
this._pendingCalls.clear();
}
private _scheduleReconnect(): void {
if (this._shouldReconnect && !this._reconnectTimer) {
this._reconnectTimer = setTimeout(() => {
this._reconnectTimer = null
this.connect().catch(() => {})
}, this._reconnectInterval)
this._reconnectTimer = null;
this.connect().catch(() => {});
}, this._reconnectInterval);
}
}
private _clearReconnectTimer(): void {
if (this._reconnectTimer) {
clearTimeout(this._reconnectTimer)
this._reconnectTimer = null
clearTimeout(this._reconnectTimer);
this._reconnectTimer = null;
}
}
}
@@ -457,5 +457,5 @@ export function connect<P extends ProtocolDef>(
url: string,
options: RpcClientOptions = {}
): Promise<RpcClient<P>> {
return new RpcClient(protocol, url, options).connect()
return new RpcClient(protocol, url, options).connect();
}

8
packages/framework/rpc/src/codec/index.ts
View File

@@ -3,7 +3,7 @@
* @en Codec Module
*/
export type { Codec } from './types'
export { json } from './json'
export { msgpack } from './msgpack'
export { textEncode, textDecode } from './polyfill'
export type { Codec } from './types';
export { json } from './json';
export { msgpack } from './msgpack';
export { textEncode, textDecode } from './polyfill';

16
packages/framework/rpc/src/codec/json.ts
View File

@@ -3,9 +3,9 @@
* @en JSON Codec
*/
import type { Packet } from '../types'
import type { Codec } from './types'
import { textDecode } from './polyfill'
import type { Packet } from '../types';
import type { Codec } from './types';
import { textDecode } from './polyfill';
/**
* @zh 创建 JSON 编解码器
@@ -17,14 +17,14 @@ import { textDecode } from './polyfill'
export function json(): Codec {
return {
encode(packet: Packet): string {
return JSON.stringify(packet)
return JSON.stringify(packet);
},
decode(data: string | Uint8Array): Packet {
const str = typeof data === 'string'
? data
: textDecode(data)
return JSON.parse(str) as Packet
},
}
: textDecode(data);
return JSON.parse(str) as Packet;
}
};
}

22
packages/framework/rpc/src/codec/msgpack.ts
View File

@@ -3,10 +3,10 @@
* @en MessagePack Codec
*/
import { Packr, Unpackr } from 'msgpackr'
import type { Packet } from '../types'
import type { Codec } from './types'
import { textEncode } from './polyfill'
import { Packr, Unpackr } from 'msgpackr';
import type { Packet } from '../types';
import type { Codec } from './types';
import { textEncode } from './polyfill';
/**
* @zh 创建 MessagePack 编解码器
@@ -16,19 +16,19 @@ import { textEncode } from './polyfill'
* @en Suitable for production, smaller size and faster speed
*/
export function msgpack(): Codec {
const encoder = new Packr({ structuredClone: true })
const decoder = new Unpackr({ structuredClone: true })
const encoder = new Packr({ structuredClone: true });
const decoder = new Unpackr({ structuredClone: true });
return {
encode(packet: Packet): Uint8Array {
return encoder.pack(packet)
return encoder.pack(packet);
},
decode(data: string | Uint8Array): Packet {
const buf = typeof data === 'string'
? textEncode(data)
: data
return decoder.unpack(buf) as Packet
},
}
: data;
return decoder.unpack(buf) as Packet;
}
};
}

72
packages/framework/rpc/src/codec/polyfill.ts
View File

@@ -12,38 +12,38 @@
*/
function getTextEncoder(): { encode(str: string): Uint8Array } {
if (typeof TextEncoder !== 'undefined') {
return new TextEncoder()
return new TextEncoder();
}
return {
encode(str: string): Uint8Array {
const utf8: number[] = []
const utf8: number[] = [];
for (let i = 0; i < str.length; i++) {
let charCode = str.charCodeAt(i)
let charCode = str.charCodeAt(i);
if (charCode < 0x80) {
utf8.push(charCode)
utf8.push(charCode);
} else if (charCode < 0x800) {
utf8.push(0xc0 | (charCode >> 6), 0x80 | (charCode & 0x3f))
utf8.push(0xc0 | (charCode >> 6), 0x80 | (charCode & 0x3f));
} else if (charCode >= 0xd800 && charCode <= 0xdbff) {
i++
const low = str.charCodeAt(i)
charCode = 0x10000 + ((charCode - 0xd800) << 10) + (low - 0xdc00)
i++;
const low = str.charCodeAt(i);
charCode = 0x10000 + ((charCode - 0xd800) << 10) + (low - 0xdc00);
utf8.push(
0xf0 | (charCode >> 18),
0x80 | ((charCode >> 12) & 0x3f),
0x80 | ((charCode >> 6) & 0x3f),
0x80 | (charCode & 0x3f)
)
);
} else {
utf8.push(
0xe0 | (charCode >> 12),
0x80 | ((charCode >> 6) & 0x3f),
0x80 | (charCode & 0x3f)
)
);
}
}
return new Uint8Array(utf8)
},
}
return new Uint8Array(utf8);
}
};
}
/**
@@ -52,55 +52,55 @@ function getTextEncoder(): { encode(str: string): Uint8Array } {
*/
function getTextDecoder(): { decode(data: Uint8Array): string } {
if (typeof TextDecoder !== 'undefined') {
return new TextDecoder()
return new TextDecoder();
}
return {
decode(data: Uint8Array): string {
let str = ''
let i = 0
let str = '';
let i = 0;
while (i < data.length) {
const byte1 = data[i++]
const byte1 = data[i++];
if (byte1 < 0x80) {
str += String.fromCharCode(byte1)
str += String.fromCharCode(byte1);
} else if ((byte1 & 0xe0) === 0xc0) {
const byte2 = data[i++]
str += String.fromCharCode(((byte1 & 0x1f) << 6) | (byte2 & 0x3f))
const byte2 = data[i++];
str += String.fromCharCode(((byte1 & 0x1f) << 6) | (byte2 & 0x3f));
} else if ((byte1 & 0xf0) === 0xe0) {
const byte2 = data[i++]
const byte3 = data[i++]
const byte2 = data[i++];
const byte3 = data[i++];
str += String.fromCharCode(
((byte1 & 0x0f) << 12) | ((byte2 & 0x3f) << 6) | (byte3 & 0x3f)
)
);
} else if ((byte1 & 0xf8) === 0xf0) {
const byte2 = data[i++]
const byte3 = data[i++]
const byte4 = data[i++]
const byte2 = data[i++];
const byte3 = data[i++];
const byte4 = data[i++];
const codePoint =
((byte1 & 0x07) << 18) |
((byte2 & 0x3f) << 12) |
((byte3 & 0x3f) << 6) |
(byte4 & 0x3f)
const offset = codePoint - 0x10000
(byte4 & 0x3f);
const offset = codePoint - 0x10000;
str += String.fromCharCode(
0xd800 + (offset >> 10),
0xdc00 + (offset & 0x3ff)
)
);
}
}
return str
},
}
return str;
}
};
}
const encoder = getTextEncoder()
const decoder = getTextDecoder()
const encoder = getTextEncoder();
const decoder = getTextDecoder();
/**
* @zh 将字符串编码为 UTF-8 字节数组
* @en Encode string to UTF-8 byte array
*/
export function textEncode(str: string): Uint8Array {
return encoder.encode(str)
return encoder.encode(str);
}
/**
@@ -108,5 +108,5 @@ export function textEncode(str: string): Uint8Array {
* @en Decode UTF-8 byte array to string
*/
export function textDecode(data: Uint8Array): string {
return decoder.decode(data)
return decoder.decode(data);
}

2
packages/framework/rpc/src/codec/types.ts
View File

@@ -3,7 +3,7 @@
* @en Codec Type Definitions
*/
import type { Packet } from '../types'
import type { Packet } from '../types';
/**
* @zh 编解码器接口

12
packages/framework/rpc/src/define.ts
View File

@@ -3,7 +3,7 @@
* @en Protocol Definition Module
*/
import type { ApiDef, MsgDef, ProtocolDef } from './types'
import type { ApiDef, MsgDef, ProtocolDef } from './types';
/**
* @zh 创建 API 定义
@@ -15,7 +15,7 @@ import type { ApiDef, MsgDef, ProtocolDef } from './types'
* ```
*/
function api<TInput = void, TOutput = void>(): ApiDef<TInput, TOutput> {
return { _type: 'api' } as ApiDef<TInput, TOutput>
return { _type: 'api' } as ApiDef<TInput, TOutput>;
}
/**
@@ -28,7 +28,7 @@ function api<TInput = void, TOutput = void>(): ApiDef<TInput, TOutput> {
* ```
*/
function msg<TData = void>(): MsgDef<TData> {
return { _type: 'msg' } as MsgDef<TData>
return { _type: 'msg' } as MsgDef<TData>;
}
/**
@@ -49,7 +49,7 @@ function msg<TData = void>(): MsgDef<TData> {
* ```
*/
function define<T extends ProtocolDef>(protocol: T): T {
return protocol
return protocol;
}
/**
@@ -59,5 +59,5 @@ function define<T extends ProtocolDef>(protocol: T): T {
export const rpc = {
define,
api,
msg,
} as const
msg
} as const;

8
packages/framework/rpc/src/index.ts
View File

@@ -38,9 +38,9 @@
* ```
*/
export { rpc } from './define'
export * from './types'
export { rpc } from './define';
export * from './types';
// Re-export client for browser/bundler compatibility
export { RpcClient, connect } from './client/index'
export type { RpcClientOptions, WebSocketAdapter, WebSocketFactory } from './client/index'
export { RpcClient, connect } from './client/index';
export type { RpcClientOptions, WebSocketAdapter, WebSocketFactory } from './client/index';

57
packages/framework/rpc/src/server/connection.ts
View File

@@ -3,37 +3,38 @@
* @en Server Connection Module
*/
import type { Connection, ConnectionStatus } from '../types'
import type { WebSocket } from 'ws';
import type { Connection, ConnectionStatus } from '../types';
/**
* @zh 服务端连接实现
* @en Server connection implementation
*/
export class ServerConnection<TData = unknown> implements Connection<TData> {
readonly id: string
readonly ip: string
data: TData
readonly id: string;
readonly ip: string;
data: TData;
private _status: ConnectionStatus = 'open'
private _socket: any
private _onClose?: () => void
private _status: ConnectionStatus = 'open';
private _socket: WebSocket;
private _onClose?: () => void;
constructor(options: {
id: string
ip: string
socket: any
socket: WebSocket
initialData: TData
onClose?: () => void
}) {
this.id = options.id
this.ip = options.ip
this.data = options.initialData
this._socket = options.socket
this._onClose = options.onClose
this.id = options.id;
this.ip = options.ip;
this.data = options.initialData;
this._socket = options.socket;
this._onClose = options.onClose;
}
get status(): ConnectionStatus {
return this._status
return this._status;
}
/**
@@ -41,8 +42,20 @@ export class ServerConnection<TData = unknown> implements Connection<TData> {
* @en Send raw data
*/
send(data: string | Uint8Array): void {
if (this._status !== 'open') return
this._socket.send(data)
if (this._status !== 'open') return;
this._socket.send(data);
}
/**
* @zh 发送二进制数据（原生 WebSocket 二进制帧）
* @en Send binary data (native WebSocket binary frame)
*
* @zh 直接发送 Uint8Array，不经过 JSON 编码，效率更高
* @en Directly sends Uint8Array without JSON encoding, more efficient
*/
sendBinary(data: Uint8Array): void {
if (this._status !== 'open') return;
this._socket.send(data, { binary: true });
}
/**
@@ -50,12 +63,12 @@ export class ServerConnection<TData = unknown> implements Connection<TData> {
* @en Close connection
*/
close(reason?: string): void {
if (this._status !== 'open') return
if (this._status !== 'open') return;
this._status = 'closing'
this._socket.close(1000, reason)
this._status = 'closed'
this._onClose?.()
this._status = 'closing';
this._socket.close(1000, reason);
this._status = 'closed';
this._onClose?.();
}
/**
@@ -63,6 +76,6 @@ export class ServerConnection<TData = unknown> implements Connection<TData> {
* @en Mark connection as closed (internal use)
*/
_markClosed(): void {
this._status = 'closed'
this._status = 'closed';
}
}

184
packages/framework/rpc/src/server/index.ts
View File

@@ -3,8 +3,8 @@
* @en RPC Server Module
*/
import { WebSocketServer, WebSocket } from 'ws'
import type { Server as HttpServer } from 'node:http'
import { WebSocketServer, WebSocket } from 'ws';
import type { Server as HttpServer } from 'node:http';
import type {
ProtocolDef,
ApiNames,
@@ -13,13 +13,13 @@ import type {
ApiOutput,
MsgData,
Packet,
PacketType,
Connection,
} from '../types'
import { RpcError, ErrorCode } from '../types'
import { json } from '../codec/json'
import type { Codec } from '../codec/types'
import { ServerConnection } from './connection'
Connection
} from '../types';
import type { IncomingMessage } from 'node:http';
import { RpcError, ErrorCode } from '../types';
import { json } from '../codec/json';
import type { Codec } from '../codec/types';
import { ServerConnection } from './connection';
// ============ Types ============
@@ -182,8 +182,8 @@ const PT = {
ApiResponse: 1,
ApiError: 2,
Message: 3,
Heartbeat: 9,
} as const
Heartbeat: 9
} as const;
/**
* @zh 创建 RPC 服务器
@@ -206,16 +206,22 @@ export function serve<P extends ProtocolDef, TConnData = unknown>(
_protocol: P,
options: ServeOptions<P, TConnData>
): RpcServer<P, TConnData> {
const codec = options.codec ?? json()
const connections: ServerConnection<TConnData>[] = []
let wss: WebSocketServer | null = null
let connIdCounter = 0
const codec = options.codec ?? json();
const connections: ServerConnection<TConnData>[] = [];
let wss: WebSocketServer | null = null;
let connIdCounter = 0;
const getClientIp = (ws: WebSocket, req: any): string => {
return req?.headers?.['x-forwarded-for']?.split(',')[0]?.trim()
const getClientIp = (_ws: WebSocket, req: IncomingMessage | undefined): string => {
const forwarded = req?.headers?.['x-forwarded-for'];
const forwardedIp = typeof forwarded === 'string'
? forwarded.split(',')[0]?.trim()
: Array.isArray(forwarded)
? forwarded[0]?.split(',')[0]?.trim()
: undefined;
return forwardedIp
|| req?.socket?.remoteAddress
|| 'unknown'
}
|| 'unknown';
};
const handleMessage = async (
conn: ServerConnection<TConnData>,
@@ -224,23 +230,23 @@ export function serve<P extends ProtocolDef, TConnData = unknown>(
try {
const packet = codec.decode(
typeof data === 'string' ? data : new Uint8Array(data)
)
);
const type = packet[0]
const type = packet[0];
if (type === PT.ApiRequest) {
const [, id, path, input] = packet as [number, number, string, unknown]
await handleApiRequest(conn, id, path, input)
const [, id, path, input] = packet as [number, number, string, unknown];
await handleApiRequest(conn, id, path, input);
} else if (type === PT.Message) {
const [, path, msgData] = packet as [number, string, unknown]
await handleMsg(conn, path, msgData)
const [, path, msgData] = packet as [number, string, unknown];
await handleMsg(conn, path, msgData);
} else if (type === PT.Heartbeat) {
conn.send(codec.encode([PT.Heartbeat]))
conn.send(codec.encode([PT.Heartbeat]));
}
} catch (err) {
options.onError?.(err as Error, conn)
options.onError?.(err as Error, conn);
}
}
};
const handleApiRequest = async (
conn: ServerConnection<TConnData>,
@@ -248,44 +254,46 @@ export function serve<P extends ProtocolDef, TConnData = unknown>(
path: string,
input: unknown
): Promise<void> => {
const handler = (options.api as any)[path]
const apiHandlers = options.api as Record<string, ApiHandler<unknown, unknown, TConnData> | undefined>;
const handler = apiHandlers[path];
if (!handler) {
const errPacket: Packet = [PT.ApiError, id, ErrorCode.NOT_FOUND, `API not found: ${path}`]
conn.send(codec.encode(errPacket))
return
const errPacket: Packet = [PT.ApiError, id, ErrorCode.NOT_FOUND, `API not found: ${path}`];
conn.send(codec.encode(errPacket));
return;
}
try {
const result = await handler(input, conn)
const resPacket: Packet = [PT.ApiResponse, id, result]
conn.send(codec.encode(resPacket))
const result = await handler(input, conn);
const resPacket: Packet = [PT.ApiResponse, id, result];
conn.send(codec.encode(resPacket));
} catch (err) {
if (err instanceof RpcError) {
const errPacket: Packet = [PT.ApiError, id, err.code, err.message]
conn.send(codec.encode(errPacket))
const errPacket: Packet = [PT.ApiError, id, err.code, err.message];
conn.send(codec.encode(errPacket));
} else {
const errPacket: Packet = [PT.ApiError, id, ErrorCode.INTERNAL_ERROR, 'Internal server error']
conn.send(codec.encode(errPacket))
options.onError?.(err as Error, conn)
const errPacket: Packet = [PT.ApiError, id, ErrorCode.INTERNAL_ERROR, 'Internal server error'];
conn.send(codec.encode(errPacket));
options.onError?.(err as Error, conn);
}
}
}
};
const handleMsg = async (
conn: ServerConnection<TConnData>,
path: string,
data: unknown
): Promise<void> => {
const handler = options.msg?.[path as MsgNames<P>]
const msgHandlers = options.msg as Record<string, MsgHandler<unknown, TConnData> | undefined> | undefined;
const handler = msgHandlers?.[path];
if (handler) {
await (handler as any)(data, conn)
await handler(data, conn);
}
}
};
const server: RpcServer<P, TConnData> = {
get connections() {
return connections as ReadonlyArray<Connection<TConnData>>
return connections as ReadonlyArray<Connection<TConnData>>;
},
async start() {
@@ -293,18 +301,18 @@ export function serve<P extends ProtocolDef, TConnData = unknown>(
// 根据配置创建 WebSocketServer
if (options.server) {
// 附加到已有的 HTTP 服务器
wss = new WebSocketServer({ server: options.server })
wss = new WebSocketServer({ server: options.server });
} else if (options.port) {
// 独立创建
wss = new WebSocketServer({ port: options.port })
wss = new WebSocketServer({ port: options.port });
} else {
throw new Error('Either port or server must be provided')
throw new Error('Either port or server must be provided');
}
wss.on('connection', async (ws, req) => {
const id = String(++connIdCounter)
const ip = getClientIp(ws, req)
const initialData = options.createConnData?.() ?? ({} as TConnData)
const id = String(++connIdCounter);
const ip = getClientIp(ws, req);
const initialData = options.createConnData?.() ?? ({} as TConnData);
const conn = new ServerConnection<TConnData>({
id,
@@ -312,70 +320,70 @@ export function serve<P extends ProtocolDef, TConnData = unknown>(
socket: ws,
initialData,
onClose: () => {
const idx = connections.indexOf(conn)
if (idx !== -1) connections.splice(idx, 1)
},
})
const idx = connections.indexOf(conn);
if (idx !== -1) connections.splice(idx, 1);
}
});
connections.push(conn)
connections.push(conn);
ws.on('message', (data) => {
handleMessage(conn, data as string | Buffer)
})
handleMessage(conn, data as string | Buffer);
});
ws.on('close', async (code, reason) => {
conn._markClosed()
const idx = connections.indexOf(conn)
if (idx !== -1) connections.splice(idx, 1)
await options.onDisconnect?.(conn, reason?.toString())
})
conn._markClosed();
const idx = connections.indexOf(conn);
if (idx !== -1) connections.splice(idx, 1);
await options.onDisconnect?.(conn, reason?.toString());
});
ws.on('error', (err) => {
options.onError?.(err, conn)
})
options.onError?.(err, conn);
});
await options.onConnect?.(conn)
})
await options.onConnect?.(conn);
});
// 如果使用已有的 HTTP 服务器，WebSocketServer 不会触发 listening 事件
if (options.server) {
options.onStart?.(0) // 端口由 HTTP 服务器管理
resolve()
options.onStart?.(0); // 端口由 HTTP 服务器管理
resolve();
} else {
wss.on('listening', () => {
options.onStart?.(options.port!)
resolve()
})
options.onStart?.(options.port!);
resolve();
});
}
})
});
},
async stop() {
return new Promise((resolve, reject) => {
if (!wss) {
resolve()
return
resolve();
return;
}
for (const conn of connections) {
conn.close('Server shutting down')
conn.close('Server shutting down');
}
wss.close((err) => {
if (err) reject(err)
else resolve()
})
})
if (err) reject(err);
else resolve();
});
});
},
send(conn, name, data) {
const packet: Packet = [PT.Message, name as string, data]
;(conn as ServerConnection<TConnData>).send(codec.encode(packet))
;(conn as ServerConnection<TConnData>).send(codec.encode(packet));
},
broadcast(name, data, opts) {
const packet: Packet = [PT.Message, name as string, data]
const encoded = codec.encode(packet)
const packet: Packet = [PT.Message, name as string, data];
const encoded = codec.encode(packet);
const excludeSet = new Set(
Array.isArray(opts?.exclude)
@@ -383,15 +391,15 @@ export function serve<P extends ProtocolDef, TConnData = unknown>(
: opts?.exclude
? [opts.exclude]
: []
)
);
for (const conn of connections) {
if (!excludeSet.has(conn)) {
conn.send(encoded)
conn.send(encoded);
}
}
},
}
}
};
return server
return server;
}

35
packages/framework/rpc/src/types.ts
View File

@@ -29,8 +29,8 @@ export interface MsgDef<TData = unknown> {
* @en Protocol definition
*/
export interface ProtocolDef {
readonly api: Record<string, ApiDef<any, any>>
readonly msg: Record<string, MsgDef<any>>
readonly api: Record<string, ApiDef<unknown, unknown>>
readonly msg: Record<string, MsgDef<unknown>>
}
// ============ Type Inference ============
@@ -39,13 +39,13 @@ export interface ProtocolDef {
* @zh 提取 API 输入类型
* @en Extract API input type
*/
export type ApiInput<T> = T extends ApiDef<infer I, any> ? I : never
export type ApiInput<T> = T extends ApiDef<infer I, unknown> ? I : never
/**
* @zh 提取 API 输出类型
* @en Extract API output type
*/
export type ApiOutput<T> = T extends ApiDef<any, infer O> ? O : never
export type ApiOutput<T> = T extends ApiDef<unknown, infer O> ? O : never
/**
* @zh 提取消息数据类型
@@ -120,8 +120,9 @@ export const PacketType = {
ApiResponse: 1,
ApiError: 2,
Message: 3,
Heartbeat: 9,
} as const
Binary: 4,
Heartbeat: 9
} as const;
export type PacketType = typeof PacketType[keyof typeof PacketType]
@@ -173,6 +174,19 @@ export type MessagePacket = [
*/
export type HeartbeatPacket = [type: typeof PacketType.Heartbeat]
/**
* @zh 二进制数据包
* @en Binary data packet
*
* @zh 用于传输原始二进制数据，如 ECS 状态同步
* @en Used for raw binary data transmission, such as ECS state sync
*/
export type BinaryPacket = [
type: typeof PacketType.Binary,
channel: number,
data: Uint8Array
]
/**
* @zh 所有数据包类型
* @en All packet types
@@ -182,6 +196,7 @@ export type Packet =
| ApiResponsePacket
| ApiErrorPacket
| MessagePacket
| BinaryPacket
| HeartbeatPacket
// ============ Error Types ============
@@ -196,8 +211,8 @@ export class RpcError extends Error {
message: string,
public readonly details?: unknown
) {
super(message)
this.name = 'RpcError'
super(message);
this.name = 'RpcError';
}
}
@@ -211,7 +226,7 @@ export const ErrorCode = {
UNAUTHORIZED: 'UNAUTHORIZED',
INTERNAL_ERROR: 'INTERNAL_ERROR',
TIMEOUT: 'TIMEOUT',
CONNECTION_CLOSED: 'CONNECTION_CLOSED',
} as const
CONNECTION_CLOSED: 'CONNECTION_CLOSED'
} as const;
export type ErrorCode = typeof ErrorCode[keyof typeof ErrorCode]