a collection of lightweight TypeScript packages for AT Protocol, the protocol powering Bluesky
atproto
bluesky
typescript
npm
refactor(xrpc-server): move adapter agnostic subscription tests
+522
-382
+1
-186
packages/servers/xrpc-server-bun/lib/index.test.ts
+1
-186
packages/servers/xrpc-server-bun/lib/index.test.ts
···
2
2
3
3
import { ComAtprotoLabelDefs, ComAtprotoLabelSubscribeLabels } from '@atcute/atproto';
4
4
import { decode, decodeFirst } from '@atcute/cbor';
5
-
import { XRPCRouter, XRPCSubscriptionError } from '@atcute/xrpc-server';
5
+
import { XRPCRouter } from '@atcute/xrpc-server';
6
6
7
7
import { createBunWebSocket } from './index.js';
8
8
···
80
80
client.close();
81
81
});
82
82
83
-
it('handles subscription with params', async () => {
84
-
const ws = createBunWebSocket();
85
-
const router = new XRPCRouter({ websocket: ws.adapter });
86
-
87
-
let receivedCursor: number | undefined;
88
-
89
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
90
-
async *handler({ params }) {
91
-
receivedCursor = params.cursor;
92
-
yield {
93
-
$type: 'com.atproto.label.subscribeLabels#labels',
94
-
seq: params.cursor ?? 0,
95
-
labels: [],
96
-
};
97
-
},
98
-
});
99
-
100
-
using server = Bun.serve({
101
-
...ws.wrap(router),
102
-
port: 0,
103
-
});
104
-
105
-
const client = new WebSocket(
106
-
`ws://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels?cursor=42`,
107
-
);
108
-
109
-
await new Promise<void>((resolve, reject) => {
110
-
client.onmessage = () => {
111
-
resolve();
112
-
};
113
-
client.onerror = () => {
114
-
reject(new Error('WebSocket error'));
115
-
};
116
-
});
117
-
118
-
expect(receivedCursor).toBe(42);
119
-
client.close();
120
-
});
121
-
122
-
it('handles multiple messages from subscription', async () => {
123
-
const ws = createBunWebSocket();
124
-
const router = new XRPCRouter({ websocket: ws.adapter });
125
-
126
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
127
-
async *handler() {
128
-
yield {
129
-
$type: 'com.atproto.label.subscribeLabels#labels',
130
-
seq: 1,
131
-
labels: [],
132
-
};
133
-
yield {
134
-
$type: 'com.atproto.label.subscribeLabels#labels',
135
-
seq: 2,
136
-
labels: [],
137
-
};
138
-
yield {
139
-
$type: 'com.atproto.label.subscribeLabels#labels',
140
-
seq: 3,
141
-
labels: [],
142
-
};
143
-
},
144
-
});
145
-
146
-
using server = Bun.serve({
147
-
...ws.wrap(router),
148
-
port: 0,
149
-
});
150
-
151
-
const client = new WebSocket(`ws://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels`);
152
-
153
-
const frames: any[] = [];
154
-
await new Promise<void>((resolve, reject) => {
155
-
client.onmessage = (event) => {
156
-
const buffer = event.data;
157
-
const uint8 = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
158
-
159
-
const { header, body } = decodeFrame(uint8);
160
-
161
-
frames.push({ header, body });
162
-
163
-
if (frames.length === 3) {
164
-
resolve();
165
-
}
166
-
};
167
-
client.onerror = () => {
168
-
reject(new Error('WebSocket error'));
169
-
};
170
-
});
171
-
172
-
expect(frames).toEqual([
173
-
{ header: { op: 1, t: '#labels' }, body: { labels: [], seq: 1 } },
174
-
{ header: { op: 1, t: '#labels' }, body: { labels: [], seq: 2 } },
175
-
{ header: { op: 1, t: '#labels' }, body: { labels: [], seq: 3 } },
176
-
]);
177
-
178
-
client.close();
179
-
});
180
-
181
83
it('stops sending when client disconnects', async () => {
182
84
const ws = createBunWebSocket();
183
85
const router = new XRPCRouter({ websocket: ws.adapter });
···
240
142
client.close();
241
143
});
242
144
243
-
it('sends error frame on XRPCSubscriptionError', async () => {
244
-
const ws = createBunWebSocket();
245
-
const router = new XRPCRouter({ websocket: ws.adapter });
246
-
247
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
248
-
async *handler() {
249
-
yield {
250
-
$type: 'com.atproto.label.subscribeLabels#labels',
251
-
seq: 1,
252
-
labels: [],
253
-
};
254
-
255
-
throw new XRPCSubscriptionError({
256
-
error: 'FutureCursor',
257
-
description: `Cursor is in the future`,
258
-
});
259
-
},
260
-
});
261
-
262
-
using server = Bun.serve({
263
-
...ws.wrap(router),
264
-
port: 0,
265
-
});
266
-
267
-
const client = new WebSocket(`ws://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels`);
268
-
269
-
const frames: any[] = [];
270
-
await new Promise<void>((resolve, reject) => {
271
-
client.onmessage = (event) => {
272
-
const buffer = event.data;
273
-
const uint8 = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
274
-
275
-
const { header, body } = decodeFrame(uint8);
276
-
277
-
frames.push({ header, body });
278
-
resolve();
279
-
};
280
-
client.onerror = () => {
281
-
reject(new Error('WebSocket error'));
282
-
};
283
-
});
284
-
285
-
expect(frames).toEqual([
286
-
{
287
-
header: { op: 1, t: '#labels' },
288
-
body: {
289
-
labels: [],
290
-
seq: 1,
291
-
},
292
-
},
293
-
{
294
-
header: { op: -1 },
295
-
body: {
296
-
error: 'FutureCursor',
297
-
message: 'Cursor is in the future',
298
-
},
299
-
},
300
-
]);
301
-
302
-
client.close();
303
-
});
304
-
305
-
it('rejects non-WebSocket upgrade requests', async () => {
306
-
const ws = createBunWebSocket();
307
-
const router = new XRPCRouter({ websocket: ws.adapter });
308
-
309
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
310
-
async *handler() {
311
-
yield {
312
-
$type: 'com.atproto.label.subscribeLabels#labels',
313
-
seq: 1,
314
-
labels: [],
315
-
};
316
-
},
317
-
});
318
-
319
-
using server = Bun.serve({
320
-
...ws.wrap(router),
321
-
port: 0,
322
-
});
323
-
324
-
const response = await fetch(`http://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels`);
325
-
const error = await response.json();
326
-
327
-
expect(response.status).toBe(400);
328
-
expect(error).toEqual(expect.objectContaining({ error: 'InvalidRequest' }));
329
-
});
330
145
});
+1
-176
packages/servers/xrpc-server-node/lib/index.test.ts
+1
-176
packages/servers/xrpc-server-node/lib/index.test.ts
···
4
4
5
5
import { ComAtprotoLabelDefs, ComAtprotoLabelSubscribeLabels } from '@atcute/atproto';
6
6
import { decode, decodeFirst } from '@atcute/cbor';
7
-
import { XRPCRouter, XRPCSubscriptionError } from '@atcute/xrpc-server';
7
+
import { XRPCRouter } from '@atcute/xrpc-server';
8
8
9
9
import { createNodeWebSocket, type NodeWebSocket } from './index.js';
10
10
···
101
101
client.close();
102
102
});
103
103
104
-
it('handles subscription with params', async () => {
105
-
const ws = createNodeWebSocket();
106
-
const router = new XRPCRouter({ websocket: ws.adapter });
107
-
108
-
let receivedCursor: number | undefined;
109
-
110
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
111
-
async *handler({ params }) {
112
-
receivedCursor = params.cursor;
113
-
yield {
114
-
$type: 'com.atproto.label.subscribeLabels#labels',
115
-
seq: params.cursor ?? 0,
116
-
labels: [],
117
-
};
118
-
},
119
-
});
120
-
121
-
using server = await createHttpServer(router, ws);
122
-
123
-
const client = new WebSocket(
124
-
`ws://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels?cursor=42`,
125
-
);
126
-
client.binaryType = 'arraybuffer';
127
-
128
-
await new Promise<void>((resolve, reject) => {
129
-
client.onmessage = () => {
130
-
resolve();
131
-
};
132
-
client.onerror = () => {
133
-
reject(new Error('WebSocket error'));
134
-
};
135
-
});
136
-
137
-
expect(receivedCursor).toBe(42);
138
-
client.close();
139
-
});
140
-
141
-
it('handles multiple messages from subscription', async () => {
142
-
const ws = createNodeWebSocket();
143
-
const router = new XRPCRouter({ websocket: ws.adapter });
144
-
145
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
146
-
async *handler() {
147
-
yield {
148
-
$type: 'com.atproto.label.subscribeLabels#labels',
149
-
seq: 1,
150
-
labels: [],
151
-
};
152
-
yield {
153
-
$type: 'com.atproto.label.subscribeLabels#labels',
154
-
seq: 2,
155
-
labels: [],
156
-
};
157
-
yield {
158
-
$type: 'com.atproto.label.subscribeLabels#labels',
159
-
seq: 3,
160
-
labels: [],
161
-
};
162
-
},
163
-
});
164
-
165
-
using server = await createHttpServer(router, ws);
166
-
167
-
const client = new WebSocket(`ws://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels`);
168
-
client.binaryType = 'arraybuffer';
169
-
170
-
const frames: any[] = [];
171
-
await new Promise<void>((resolve, reject) => {
172
-
client.onmessage = (event) => {
173
-
const buffer = event.data;
174
-
const uint8 = new Uint8Array(buffer);
175
-
176
-
const { header, body } = decodeFrame(uint8);
177
-
178
-
frames.push({ header, body });
179
-
180
-
if (frames.length === 3) {
181
-
resolve();
182
-
}
183
-
};
184
-
client.onerror = () => {
185
-
reject(new Error('WebSocket error'));
186
-
};
187
-
});
188
-
189
-
expect(frames).toEqual([
190
-
{ header: { op: 1, t: '#labels' }, body: { labels: [], seq: 1 } },
191
-
{ header: { op: 1, t: '#labels' }, body: { labels: [], seq: 2 } },
192
-
{ header: { op: 1, t: '#labels' }, body: { labels: [], seq: 3 } },
193
-
]);
194
-
195
-
client.close();
196
-
});
197
104
198
105
it('stops sending when client disconnects', async () => {
199
106
const ws = createNodeWebSocket();
···
255
162
client.close();
256
163
});
257
164
258
-
it('sends error frame on XRPCSubscriptionError', async () => {
259
-
const ws = createNodeWebSocket();
260
-
const router = new XRPCRouter({ websocket: ws.adapter });
261
-
262
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
263
-
async *handler() {
264
-
yield {
265
-
$type: 'com.atproto.label.subscribeLabels#labels',
266
-
seq: 1,
267
-
labels: [],
268
-
};
269
-
270
-
throw new XRPCSubscriptionError({
271
-
error: 'FutureCursor',
272
-
description: `Cursor is in the future`,
273
-
});
274
-
},
275
-
});
276
-
277
-
using server = await createHttpServer(router, ws);
278
-
279
-
const client = new WebSocket(`ws://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels`);
280
-
client.binaryType = 'arraybuffer';
281
-
282
-
const frames: any[] = [];
283
-
await new Promise<void>((resolve, reject) => {
284
-
client.onmessage = (event) => {
285
-
const buffer = event.data;
286
-
const uint8 = new Uint8Array(buffer);
287
-
288
-
const { header, body } = decodeFrame(uint8);
289
-
290
-
frames.push({ header, body });
291
-
resolve();
292
-
};
293
-
client.onerror = () => {
294
-
reject(new Error('WebSocket error'));
295
-
};
296
-
});
297
-
298
-
expect(frames).toEqual([
299
-
{
300
-
header: { op: 1, t: '#labels' },
301
-
body: {
302
-
labels: [],
303
-
seq: 1,
304
-
},
305
-
},
306
-
{
307
-
header: { op: -1 },
308
-
body: {
309
-
error: 'FutureCursor',
310
-
message: 'Cursor is in the future',
311
-
},
312
-
},
313
-
]);
314
-
315
-
client.close();
316
-
});
317
-
318
-
it('rejects non-WebSocket upgrade requests', async () => {
319
-
const ws = createNodeWebSocket();
320
-
const router = new XRPCRouter({ websocket: ws.adapter });
321
-
322
-
router.addSubscription(ComAtprotoLabelSubscribeLabels.mainSchema, {
323
-
async *handler() {
324
-
yield {
325
-
$type: 'com.atproto.label.subscribeLabels#labels',
326
-
seq: 1,
327
-
labels: [],
328
-
};
329
-
},
330
-
});
331
-
332
-
using server = await createHttpServer(router, ws);
333
-
334
-
const response = await fetch(`http://localhost:${server.port}/xrpc/com.atproto.label.subscribeLabels`);
335
-
const error = await response.json();
336
-
337
-
expect(response.status).toBe(400);
338
-
expect(error).toEqual(expect.objectContaining({ error: 'InvalidRequest' }));
339
-
});
340
165
});
+277
-17
packages/servers/xrpc-server/lib/main/router.test.ts
+277
-17
packages/servers/xrpc-server/lib/main/router.test.ts
···
1
-
import { describe, expect, it, vi } from 'vitest';
1
+
import { describe, expect, it, vi, type AsymmetricMatchersContaining } from 'vitest';
2
2
3
+
import { decode, decodeFirst } from '@atcute/cbor';
3
4
import * as v from '@atcute/lexicons/validations';
4
5
5
6
import { json } from './response.js';
6
7
import { defaultNotFoundHandler, XRPCRouter } from './router.js';
7
-
import type { WebSocketAdapter } from './types/websocket.js';
8
-
import { InvalidRequestError } from './xrpc-error.js';
8
+
import { MockWebSocketAdapter } from './utils/websocket-mock.js';
9
+
import { InvalidRequestError, XRPCSubscriptionError } from './xrpc-error.js';
9
10
10
11
describe('XRPCRouter', () => {
11
12
describe('routing', () => {
···
861
862
});
862
863
});
863
864
864
-
// we won't be testing actual subscriptions here
865
865
describe('subscription', () => {
866
-
const noopAdapter: WebSocketAdapter = {
867
-
async upgrade(_request, _handler) {
868
-
return undefined;
869
-
},
870
-
};
871
-
872
-
it('handles defining subscriptions', () => {
866
+
it('handles subscriptions', async () => {
873
867
const subscriptionSchema = v.subscription('com.example.subscription', {
874
868
params: null,
875
869
message: v.object({ random: v.integer() }),
876
870
});
877
871
878
-
const router = new XRPCRouter({ websocket: noopAdapter });
872
+
const adapter = new MockWebSocketAdapter();
873
+
const router = new XRPCRouter({ websocket: adapter });
879
874
880
875
router.addSubscription(subscriptionSchema, {
881
876
async *handler() {
882
877
yield { random: 123 };
883
878
},
884
879
});
880
+
881
+
const mock = adapter.attach(router);
882
+
using client = await mock.subscribe(`/xrpc/com.example.subscription`);
883
+
884
+
let frames: Uint8Array[] = [];
885
+
await new Promise<void>((resolve) => {
886
+
client.events.on('message', (data) => {
887
+
frames.push(data);
888
+
});
889
+
890
+
client.events.on('close', () => {
891
+
resolve();
892
+
});
893
+
});
894
+
895
+
expect(decodeFrames(frames)).toEqual([{ header: { op: 1 }, body: { random: 123 } }]);
885
896
});
886
897
887
-
it('handles defining subscriptions with variants', () => {
898
+
it('handles subscription with variants', async () => {
888
899
const subscriptionSchema = v.subscription('com.example.subscription', {
889
900
params: null,
890
901
message: v.variant([
891
-
v.object({ $type: v.literal('foo'), foo: v.string() }),
892
-
v.object({ $type: v.literal('bar'), bar: v.integer() }),
902
+
v.object({ $type: v.literal('com.example.subscription#foo'), foo: v.string() }),
903
+
v.object({ $type: v.literal('com.example.subscription#bar'), bar: v.integer() }),
893
904
]),
894
905
});
895
906
896
-
const router = new XRPCRouter({ websocket: noopAdapter });
907
+
const adapter = new MockWebSocketAdapter();
908
+
const router = new XRPCRouter({ websocket: adapter });
909
+
910
+
router.addSubscription(subscriptionSchema, {
911
+
async *handler() {
912
+
yield { $type: 'com.example.subscription#foo', foo: 'foo' };
913
+
yield { $type: 'com.example.subscription#bar', bar: 123 };
914
+
},
915
+
});
916
+
917
+
const mock = adapter.attach(router);
918
+
using client = await mock.subscribe(`/xrpc/com.example.subscription`);
919
+
920
+
let frames: Uint8Array[] = [];
921
+
await new Promise<void>((resolve) => {
922
+
client.events.on('message', (data) => {
923
+
frames.push(data);
924
+
});
925
+
926
+
client.events.on('close', () => {
927
+
resolve();
928
+
});
929
+
});
930
+
931
+
expect(decodeFrames(frames)).toEqual([
932
+
{ header: { op: 1, t: '#foo' }, body: { foo: 'foo' } },
933
+
{ header: { op: 1, t: '#bar' }, body: { bar: 123 } },
934
+
]);
935
+
});
936
+
937
+
it('handles subscriptions with params', async () => {
938
+
const subscriptionSchema = v.subscription('com.example.subscription', {
939
+
params: v.object({ cursor: v.optional(v.integer()) }),
940
+
message: v.object({ seq: v.integer() }),
941
+
});
942
+
943
+
const adapter = new MockWebSocketAdapter();
944
+
const router = new XRPCRouter({ websocket: adapter });
945
+
946
+
let receivedCursor: number | undefined;
947
+
948
+
router.addSubscription(subscriptionSchema, {
949
+
async *handler({ params }) {
950
+
receivedCursor = params.cursor;
951
+
yield { seq: params.cursor ?? 0 };
952
+
},
953
+
});
954
+
955
+
const mock = adapter.attach(router);
956
+
using client = await mock.subscribe(`/xrpc/com.example.subscription?cursor=42`);
957
+
958
+
await new Promise<void>((resolve) => {
959
+
client.events.on('close', () => resolve());
960
+
});
961
+
962
+
expect(receivedCursor).toBe(42);
963
+
});
964
+
965
+
it('handles multiple subscription messages', async () => {
966
+
const subscriptionSchema = v.subscription('com.example.subscription', {
967
+
params: null,
968
+
message: v.object({ seq: v.integer() }),
969
+
});
970
+
971
+
const adapter = new MockWebSocketAdapter();
972
+
const router = new XRPCRouter({ websocket: adapter });
973
+
974
+
router.addSubscription(subscriptionSchema, {
975
+
async *handler() {
976
+
yield { seq: 1 };
977
+
yield { seq: 2 };
978
+
yield { seq: 3 };
979
+
},
980
+
});
981
+
982
+
const mock = adapter.attach(router);
983
+
using client = await mock.subscribe(`/xrpc/com.example.subscription`);
984
+
985
+
const frames: Uint8Array[] = [];
986
+
await new Promise<void>((resolve) => {
987
+
client.events.on('message', (data) => {
988
+
frames.push(data);
989
+
});
990
+
991
+
client.events.on('close', () => {
992
+
resolve();
993
+
});
994
+
});
995
+
996
+
expect(decodeFrames(frames)).toEqual([
997
+
{ header: { op: 1 }, body: { seq: 1 } },
998
+
{ header: { op: 1 }, body: { seq: 2 } },
999
+
{ header: { op: 1 }, body: { seq: 3 } },
1000
+
]);
1001
+
});
1002
+
1003
+
it('stops sending when client disconnects', async () => {
1004
+
const subscriptionSchema = v.subscription('com.example.subscription', {
1005
+
params: null,
1006
+
message: v.object({ seq: v.integer() }),
1007
+
});
1008
+
1009
+
const adapter = new MockWebSocketAdapter();
1010
+
const router = new XRPCRouter({ websocket: adapter });
1011
+
1012
+
const { promise: aborted, resolve } = Promise.withResolvers<void>();
1013
+
let messageCount = 0;
1014
+
1015
+
router.addSubscription(subscriptionSchema, {
1016
+
async *handler({ signal }) {
1017
+
while (!signal.aborted) {
1018
+
yield { seq: messageCount++ };
1019
+
}
1020
+
1021
+
resolve();
1022
+
},
1023
+
});
1024
+
1025
+
const mock = adapter.attach(router);
1026
+
using client = await mock.subscribe(`/xrpc/com.example.subscription`);
1027
+
1028
+
const frames: Uint8Array[] = [];
1029
+
await new Promise<void>((resolve) => {
1030
+
client.events.on('message', (data) => {
1031
+
frames.push(data);
1032
+
1033
+
if (frames.length === 2) {
1034
+
client.dispose();
1035
+
}
1036
+
});
1037
+
1038
+
client.events.on('close', () => {
1039
+
resolve();
1040
+
});
1041
+
});
1042
+
1043
+
expect(decodeFrames(frames)).toEqual([
1044
+
{ header: { op: 1 }, body: { seq: 0 } },
1045
+
{ header: { op: 1 }, body: { seq: 1 } },
1046
+
]);
1047
+
1048
+
await aborted;
1049
+
});
1050
+
1051
+
it('sends error frame on XRPCSubscriptionError', async () => {
1052
+
const subscriptionSchema = v.subscription('com.example.subscription', {
1053
+
params: null,
1054
+
message: v.object({ seq: v.integer() }),
1055
+
});
1056
+
1057
+
const adapter = new MockWebSocketAdapter();
1058
+
const router = new XRPCRouter({ websocket: adapter });
1059
+
1060
+
router.addSubscription(subscriptionSchema, {
1061
+
async *handler() {
1062
+
yield { seq: 1 };
1063
+
1064
+
throw new XRPCSubscriptionError({
1065
+
error: 'FutureCursor',
1066
+
description: 'Cursor is in the future',
1067
+
});
1068
+
},
1069
+
});
1070
+
1071
+
const mock = adapter.attach(router);
1072
+
using client = await mock.subscribe(`/xrpc/com.example.subscription`);
1073
+
1074
+
const frames: Uint8Array[] = [];
1075
+
await new Promise<void>((resolve) => {
1076
+
client.events.on('message', (data) => {
1077
+
frames.push(data);
1078
+
});
1079
+
1080
+
client.events.on('close', () => {
1081
+
resolve();
1082
+
});
1083
+
});
1084
+
1085
+
expect(decodeFrames(frames)).toEqual([
1086
+
{ header: { op: 1 }, body: { seq: 1 } },
1087
+
{ header: { op: -1 }, body: { error: 'FutureCursor', message: 'Cursor is in the future' } },
1088
+
]);
1089
+
});
1090
+
1091
+
it('rejects non-WebSocket upgrade requests', async () => {
1092
+
const subscriptionSchema = v.subscription('com.example.subscription', {
1093
+
params: null,
1094
+
message: v.object({ seq: v.integer() }),
1095
+
});
1096
+
1097
+
const adapter = new MockWebSocketAdapter();
1098
+
const router = new XRPCRouter({ websocket: adapter });
897
1099
898
1100
router.addSubscription(subscriptionSchema, {
899
1101
async *handler() {
900
-
yield { $type: 'foo', foo: '123' };
1102
+
yield { seq: 1 };
1103
+
},
1104
+
});
1105
+
1106
+
const request = new Request('http://localhost/xrpc/com.example.subscription');
1107
+
const response = await router.fetch(request);
1108
+
const body = await response.json();
1109
+
1110
+
expect(response.status).toBe(400);
1111
+
expect(body).toEqual(expect.objectContaining({ error: 'InvalidRequest' }));
1112
+
});
1113
+
1114
+
it('invokes handleSubscriptionException for unexpected errors', async () => {
1115
+
const subscriptionSchema = v.subscription('com.example.subscription', {
1116
+
params: null,
1117
+
message: v.object({ seq: v.integer() }),
1118
+
});
1119
+
1120
+
const adapter = new MockWebSocketAdapter();
1121
+
const handleSubscriptionException = vi.fn();
1122
+
const router = new XRPCRouter({ websocket: adapter, handleSubscriptionException });
1123
+
1124
+
router.addSubscription(subscriptionSchema, {
1125
+
async *handler() {
1126
+
throw new Error('boom');
901
1127
},
902
1128
});
1129
+
1130
+
const mock = adapter.attach(router);
1131
+
using client = await mock.subscribe(`/xrpc/com.example.subscription`);
1132
+
1133
+
await new Promise<void>((resolve) => {
1134
+
client.events.on('close', (event) => {
1135
+
expect(event).toEqual({ code: 1011, reason: 'internal server error', wasClean: true });
1136
+
resolve();
1137
+
});
1138
+
});
1139
+
1140
+
expect(handleSubscriptionException).toBeCalledTimes(1);
1141
+
1142
+
expect(handleSubscriptionException).toBeCalledWith(expect.any(Error), expect.any(Request));
1143
+
expect(handleSubscriptionException).toBeCalledWith(
1144
+
expect.objectContaining({ message: 'boom' }),
1145
+
expect.anything(),
1146
+
);
903
1147
});
904
1148
});
905
1149
});
1150
+
1151
+
interface Frame {
1152
+
header: unknown;
1153
+
body: unknown;
1154
+
}
1155
+
1156
+
function decodeFrame(frame: Uint8Array): Frame {
1157
+
const [header, remainder] = decodeFirst(frame);
1158
+
const body = decode(remainder);
1159
+
1160
+
return { header, body };
1161
+
}
1162
+
1163
+
function decodeFrames(frames: Uint8Array[]): Frame[] {
1164
+
return frames.map((frame) => decodeFrame(frame));
1165
+
}
+16
-3
packages/servers/xrpc-server/lib/main/router.ts
+16
-3
packages/servers/xrpc-server/lib/main/router.ts
···
38
38
39
39
export type NotFoundHandler = (request: Request) => Promisable<Response>;
40
40
export type ExceptionHandler = (error: unknown, request: Request) => Promisable<Response>;
41
+
export type SubscriptionExceptionHandler = (error: unknown, request: Request) => void;
41
42
42
43
export const defaultExceptionHandler: ExceptionHandler = (error: unknown) => {
43
44
if (error instanceof XRPCError) {
···
56
57
57
58
export const defaultNotFoundHandler: NotFoundHandler = () => {
58
59
return new Response('Not Found', { status: 404 });
60
+
};
61
+
62
+
export const defaultSubscriptionExceptionHandler: SubscriptionExceptionHandler = (error: unknown) => {
63
+
throw error;
59
64
};
60
65
61
66
export interface XRPCRouterOptions {
62
67
middlewares?: FetchMiddleware[];
63
68
handleNotFound?: NotFoundHandler;
64
69
handleException?: ExceptionHandler;
70
+
handleSubscriptionException?: SubscriptionExceptionHandler;
65
71
websocket?: WebSocketAdapter;
66
72
}
67
73
···
69
75
#handlers: Record<string, InternalRouteData> = {};
70
76
#handleNotFound: NotFoundHandler;
71
77
#handleException: ExceptionHandler;
78
+
#handleSubscriptionException: SubscriptionExceptionHandler;
72
79
#websocket?: WebSocketAdapter;
73
80
74
81
fetch: (request: Request) => Promise<Response>;
···
77
84
middlewares = [],
78
85
handleException = defaultExceptionHandler,
79
86
handleNotFound = defaultNotFoundHandler,
87
+
handleSubscriptionException = defaultSubscriptionExceptionHandler,
80
88
websocket,
81
89
}: XRPCRouterOptions = {}) {
82
90
const runner = createAsyncMiddlewareRunner([...middlewares, (request) => this.#dispatch(request)]);
···
84
92
this.fetch = (request) => runner(request);
85
93
this.#handleException = handleException;
86
94
this.#handleNotFound = handleNotFound;
95
+
this.#handleSubscriptionException = handleSubscriptionException;
87
96
this.#websocket = websocket;
88
97
}
89
98
···
301
310
}
302
311
303
312
const upgrade = await websocket.upgrade(request, async (ws) => {
313
+
const signal = ws.signal;
314
+
304
315
const context: UnknownSubscriptionContext = {
305
316
request: request,
306
317
params: params,
307
-
signal: ws.signal,
318
+
signal: signal,
308
319
};
309
320
310
321
try {
311
322
for await (const message of handler(context)) {
312
-
if (ws.signal.aborted) {
323
+
if (signal.aborted) {
313
324
break;
314
325
}
315
326
···
319
330
const frame = encodeMessageFrame(body, type);
320
331
await ws.send(frame);
321
332
}
333
+
334
+
ws.close(1000);
322
335
} catch (err) {
323
336
if (err instanceof XRPCSubscriptionError) {
324
337
const frame = encodeErrorFrame(err.error, err.description);
···
332
345
}
333
346
334
347
ws.close(1011, `internal server error`);
335
-
throw err;
348
+
this.#handleSubscriptionException(err, request);
336
349
}
337
350
});
338
351
+116
packages/servers/xrpc-server/lib/main/utils/event-emitter.ts
+116
packages/servers/xrpc-server/lib/main/utils/event-emitter.ts
···
1
+
/** Converts a tuple into a listener function */
2
+
export type ListenerFor<T extends any[]> = (...args: T) => void;
3
+
4
+
/** Generic record of the event name and its argument tuple */
5
+
export type EventMap = {
6
+
[key: string | symbol]: any[];
7
+
};
8
+
9
+
type MaybeArray<T> = T | T[];
10
+
11
+
/** Event emitter */
12
+
export class EventEmitter<Events extends EventMap> {
13
+
#events?: { [E in keyof Events]?: MaybeArray<ListenerFor<Events[E]>> };
14
+
15
+
/**
16
+
* Appends a listener for the specified event name
17
+
* @param name Name of the event
18
+
* @param listener Callback that should be invoked when an event is dispatched
19
+
* @returns Cleanup function that can be called to remove it
20
+
*/
21
+
on<E extends keyof Events>(name: E, listener: ListenerFor<Events[E]>): () => void {
22
+
let events = this.#events;
23
+
let existing: MaybeArray<ListenerFor<Events[E]>> | undefined;
24
+
25
+
if (events === undefined) {
26
+
events = this.#events = Object.create(null);
27
+
} else {
28
+
existing = events[name];
29
+
}
30
+
31
+
if (existing === undefined) {
32
+
events![name] = listener;
33
+
} else if (typeof existing === 'function') {
34
+
events![name] = [existing, listener];
35
+
} else {
36
+
events![name] = existing.concat(listener);
37
+
}
38
+
39
+
// @ts-expect-error: complains about `listener`
40
+
return this.off.bind(this, name, listener);
41
+
}
42
+
43
+
/**
44
+
* Remove listener from the specified event name
45
+
* @param name Name of the event
46
+
* @param listener Callback to remove
47
+
*/
48
+
off<E extends keyof Events>(name: E, listener: ListenerFor<Events[E]>): void {
49
+
const events = this.#events;
50
+
51
+
if (events === undefined) {
52
+
return;
53
+
}
54
+
55
+
const list = events[name];
56
+
57
+
if (list == undefined) {
58
+
return;
59
+
}
60
+
61
+
if (list === listener) {
62
+
delete events[name];
63
+
} else if (typeof list !== 'function') {
64
+
const index = list.indexOf(listener);
65
+
66
+
if (index !== -1) {
67
+
if (list.length === 2) {
68
+
// ^ flips the bit, it's either 0 or 1 here.
69
+
events[name] = list[index ^ 1];
70
+
} else {
71
+
events[name] = list.toSpliced(index, 1);
72
+
}
73
+
}
74
+
}
75
+
}
76
+
77
+
/**
78
+
* Emit an event with the specified name and its payload
79
+
* @param name Name of the event
80
+
* @param args Payload for the event
81
+
* @returns Whether a listener has been called
82
+
*/
83
+
emit<E extends keyof Events>(name: E, ...args: Events[E]): boolean {
84
+
const events = this.#events;
85
+
86
+
if (events === undefined) {
87
+
return false;
88
+
}
89
+
90
+
const handler = events[name];
91
+
92
+
if (handler === undefined) {
93
+
return false;
94
+
}
95
+
96
+
if (typeof handler === 'function') {
97
+
handler.apply(this, args);
98
+
} else {
99
+
for (let idx = 0, len = handler.length; idx < len; idx++) {
100
+
handler[idx].apply(this, args);
101
+
}
102
+
}
103
+
104
+
return true;
105
+
}
106
+
107
+
/**
108
+
* Determines if there is a listener on a specified event name
109
+
* @param name Name of the event
110
+
* @returns Whether there is a listener registered
111
+
*/
112
+
has(name: keyof Events): boolean {
113
+
const events = this.#events;
114
+
return events !== undefined && name in events;
115
+
}
116
+
}
+111
packages/servers/xrpc-server/lib/main/utils/websocket-mock.ts
+111
packages/servers/xrpc-server/lib/main/utils/websocket-mock.ts
···
1
+
import { AsyncLocalStorage } from 'node:async_hooks';
2
+
3
+
import type { Promisable } from '../../types/misc.js';
4
+
import type { XRPCRouter } from '../router.js';
5
+
import type { WebSocketAdapter, WebSocketConnection } from '../types/websocket.js';
6
+
import { EventEmitter } from './event-emitter.js';
7
+
8
+
interface WebSocketHandlerContext {
9
+
handler: ((ws: WebSocketConnection) => Promisable<void>) | null;
10
+
}
11
+
12
+
export interface SubscriptionClient extends Disposable {
13
+
events: EventEmitter<{
14
+
message: [data: Uint8Array];
15
+
close: [event: { code: number; reason: string; wasClean: boolean }];
16
+
}>;
17
+
dispose(): void;
18
+
}
19
+
20
+
export interface SubscriptionMock {
21
+
subscribe(url: string): Promise<SubscriptionClient>;
22
+
}
23
+
24
+
export class MockWebSocketAdapter implements WebSocketAdapter {
25
+
#context = new AsyncLocalStorage<WebSocketHandlerContext>();
26
+
27
+
upgrade(
28
+
_request: Request,
29
+
handler: (ws: WebSocketConnection) => Promisable<void>,
30
+
): Promisable<Response | undefined> {
31
+
const ctx = this.#context.getStore();
32
+
if (!ctx) {
33
+
return undefined;
34
+
}
35
+
36
+
ctx.handler = handler;
37
+
38
+
return new Response(null);
39
+
}
40
+
41
+
attach(router: XRPCRouter): SubscriptionMock {
42
+
return {
43
+
subscribe: async (url) => {
44
+
const ctx: WebSocketHandlerContext = {
45
+
handler: null,
46
+
};
47
+
48
+
await this.#context.run(ctx, async () => {
49
+
const urlp = new URL(url, 'http://localhost');
50
+
const request = new Request(urlp, {
51
+
headers: {
52
+
upgrade: 'websocket',
53
+
connection: 'upgrade',
54
+
},
55
+
});
56
+
57
+
const response = await router.fetch(request);
58
+
59
+
return response;
60
+
});
61
+
62
+
if (!ctx.handler) {
63
+
throw new Error(`WebSocket upgrade succeeded but no handler was set`);
64
+
}
65
+
66
+
const events = new EventEmitter<{
67
+
message: [data: Uint8Array];
68
+
close: [event: { code: number; reason: string; wasClean: boolean }];
69
+
}>();
70
+
71
+
const controller = new AbortController();
72
+
const signal = controller.signal;
73
+
74
+
const connection: WebSocketConnection = {
75
+
signal: signal,
76
+
send(data) {
77
+
events.emit('message', data);
78
+
},
79
+
close(code = 1000, reason = '') {
80
+
if (!signal.aborted) {
81
+
events.emit('close', { code, reason, wasClean: true });
82
+
controller.abort();
83
+
}
84
+
},
85
+
};
86
+
87
+
{
88
+
const handler = ctx.handler;
89
+
setTimeout(() => {
90
+
handler(connection);
91
+
}, 1);
92
+
}
93
+
94
+
const client: SubscriptionClient = {
95
+
events,
96
+
dispose() {
97
+
if (!signal.aborted) {
98
+
events.emit('close', { code: 1000, reason: '', wasClean: true });
99
+
controller.abort();
100
+
}
101
+
},
102
+
[Symbol.dispose]() {
103
+
this.dispose();
104
+
},
105
+
};
106
+
107
+
return client;
108
+
},
109
+
};
110
+
}
111
+
}