a collection of lightweight TypeScript packages for AT Protocol, the protocol powering Bluesky
atproto
bluesky
typescript
npm
refactor(xrpc-server): remove event emitter implementation used in tests
+33
-142
+12
-12
packages/servers/xrpc-server/lib/main/router.test.ts
+12
-12
packages/servers/xrpc-server/lib/main/router.test.ts
···
956
956
957
957
let frames: Uint8Array[] = [];
958
958
await new Promise<void>((resolve) => {
959
-
client.events.on('message', (data) => {
959
+
client.onMessage.subscribe((data) => {
960
960
frames.push(data);
961
961
});
962
962
963
-
client.events.on('close', () => {
963
+
client.onClose.subscribe(() => {
964
964
resolve();
965
965
});
966
966
});
···
992
992
993
993
let frames: Uint8Array[] = [];
994
994
await new Promise<void>((resolve) => {
995
-
client.events.on('message', (data) => {
995
+
client.onMessage.subscribe((data) => {
996
996
frames.push(data);
997
997
});
998
998
999
-
client.events.on('close', () => {
999
+
client.onClose.subscribe(() => {
1000
1000
resolve();
1001
1001
});
1002
1002
});
···
1029
1029
using client = await mock.subscribe(`/xrpc/com.example.subscription?cursor=42`);
1030
1030
1031
1031
await new Promise<void>((resolve) => {
1032
-
client.events.on('close', () => resolve());
1032
+
client.onClose.subscribe(() => resolve());
1033
1033
});
1034
1034
1035
1035
expect(receivedCursor).toBe(42);
···
1057
1057
1058
1058
const frames: Uint8Array[] = [];
1059
1059
await new Promise<void>((resolve) => {
1060
-
client.events.on('message', (data) => {
1060
+
client.onMessage.subscribe((data) => {
1061
1061
frames.push(data);
1062
1062
});
1063
1063
1064
-
client.events.on('close', () => {
1064
+
client.onClose.subscribe(() => {
1065
1065
resolve();
1066
1066
});
1067
1067
});
···
1100
1100
1101
1101
const frames: Uint8Array[] = [];
1102
1102
await new Promise<void>((resolve) => {
1103
-
client.events.on('message', (data) => {
1103
+
client.onMessage.subscribe((data) => {
1104
1104
frames.push(data);
1105
1105
1106
1106
if (frames.length === 2) {
···
1108
1108
}
1109
1109
});
1110
1110
1111
-
client.events.on('close', () => {
1111
+
client.onClose.subscribe(() => {
1112
1112
resolve();
1113
1113
});
1114
1114
});
···
1146
1146
1147
1147
const frames: Uint8Array[] = [];
1148
1148
await new Promise<void>((resolve) => {
1149
-
client.events.on('message', (data) => {
1149
+
client.onMessage.subscribe((data) => {
1150
1150
frames.push(data);
1151
1151
});
1152
1152
1153
-
client.events.on('close', () => {
1153
+
client.onClose.subscribe(() => {
1154
1154
resolve();
1155
1155
});
1156
1156
});
···
1205
1205
using client = await mock.subscribe(`/xrpc/com.example.subscription`);
1206
1206
1207
1207
await new Promise<void>((resolve) => {
1208
-
client.events.on('close', (event) => {
1208
+
client.onClose.subscribe((event) => {
1209
1209
expect(event).toEqual({ code: 1011, reason: 'internal server error', wasClean: true });
1210
1210
resolve();
1211
1211
});
-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
-
}
+17
-14
packages/servers/xrpc-server/lib/main/utils/websocket-mock.ts
+17
-14
packages/servers/xrpc-server/lib/main/utils/websocket-mock.ts
···
1
1
import { AsyncLocalStorage } from 'node:async_hooks';
2
2
3
+
import { SimpleEventEmitter } from '@mary-ext/simple-event-emitter';
4
+
3
5
import type { Promisable } from '../../types/misc.ts';
4
6
import type { XRPCRouter } from '../router.ts';
5
7
import type { WebSocketAdapter, WebSocketConnection } from '../types/websocket.ts';
6
-
7
-
import { EventEmitter } from './event-emitter.ts';
8
8
9
9
interface WebSocketHandlerContext {
10
10
handler: ((ws: WebSocketConnection) => Promisable<void>) | null;
11
11
}
12
12
13
+
export interface CloseEvent {
14
+
code: number;
15
+
reason: string;
16
+
wasClean: boolean;
17
+
}
18
+
13
19
export interface SubscriptionClient extends Disposable {
14
-
events: EventEmitter<{
15
-
message: [data: Uint8Array];
16
-
close: [event: { code: number; reason: string; wasClean: boolean }];
17
-
}>;
20
+
onMessage: SimpleEventEmitter<[data: Uint8Array]>;
21
+
onClose: SimpleEventEmitter<[event: CloseEvent]>;
18
22
dispose(): void;
19
23
}
20
24
···
64
68
throw new Error(`WebSocket upgrade succeeded but no handler was set`);
65
69
}
66
70
67
-
const events = new EventEmitter<{
68
-
message: [data: Uint8Array];
69
-
close: [event: { code: number; reason: string; wasClean: boolean }];
70
-
}>();
71
+
const onMessage = new SimpleEventEmitter<[data: Uint8Array]>();
72
+
const onClose = new SimpleEventEmitter<[event: CloseEvent]>();
71
73
72
74
const controller = new AbortController();
73
75
const signal = controller.signal;
···
75
77
const connection: WebSocketConnection = {
76
78
signal: signal,
77
79
send(data) {
78
-
events.emit('message', data);
80
+
onMessage.emit(data);
79
81
},
80
82
close(code = 1000, reason = '') {
81
83
if (!signal.aborted) {
82
-
events.emit('close', { code, reason, wasClean: true });
84
+
onClose.emit({ code, reason, wasClean: true });
83
85
controller.abort();
84
86
}
85
87
},
···
93
95
}
94
96
95
97
const client: SubscriptionClient = {
96
-
events,
98
+
onMessage,
99
+
onClose,
97
100
dispose() {
98
101
if (!signal.aborted) {
99
-
events.emit('close', { code: 1000, reason: '', wasClean: true });
102
+
onClose.emit({ code: 1000, reason: '', wasClean: true });
100
103
controller.abort();
101
104
}
102
105
},
+1
packages/servers/xrpc-server/package.json
+1
packages/servers/xrpc-server/package.json
+3
pnpm-lock.yaml
+3
pnpm-lock.yaml