WebSocketTransport.js
2.41 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
/**
* WebSocketTransport - WebSocket 传输层实现
*
* 通过 WebSocket 连接到远程 AgentBridge，使用 NDJSON 格式通信。
* 与 ProcessTransport 共享基类的 _handleData() 实现。
*/
const WebSocket = require('ws');
const { Transport } = require('./Transport');
class WebSocketTransport extends Transport {
constructor() {
super();
this.ws = null;
this._url = null;
}
/**
* 检查是否已连接
*/
get isConnected() {
return this.ws !== null && this.ws.readyState === WebSocket.OPEN;
}
/**
* 连接到 WebSocket Server (AgentBridge)
* @param {string} url - ws:// 地址
* @param {string[]} _args - 忽略（兼容 Transport 接口签名）
* @param {object} options - { timeout }
*/
async connect(url, _args = [], options = {}) {
this._url = url;
const timeout = options.timeout || 10000;
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
reject(new Error(`WebSocket connect timeout: ${url} (${timeout}ms)`));
}, timeout);
console.log(`[WebSocketTransport] 连接: ${url}`);
this.ws = new WebSocket(url);
this.ws.on('open', () => {
clearTimeout(timer);
console.log(`[WebSocketTransport] 已连接: ${url}`);
resolve();
});
this.ws.on('error', (err) => {
clearTimeout(timer);
console.error(`[WebSocketTransport] 错误: ${err.message}`);
this.emit('error', err);
reject(err);
});
this.ws.on('message', (data) => {
// 使用基类的 _handleData 处理 NDJSON
this._handleData(data.toString());
});
this.ws.on('close', (code, reason) => {
console.log(`[WebSocketTransport] 连接关闭: code=${code}`);
this.emit('close', code, reason?.toString());
});
});
}
/**
* 发送消息
* @param {object|string} msg - 消息对象或已序列化的字符串
*/
send(msg) {
if (!this.isConnected) {
throw new Error('Transport is closed');
}
// 支持传入 string 或 object（与 PipeTransport 行为一致）
const str = typeof msg === 'string' ? msg : JSON.stringify(msg) + '\n';
this.ws.send(str);
}
/**
* 关闭连接
*/
close() {
if (this.ws) {
try {
this.ws.close();
} catch {
// ignore
}
this.ws = null;
}
this.buffer = '';
}
}
module.exports = { WebSocketTransport };