Skip to content

Commit ac0b282

Browse files
awlaytonawlayton
committed
perf(websockets): streamline websocket handling
1 parent 0eb3557 commit ac0b282

File tree

16 files changed

+133
-142
lines changed

16 files changed

+133
-142
lines changed

‎oada/libs/lib-arangodb/package.json‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@
5050
"dependencies": {
5151
"@oada/lib-config": "^3.5.1",
5252
"@oada/oadaify": "^2.1.0",
53-
"@oada/types": "^3.0.6",
53+
"@oada/types": "^3.1.1",
5454
"arangojs": "^7.8.0",
5555
"bcryptjs": "^2.4.3",
5656
"clone-deep": "^4.0.1",
@@ -69,7 +69,7 @@
6969
"@types/deep-equal": "^1.0.1",
7070
"@types/flat": "^5.0.2",
7171
"@types/json-pointer": "^1.0.31",
72-
"@types/node": "^18.6.1",
72+
"@types/node": "^18.6.2",
7373
"ava": "4.0.0-rc.1",
7474
"type-fest": "^2.17.0"
7575
},

‎oada/libs/lib-kafka/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@
4545
"@types/bluebird": "^3.5.36",
4646
"@types/convict": "^6.1.1",
4747
"@types/debug": "^4.1.7",
48-
"@types/node": "^18.6.1",
48+
"@types/node": "^18.6.2",
4949
"@types/uuid": "^8.3.4",
5050
"ava": "4.0.0-rc.1"
5151
},

‎oada/libs/lib-prom/package.json‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@
3838
"devDependencies": {
3939
"@ava/typescript": "^3.0.1",
4040
"@types/convict": "^6.1.1",
41-
"@types/node": "^18.6.1",
41+
"@types/node": "^18.6.2",
4242
"@types/ws": "^8.5.3",
4343
"ava": "4.0.0-rc.1",
4444
"fastify-plugin": "^4.0.0"
@@ -47,6 +47,6 @@
4747
"node": "18.5.0"
4848
},
4949
"optionalDependencies": {
50-
"nstats": "^4.2.0"
50+
"nstats": "^5.0.0"
5151
}
5252
}

‎oada/libs/pino-debug/package.json‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,15 @@
2020
"dependencies": {
2121
"cls-rtracer": "^2.6.2",
2222
"is-interactive": "^1.0.0",
23-
"pino": "^8.3.0",
23+
"pino": "^8.3.1",
2424
"pino-caller": "^3.2.0",
2525
"pino-debug": "^2.0.0",
2626
"pino-pretty": "^8.1.0",
2727
"tslib": "^2.4.0"
2828
},
2929
"devDependencies": {
3030
"@types/debug": "^4.1.7",
31-
"@types/node": "^18.6.1"
31+
"@types/node": "^18.6.2"
3232
},
3333
"peerDependencies": {
3434
"debug": "*"

‎oada/package.json‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
"@tsconfig/node18": "^1.0.1",
1919
"@types/eslint": "^8.4.5",
2020
"@types/mocha": "^9.1.1",
21-
"@types/node": "^18.6.1",
21+
"@types/node": "^18.6.2",
2222
"@typescript-eslint/eslint-plugin": "^5.31.0",
2323
"@typescript-eslint/parser": "^5.31.0",
2424
"@yarnpkg/sdks": "^3.0.0-rc.14",
@@ -43,7 +43,7 @@
4343
"eslint-plugin-optimize-regex": "^1.2.1",
4444
"eslint-plugin-prettier": "^4.2.1",
4545
"eslint-plugin-promise": "^6.0.0",
46-
"eslint-plugin-regexp": "^1.7.0",
46+
"eslint-plugin-regexp": "^1.8.0",
4747
"eslint-plugin-security": "^1.5.0",
4848
"eslint-plugin-sonarjs": "^0.14.0",
4949
"eslint-plugin-unicorn": "^43.0.2",

‎oada/services/auth/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@
4848
"@oada/lib-prom": "workspace:^",
4949
"@oada/lookup": "^2.0.0",
5050
"@oada/pino-debug": "^3.5.1",
51-
"@oada/types": "^3.0.6",
51+
"@oada/types": "^3.1.1",
5252
"@oada/well-known-json": "^2.0.1",
5353
"arangojs": "^7.8.0",
5454
"axios": "^0.27.2",

‎oada/services/http-handler/package.json‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -40,21 +40,22 @@
4040
"@fastify/sensible": "^5.1.0",
4141
"@fastify/websocket": "^6.0.1",
4242
"@oada/error": "^2.0.1",
43-
"@oada/formats-server": "^3.0.2",
43+
"@oada/formats-server": "^3.1.1",
4444
"@oada/lib-arangodb": "^3.5.1",
4545
"@oada/lib-config": "^3.5.1",
4646
"@oada/lib-kafka": "^3.5.1",
4747
"@oada/lib-prom": "workspace:^",
4848
"@oada/oadaify": "^2.1.0",
4949
"@oada/permissions-handler": "^3.5.1",
5050
"@oada/pino-debug": "^3.5.1",
51-
"@oada/types": "^3.0.6",
51+
"@oada/types": "^3.1.1",
5252
"cacache": "^16.1.1",
5353
"debug": "^4.3.4",
5454
"es-main": "^1.2.0",
55+
"eventemitter3": "^4.0.7",
5556
"fastify": "4.3.0",
5657
"fastify-graceful-shutdown": "^3.2.0",
57-
"fastify-healthcheck": "^4.0.0",
58+
"fastify-healthcheck": "^4.1.0",
5859
"json-ptr": "^3.1.1",
5960
"ksuid": "^3.0.0",
6061
"tslib": "^2.4.0",
@@ -66,12 +67,12 @@
6667
"@oada/write-handler": "^3.5.1",
6768
"@types/cacache": "^15.0.1",
6869
"@types/debug": "^4.1.7",
69-
"@types/node": "^18.6.1",
70+
"@types/node": "^18.6.2",
7071
"@types/type-is": "^1.6.3",
7172
"@types/uuid": "^8.3.4",
7273
"@types/ws": "^8.5.3",
7374
"connect": "^3.7.0",
74-
"find-my-way": "^7.0.0",
75+
"find-my-way": "^7.0.1",
7576
"light-my-request": "^5.3.0"
7677
},
7778
"volta": {

‎oada/services/http-handler/src/websockets.ts‎

Lines changed: 15 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,11 @@
1717

1818
import { config } from './config.js';
1919

20-
import { EventEmitter } from 'node:events';
2120
import { strict as assert } from 'node:assert';
2221
import { promisify } from 'node:util';
2322

23+
import EventEmitter from 'eventemitter3';
2424
import type { FastifyPluginAsync } from 'fastify';
25-
import { JsonPointer } from 'json-ptr';
2625
import type LightMyRequest from 'light-my-request';
2726
import type WebSocket from 'ws';
2827
import fastifyWebsocket from '@fastify/websocket';
@@ -54,13 +53,13 @@ const warn = log('websockets:warn');
5453
const debug = log('websockets:debug');
5554
const trace = log('websockets:trace');
5655

57-
const emitter = new EventEmitter();
56+
const emitter = new EventEmitter<string, { change: Change }>();
5857

5958
class Watch {
6059
/**
61-
* @description Maps requestId to path_leftover
60+
* @description Set of `requestId`s
6261
*/
63-
readonly requests = new Map<string, string>();
62+
readonly requests = new Set<string>();
6463
readonly resourceId;
6564

6665
readonly #send: (value: unknown) => Promise<void>;
@@ -82,9 +81,9 @@ class Watch {
8281
emitter.removeListener(this.resourceId, this.#cb);
8382
}
8483

85-
async sendChange(resp: SocketChange) {
86-
trace(resp, 'Sending change');
87-
await this.#send(JSON.stringify(resp));
84+
async sendChange(change: SocketChange) {
85+
trace({ change }, 'Sending change');
86+
await this.#send(JSON.stringify(change));
8887
}
8988

9089
#handler({ change }: { change: Change }): void {
@@ -94,30 +93,10 @@ class Watch {
9493
const message = {
9594
resourceId,
9695
change,
97-
requestId: [] as string[],
98-
path_leftover: [] as string[],
96+
requestId: Array.from(requests) as [string, ...string[]],
9997
};
100-
for (const [requestId, pathLeftover] of requests) {
101-
// Find requests with changes
102-
const pathChange: unknown = JsonPointer.get(
103-
change?.[0]?.body ?? {},
104-
pathLeftover
105-
);
106-
if (pathChange === undefined) {
107-
// No relevant change
108-
continue;
109-
}
110-
111-
message.requestId.push(requestId);
112-
message.path_leftover.push(pathLeftover);
113-
}
114-
115-
if (message.requestId.length <= 0) {
116-
// No-one to notify?
117-
return;
118-
}
11998

120-
void this.sendChange(message as SocketChange);
99+
void this.sendChange(message);
121100
}
122101
}
123102

@@ -382,7 +361,7 @@ const plugin: FastifyPluginAsync = async (fastify) => {
382361

383362
const watch =
384363
watches.get(resourceId) ?? new Watch(resourceId, socket);
385-
watch.requests.set(message.requestId, pathLeftover);
364+
watch.requests.add(message.requestId);
386365
watches.set(resourceId, watch);
387366

388367
// Emit all new changes from the given rev in the request
@@ -429,7 +408,6 @@ const plugin: FastifyPluginAsync = async (fastify) => {
429408
await watch.sendChange({
430409
requestId: [message.requestId],
431410
resourceId,
432-
path_leftover: [pathLeftover],
433411
change,
434412
});
435413
}
@@ -517,14 +495,18 @@ const plugin: FastifyPluginAsync = async (fastify) => {
517495
return;
518496
}
519497

498+
if (emitter.listeners(request.resource_id).length === 0) {
499+
// No WATCHes
500+
return;
501+
}
502+
520503
try {
521504
const change = await changes.getChangeArray(
522505
request.resource_id,
523506
request._rev
524507
);
525508
trace({ change }, `Emitted change for ${request.resource_id}`);
526509
emitter.emit(request.resource_id, {
527-
path_leftover: request.path_leftover,
528510
change,
529511
});
530512
if (change?.[0]?.type === 'delete') {

‎oada/services/permissions-handler/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@
3333
},
3434
"devDependencies": {
3535
"@types/debug": "^4.1.7",
36-
"@types/node": "^18.6.1",
36+
"@types/node": "^18.6.2",
3737
"@types/type-is": "^1.6.3",
3838
"typescript": "^4.7.4"
3939
},

‎oada/services/rev-graph-update/package.json‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,10 +44,10 @@
4444
},
4545
"devDependencies": {
4646
"@ava/typescript": "^3.0.1",
47-
"@oada/client": "^4.1.2",
47+
"@oada/client": "^4.1.3",
4848
"@oada/write-handler": "^3.5.1",
4949
"@types/debug": "^4.1.7",
50-
"@types/node": "^18.6.1",
50+
"@types/node": "^18.6.2",
5151
"@types/randomstring": "^1.1.8",
5252
"ava": "4.0.0-rc.1",
5353
"type-fest": "^2.17.0",

0 commit comments

Comments
 (0)