Skip to content

Commit 0e0eb1d

Browse files
fenyingfenying
committed
build(project): refactored writing logic and added full stream supports
1 parent 4163ee9 commit 0e0eb1d

File tree

14 files changed

+566
-389
lines changed

14 files changed

+566
-389
lines changed

‎CHANGES.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,18 @@
11
# Changes
22

3+
## v0.2.2
4+
5+
- feat(connection): added `drain` event supports.
6+
- feat(connection): added `maxMessageSize` readonly property on connections.
7+
- feat(connection): now `createMessageWriter` method returns a `Writable` stream.
8+
- feat(connection): all `write` method supports `callback` parameter for draining.
9+
- feat(connection): refactored the writing logic.
10+
- fix(connection): close the connection correctly on received CLOSE opcode.
11+
- fix(encoding): simplified the code of encoder.
12+
- fix(encoding): failed to decode an empty FIN frame under SIMPLE mode.
13+
- fix(encoding): failed to decode an empty FIN frame under STANDARD mode.
14+
- fix(encoding): a message will be consume multiple times under STANDARD mode.
15+
316
## v0.2.1
417

518
- fix(client): connection close during handshaking should not throw an error.

‎package-lock.json‎

Lines changed: 2 additions & 2 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@litert/websocket",
3-
"version": "0.2.1",
3+
"version": "0.2.2",
44
"description": "The websocket protocol library for LiteRT.",
55
"main": "lib/index.js",
66
"scripts": {

‎src/examples/http/client.ts‎

Lines changed: 64 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -14,18 +14,24 @@
1414
* limitations under the License.
1515
*/
1616

17+
import * as NodeFS from 'node:fs';
1718
import * as $WS from '../../lib';
1819

20+
function writeLog(msg: string): void {
21+
22+
console.info(`[${new Date().toISOString()}] ${msg}`);
23+
}
24+
1925
(async () => {
2026

2127
try {
2228

2329
const cli = await $WS.wsConnect({
2430
'host': '127.0.0.1',
2531
'port': 42096,
26-
'connectTimeout': 500,
32+
'connectTimeout': 50000,
2733
'frameReceiveMode': $WS.EFrameReceiveMode[
28-
process.argv.find(i => i.startsWith('--frame-receive-mode'))
34+
process.argv.find(i => i.startsWith('--frame-receive-mode='))
2935
?.slice('--frame-receive-mode='.length)?.toUpperCase() as 'STANDARD' ?? 'STANDARD'
3036
] ?? $WS.EFrameReceiveMode.STANDARD,
3137
});
@@ -40,65 +46,105 @@ import * as $WS from '../../lib';
4046

4147
switch (msg.opcode) {
4248
case $WS.EOpcode.CLOSE:
43-
console.log(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]: code = ${Buffer.concat(msg.data).readUint16BE()}`);
49+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]: code = ${Buffer.concat(msg.data).readUint16BE()}`);
4450
break;
4551
case $WS.EOpcode.PING:
46-
console.log(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]: ${Buffer.concat(msg.data).toString()}`);
52+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]: ${Buffer.concat(msg.data).toString()}`);
4753
cli.pong(Buffer.concat(msg.data));
4854
break;
4955
default:
50-
console.log(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]:`, Buffer.concat(msg.data).toString());
56+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]:` + Buffer.concat(msg.data).toString());
5157
}
5258
return;
5359
}
5460

5561
switch (msg.opcode) {
5662
case $WS.EOpcode.CLOSE:
5763
msg.toBuffer().then((buf) => {
58-
console.log(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]: code = ${buf.readUint16BE()}`);
64+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]: code = ${buf.readUint16BE()}`);
5965
}, (e) => {
60-
console.error(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]:`, e);
66+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]: ` + e);
6167
});
6268
break;
6369
case $WS.EOpcode.PING:
6470
msg.toBuffer().then((buf) => {
65-
console.log(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]: ${buf.toString()}`);
71+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]: ${buf.toString()}`);
6672
cli.pong(buf);
6773
}, (e) => {
68-
console.error(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]:`, e);
74+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]: ` + e);
6975
});
7076
break;
7177
default:
7278
msg.toString().then((buf) => {
73-
console.log(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]:`, buf);
79+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]:` + buf);
7480
}, (e) => {
75-
console.error(`[${new Date().toISOString()}] Recv [${$WS.EOpcode[msg.opcode]}]:`, e);
81+
writeLog(`Recv [${$WS.EOpcode[msg.opcode]}]: ` + e);
7682
});
7783
}
7884
});
7985

86+
let i = 0;
8087
const timer = setInterval(function(): void {
8188

82-
switch (Math.floor(Math.random() * 3)) {
83-
case 2:
89+
if (!cli.writable) {
90+
91+
clearInterval(timer);
92+
return;
93+
}
94+
95+
switch (Math.floor(Math.random() * 7)) {
96+
case 0:
8497
if (cli.frameReceiveMode !== $WS.EFrameReceiveMode.LITE) {
8598

99+
writeLog(`Sent fragmented text`);
86100
const writer = cli.createMessageWriter($WS.EOpcode.TEXT);
87101

88102
writer.write('hello ');
89103
writer.write('world ');
104+
cli.writeText('hi');
90105
writer.write('angus ');
106+
writer.write('- end with non-empty frame - ');
107+
writer.end((i++).toString());
108+
break;
109+
}
110+
case 1:
111+
if (cli.frameReceiveMode !== $WS.EFrameReceiveMode.LITE) {
112+
113+
writeLog(`Sent fragmented text`);
114+
const writer = cli.createMessageWriter($WS.EOpcode.TEXT);
115+
116+
writer.write('hello ');
117+
writer.write('world ');
118+
cli.writeText('hi');
119+
writer.write('angus ');
120+
writer.write('- end with empty frame - ' + (i++).toString());
91121
writer.end();
92122
break;
93123
}
94124
// fall-through
95-
case 0:
96-
cli.writeText('biu biu biu~');
125+
case 2:
126+
writeLog('Sent non-empty text');
127+
cli.writeText('biu biu biu~' + (i++));
97128
break;
98-
case 1:
99-
cli.writeBinary(Buffer.from('biu biu biu~'));
129+
case 3:
130+
writeLog('Sent empty text');
131+
i++;
132+
cli.writeText('');
133+
break;
134+
case 4:
135+
writeLog('Sent binary');
136+
cli.writeBinary(Buffer.from('biu biu biu~' + (i++)));
137+
break;
138+
case 5:
139+
writeLog('Sent multiple segments of text');
140+
cli.writeText(['hey ', 'hey ', 'hey ', 'guy', (i++).toString()]);
141+
break;
142+
case 6:
143+
if (cli.frameReceiveMode !== $WS.EFrameReceiveMode.LITE) {
144+
writeLog('Sent file');
145+
NodeFS.createReadStream(`${__dirname}/../../tsconfig.json`).pipe(cli.createMessageWriter($WS.EOpcode.TEXT));
146+
}
100147
break;
101-
102148
}
103149

104150
}, 100);

‎src/examples/http/server.ts‎

Lines changed: 28 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -25,16 +25,25 @@ const httpServer = Http.createServer(function(req, resp): void {
2525
resp.end('BAD REQUEST\n');
2626
});
2727

28+
function writeLog(msg: string): void {
29+
30+
console.info(`[${new Date().toISOString()}] ${msg}`);
31+
}
32+
2833
async function socketBody(ws: $WS.IWebSocket): Promise<void> {
2934

3035
const clientId = `${ws.remoteAddress!}:${ws.remotePort!}`;
3136

32-
console.info(`[${new Date().toISOString()}] Client ${clientId} connected!`);
37+
writeLog(`Client ${clientId} connected!`);
3338

3439
let count = 0;
3540

41+
ws.on('close', () => {
42+
43+
writeLog(`Client ${clientId} closed`);
44+
});
3645
ws.on('error', (e) => {
37-
console.error(e);
46+
writeLog((e as any).toString());
3847
});
3948
ws.on('message', (msg) => {
4049

@@ -44,28 +53,28 @@ async function socketBody(ws: $WS.IWebSocket): Promise<void> {
4453

4554
switch (msg.opcode) {
4655
case $WS.EOpcode.CLOSE:
47-
console.log(`[${new Date().toISOString()}] Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: code = ${msg.data.length > 0 ? Buffer.concat(msg.data).readUint16BE() : 'none'}`);
56+
writeLog(`Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: code = ${msg.data.length > 0 ? Buffer.concat(msg.data).readUint16BE() : 'none'}`);
4857

4958
break;
5059
default:
51-
console.log(`[${new Date().toISOString()}] Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: code = ${Buffer.concat(msg.data).toString()}`);
60+
writeLog(`Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: code = ${Buffer.concat(msg.data).toString()}`);
5261
}
5362
return;
5463
}
5564

5665
switch (msg.opcode) {
5766
case $WS.EOpcode.CLOSE:
5867
msg.toBuffer().then((buf) => {
59-
console.log(`[${new Date().toISOString()}] Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: code = ${buf.length ? buf.readUint16BE() : 'none' }`);
68+
writeLog(`Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: code = ${buf.length ? buf.readUint16BE() : 'none' }`);
6069
}, (e) => {
61-
console.error(`[${new Date().toISOString()}] Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]:`, e);
70+
writeLog(`Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: ` + e);
6271
});
6372
break;
6473
default:
6574
msg.toString().then((buf) => {
66-
console.log(`[${new Date().toISOString()}] Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]:`, buf);
75+
writeLog(`Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}] = ${buf}`);
6776
}, (e) => {
68-
console.error(`[${new Date().toISOString()}] Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]:`, e);
77+
writeLog(`Client ${clientId} frame[${$WS.EOpcode[msg.opcode]}]: ` + e);
6978
});
7079
}
7180
});
@@ -74,46 +83,52 @@ async function socketBody(ws: $WS.IWebSocket): Promise<void> {
7483

7584
if (count === 50) {
7685

86+
writeLog(`Bye`);
7787
ws.end();
7888
break;
7989
}
8090

8191
count++;
8292
if (count % 5 === 0) {
8393

94+
writeLog(`Sent ping`);
8495
ws.ping('HELLO');
8596
}
8697
else if (count % 11 === 0) {
8798

99+
writeLog(`Sent text`);
88100
ws.writeText('hello world');
89101
}
90102
else if (count % 7 === 0) {
91103

104+
writeLog(`Sent binary`);
92105
ws.writeBinary(Buffer.from('HELLO world!'));
93106
}
94107
else if (count % 13 === 0 && ws.frameReceiveMode !== $WS.EFrameReceiveMode.LITE) {
95108

109+
writeLog(`Sent fragmented text`);
96110
const writer = ws.createMessageWriter($WS.EOpcode.TEXT);
97111

98112
writer.write('hi ');
99113
writer.end();
100114
}
101115
else {
102116

117+
writeLog(`Sent count text`);
103118
ws.writeText(`test count ${count}`);
104119
}
105120

106121
await setTimeout(100);
107122
}
108123

109124
ws.end();
110-
console.warn(`[${new Date().toISOString()}] Client ${clientId} disconnected!`);
125+
writeLog(`Client ${clientId} disconnected!`);
111126
}
112127

113128
const wsServer = $WS.createServer({
114-
'timeout': 150,
129+
'timeout': 15000,
115130
'frameReceiveMode': $WS.EFrameReceiveMode[
116-
process.argv.find(i => i.startsWith('--frame-receive-mode'))
131+
process.argv.find(i => i.startsWith('--frame-receive-mode='))
117132
?.slice('--frame-receive-mode='.length)?.toUpperCase() as 'STANDARD' ?? 'STANDARD'
118133
] ?? $WS.EFrameReceiveMode.STANDARD,
119134
});
@@ -128,10 +143,10 @@ httpServer.on('upgrade', (request, socket) => {
128143
'X-My-Header': 'Hello World!'
129144
}
130145
});
131-
socketBody(ws).catch(console.error);
146+
socketBody(ws).catch((e) => { writeLog(e.toString()); });
132147
});
133148

134149
// start the server listening on port 42096
135150
httpServer.listen(42096, '0.0.0.0', () => {
136-
console.log('WebSocket server listening on 127.0.0.1:42096');
151+
writeLog('WebSocket server listening on 127.0.0.1:42096');
137152
});

0 commit comments

Comments
 (0)