Merge pull request #1120 from neet/refactor-ws

fix: Refactor WebSocketSubscription and remove unused code

src/adapters/ws/web-socket-subscription.ts

@@ -26,8 +26,6 @@ export class WebSocketSubscription implements mastodon.streaming.Subscription {
     while (this.connector.canAcquire()) {
       this.connection = await this.connector.acquire();
 
-      const messages = toAsyncIterable(this.connection);
-
       const data = this.serializer.serialize("json", {
         type: "subscribe",
         stream: this.stream,
@@ -37,8 +35,15 @@ export class WebSocketSubscription implements mastodon.streaming.Subscription {
       this.logger?.log("debug", "↑ WEBSOCKET", data);
       this.connection.send(data);
 
-      for await (const event of this.transformIntoEvents(messages)) {
-        if (!this.matches(event)) continue;
+      const messages = toAsyncIterable(this.connection);
+
+      for await (const message of messages) {
+        const event = await this.parseMessage(message.data as string);
+
+        if (!this.test(event)) {
+          continue;
+        }
+
         this.logger?.log("debug", "↓ WEBSOCKET", event);
         yield event;
       }
@@ -70,7 +75,7 @@ export class WebSocketSubscription implements mastodon.streaming.Subscription {
     this.unsubscribe();
   }
 
-  private matches(event: mastodon.streaming.Event): boolean {
+  private test(event: mastodon.streaming.Event): boolean {
     // subscribe("hashtag", { tag: "foo" }) -> ["hashtag", "foo"]
     // subscribe("list", { list: "foo" })   -> ["list", "foo"]
     const params = this.params ?? {};
@@ -79,15 +84,6 @@ export class WebSocketSubscription implements mastodon.streaming.Subscription {
     return stream.every((s) => event.stream.includes(s));
   }
 
-  private async *transformIntoEvents(
-    messages: AsyncIterable<WebSocket.MessageEvent>,
-  ) {
-    for await (const message of messages) {
-      const event = await this.parseMessage(message.data as string);
-      yield event;
-    }
-  }
-
   private async parseMessage(
     rawEvent: string,
   ): Promise<mastodon.streaming.Event> {

src/utils/exponential-backoff.ts

@@ -25,7 +25,7 @@ export class ExponentialBackoff {
       throw new ExponentialBackoffError(this.attempts);
     }
 
-    await sleep(this.getTimeout());
+    await sleep(this.timeout);
     this.attempts++;
   }
 
@@ -45,7 +45,7 @@ export class ExponentialBackoff {
     return this.props.maxAttempts ?? Number.POSITIVE_INFINITY;
   }
 
-  getTimeout(): number {
+  private get timeout(): number {
     return this.factor * this.base ** this.attempts;
   }
 }

src/utils/noop.ts

@@ -1,7 +1,3 @@
 export function noop(): void {
   //
 }
-
-export async function noopAsync(): Promise<void> {
-  //
-}