PipeTransport.js
2.94 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
98
99
100
101
102
103
104
105
106
107
108
109
/**
* PipeTransport — 通过 stdin/stdout 管道通信
*
* 接收 Launcher 返回的 { stdin, stdout, stderr, process } 句柄，
* 绑定事件，使用基类 _handleData 做 NDJSON 解析。
*
* 职责边界：
* - ✅ 消息收发（stdin 写入 / stdout 读取）
* - ✅ 事件监听管理（connect 时绑定 / close 时解绑）
* - ❌ 进程生命周期（启动/终止由 Launcher 负责）
*/
const { Transport } = require('./Transport');
class PipeTransport extends Transport {
constructor() {
super();
this._stdin = null;
this._process = null;
}
get isConnected() {
return this._stdin !== null && !this._stdin.destroyed;
}
/**
* 绑定到已有的 stdin/stdout/stderr 流
*
* @param {object} handle - Launcher.start() 返回的句柄
* @param {Writable} handle.stdin
* @param {Readable} handle.stdout
* @param {Readable} [handle.stderr]
* @param {ChildProcess} [handle.process]
*/
async connect(handle) {
this._stdin = handle.stdin;
this._process = handle.process || null;
// stdout → NDJSON 解析
handle.stdout.on('data', (data) => {
this._handleData(data);
});
// stderr 转发
if (handle.stderr) {
handle.stderr.on('data', (data) => {
this.emit('stderr', data.toString());
});
}
// 进程退出
if (handle.process) {
handle.process.on('exit', (code, signal) => {
this.emit('close', code, signal);
});
handle.process.on('error', (err) => {
this.emit('error', err);
});
}
}
/**
* 发送消息
*/
send(msg) {
if (!this.isConnected) {
throw new Error('[PipeTransport] Transport is closed');
}
// 支持传入 string 或 object
const str = typeof msg === 'string' ? msg : JSON.stringify(msg) + '\n';
this._stdin.write(str);
}
/**
* 关闭 stdin（发送 EOF）— 用于一次性输入场景
*
* 调用后不再可写入，但 stdout/stderr 仍可接收输出。
*/
endInput() {
if (this._stdin && !this._stdin.destroyed) {
this._stdin.end();
}
}
/**
* 关闭连接 — 清理 PipeTransport 自身绑定的事件监听器
*
* 注意：不终止子进程。进程生命周期由 Launcher.stop() 管理。
* 这里只解除 connect() 时在 stdout/stderr/process 上注册的监听器，
* 避免已关闭的 transport 继续接收事件触发回调。
*/
close() {
if (this._process) {
try {
// 解除 connect() 中在底层流上绑定的事件监听器
if (this._process.stdout) this._process.stdout.removeAllListeners('data');
if (this._process.stderr) this._process.stderr.removeAllListeners('data');
this._process.removeAllListeners('exit');
this._process.removeAllListeners('error');
} catch { }
this._process = null;
}
this._stdin = null;
this.buffer = '';
this.removeAllListeners();
}
}
module.exports = { PipeTransport };