vielle.dev
forked from
pds.ls/pdsls
atproto explorer
cleanup stream component
handle.invalid
+10
-20
src/views/stream.tsx
+10
-20
src/views/stream.tsx
···
8
9
const LIMIT = 25;
10
type Parameter = { name: string; param: string | string[] | undefined };
11
-
enum StreamType {
12
-
JETSTREAM,
13
-
FIREHOSE,
14
-
}
15
16
const StreamView = () => {
17
const [searchParams, setSearchParams] = useSearchParams();
18
const [parameters, setParameters] = createSignal<Parameter[]>([]);
19
-
const streamType =
20
-
useLocation().pathname === "/firehose" ? StreamType.FIREHOSE : StreamType.JETSTREAM;
21
-
22
const [records, setRecords] = createSignal<Array<any>>([]);
23
const [connected, setConnected] = createSignal(false);
24
-
const [allEvents, setAllEvents] = createSignal(false);
25
const [notice, setNotice] = createSignal("");
26
let socket: WebSocket;
27
let firehose: Firehose;
···
30
const connectSocket = async (formData: FormData) => {
31
setNotice("");
32
if (connected()) {
33
-
if (streamType === StreamType.JETSTREAM) socket?.close();
34
else firehose?.close();
35
setConnected(false);
36
return;
···
38
setRecords([]);
39
40
let url = "";
41
-
if (streamType === StreamType.JETSTREAM) {
42
url =
43
formData.get("instance")?.toString() ?? "wss://jetstream1.us-east.bsky.network/subscribe";
44
url = url.concat("?");
···
57
});
58
59
const cursor = formData.get("cursor")?.toString();
60
-
if (streamType === StreamType.JETSTREAM) {
61
if (cursor?.length) url = url.concat(`cursor=${cursor}`);
62
if (url.endsWith("&")) url = url.slice(0, -1);
63
}
64
65
-
if (searchParams.allEvents === "on") setAllEvents(true);
66
-
67
setSearchParams({
68
instance: formData.get("instance")?.toString(),
69
collections: formData.get("collections")?.toString(),
···
81
]);
82
83
setConnected(true);
84
-
if (streamType === StreamType.JETSTREAM) {
85
socket = new WebSocket(url);
86
socket.addEventListener("message", (event) => {
87
const rec = JSON.parse(event.data);
88
-
if (allEvents() || (rec.kind !== "account" && rec.kind !== "identity"))
89
setRecords(records().concat(rec).slice(-LIMIT));
90
});
91
socket.addEventListener("error", () => {
···
177
name="instance"
178
value={
179
searchParams.instance ??
180
-
(streamType === StreamType.JETSTREAM ?
181
"wss://jetstream1.us-east.bsky.network/subscribe"
182
: "wss://bsky.network")
183
}
184
class="grow"
185
/>
186
</label>
187
-
<Show when={streamType === StreamType.JETSTREAM}>
188
<label class="flex items-center justify-end gap-x-1">
189
<span class="min-w-[5rem]">Collections</span>
190
<textarea
···
196
/>
197
</label>
198
</Show>
199
-
<Show when={streamType === StreamType.JETSTREAM}>
200
<label class="flex items-center justify-end gap-x-1">
201
<span class="min-w-[5rem]">DIDs</span>
202
<textarea
···
217
class="grow"
218
/>
219
</label>
220
-
<Show when={streamType === StreamType.JETSTREAM}>
221
<div class="flex items-center justify-end gap-x-1">
222
<input
223
type="checkbox"
224
name="allEvents"
225
id="allEvents"
226
checked={searchParams.allEvents === "on" ? true : false}
227
-
onChange={(e) => setAllEvents(e.currentTarget.checked)}
228
/>
229
<label for="allEvents" class="select-none">
230
Show account and identity events
···
8
9
const LIMIT = 25;
10
type Parameter = { name: string; param: string | string[] | undefined };
11
12
const StreamView = () => {
13
const [searchParams, setSearchParams] = useSearchParams();
14
const [parameters, setParameters] = createSignal<Parameter[]>([]);
15
+
const streamType = useLocation().pathname === "/firehose" ? "firehose" : "jetstream";
16
const [records, setRecords] = createSignal<Array<any>>([]);
17
const [connected, setConnected] = createSignal(false);
18
const [notice, setNotice] = createSignal("");
19
let socket: WebSocket;
20
let firehose: Firehose;
···
23
const connectSocket = async (formData: FormData) => {
24
setNotice("");
25
if (connected()) {
26
+
if (streamType === "jetstream") socket?.close();
27
else firehose?.close();
28
setConnected(false);
29
return;
···
31
setRecords([]);
32
33
let url = "";
34
+
if (streamType === "jetstream") {
35
url =
36
formData.get("instance")?.toString() ?? "wss://jetstream1.us-east.bsky.network/subscribe";
37
url = url.concat("?");
···
50
});
51
52
const cursor = formData.get("cursor")?.toString();
53
+
if (streamType === "jetstream") {
54
if (cursor?.length) url = url.concat(`cursor=${cursor}`);
55
if (url.endsWith("&")) url = url.slice(0, -1);
56
}
57
58
setSearchParams({
59
instance: formData.get("instance")?.toString(),
60
collections: formData.get("collections")?.toString(),
···
72
]);
73
74
setConnected(true);
75
+
if (streamType === "jetstream") {
76
socket = new WebSocket(url);
77
socket.addEventListener("message", (event) => {
78
const rec = JSON.parse(event.data);
79
+
if (searchParams.allEvents === "on" || (rec.kind !== "account" && rec.kind !== "identity"))
80
setRecords(records().concat(rec).slice(-LIMIT));
81
});
82
socket.addEventListener("error", () => {
···
168
name="instance"
169
value={
170
searchParams.instance ??
171
+
(streamType === "jetstream" ?
172
"wss://jetstream1.us-east.bsky.network/subscribe"
173
: "wss://bsky.network")
174
}
175
class="grow"
176
/>
177
</label>
178
+
<Show when={streamType === "jetstream"}>
179
<label class="flex items-center justify-end gap-x-1">
180
<span class="min-w-[5rem]">Collections</span>
181
<textarea
···
187
/>
188
</label>
189
</Show>
190
+
<Show when={streamType === "jetstream"}>
191
<label class="flex items-center justify-end gap-x-1">
192
<span class="min-w-[5rem]">DIDs</span>
193
<textarea
···
208
class="grow"
209
/>
210
</label>
211
+
<Show when={streamType === "jetstream"}>
212
<div class="flex items-center justify-end gap-x-1">
213
<input
214
type="checkbox"
215
name="allEvents"
216
id="allEvents"
217
checked={searchParams.allEvents === "on" ? true : false}
218
/>
219
<label for="allEvents" class="select-none">
220
Show account and identity events