Merge pull request #1040 from neet/dispose-streaming-client

feat: Implement Disposable for mastodon.streaming.Client

examples/timeline-with-streaming.ts

@@ -1,14 +1,15 @@
 import { createStreamingAPIClient } from "masto";
 
 const subscribe = async (): Promise<void> => {
-  const masto = createStreamingAPIClient({
+  using masto = createStreamingAPIClient({
     streamingApiUrl: "<STREAMING API URL>",
     accessToken: "<TOKEN>",
   });
 
+  using events = masto.hashtag.subscribe({ tag: "mastojs" });
   console.log("subscribed to #mastojs");
 
-  for await (const event of masto.hashtag.subscribe({ tag: "mastojs" })) {
+  for await (const event of events) {
     switch (event.event) {
       case "update": {
         console.log("new post", event.payload.content);

jest.config.js

@@ -6,6 +6,7 @@ export default {
       testEnvironment: "node",
       testMatch: ["<rootDir>/src/**/*.spec.ts"],
       transform: { "^.+\\.tsx?$": "ts-jest" },
+      setupFilesAfterEnv: ["<rootDir>/test-utils/jest-setup-after-env-unit.ts"],
     },
     {
       displayName: "e2e",

src/adapters/action/dispatcher-ws.spec.ts

@@ -22,6 +22,20 @@ describe("DispatcherWs", () => {
         data: undefined,
         meta: {},
       });
-    }).toThrowError(MastoUnexpectedError);
+    }).toThrow(MastoUnexpectedError);
+  });
+
+  it("can be disposed", () => {
+    const connector = new WebSocketConnectorImpl({
+      constructorParameters: ["wss://example.com"],
+    });
+    const dispatcher = new WebSocketActionDispatcher(
+      connector,
+      new SerializerNativeImpl(),
+      createLogger("error"),
+    );
+
+    dispatcher[Symbol.dispose]();
+    expect(connector.canAcquire()).toBe(false);
   });
 });

src/adapters/action/dispatcher-ws.ts

@@ -45,4 +45,8 @@ export class WebSocketActionDispatcher
       { ...data },
     ) as T;
   }
+
+  [Symbol.dispose](): void {
+    this.connector.close();
+  }
 }

src/adapters/action/proxy.spec.ts

@@ -102,4 +102,18 @@ describe("RequestBuilder", () => {
     expect(() => api()).toThrow(TypeError);
     expect(() => api.close()).not.toThrow(TypeError);
   });
+
+  it("can be disposed", () => {
+    const dispatcher = {
+      dispatch: async <T>(_: AnyAction) => {
+        return {} as T;
+      },
+      [Symbol.dispose]: jest.fn(),
+    };
+    const api: any = createActionProxy(dispatcher, { context: ["root"] });
+
+    api[Symbol.dispose]();
+
+    expect(dispatcher[Symbol.dispose]).toHaveBeenCalled();
+  });
 });

src/adapters/action/proxy.ts

@@ -60,6 +60,9 @@ const get =
     if (typeof property === "string" && SPECIAL_PROPERTIES.has(property)) {
       return;
     }
+    if (property === Symbol.dispose) {
+      return actionDispatcher[Symbol.dispose];
+    }
     if (typeof property === "symbol") {
       return;
     }

src/interfaces/action.ts

@@ -12,6 +12,7 @@ export type AnyAction = Action<string>;
 
 export interface ActionDispatcher<T extends AnyAction> {
   dispatch<U>(action: T): U | Promise<U>;
+  [Symbol.dispose]?(): void;
 }
 
 export interface ActionDispatcherHook<T extends AnyAction, U = unknown> {

src/mastodon/streaming/client.ts

@@ -8,18 +8,13 @@ export interface SubscribeHashtagParams {
   readonly tag: string;
 }
 
-export interface Subscription {
-  unsubscribe(): void;
+export interface Subscription extends AsyncIterable<Event>, Disposable {
   values(): AsyncIterableIterator<Event>;
-  [Symbol.asyncIterator](): AsyncIterator<Event, undefined>;
 
-  /**
-   * @experimental This is an experimental API.
-   */
-  [Symbol.dispose](): void;
+  unsubscribe(): void;
 }
 
-export interface Client {
+export interface Client extends Disposable {
   public: {
     subscribe(): Subscription;
     media: {

test-utils/jest-setup-after-env-unit.ts

@@ -0,0 +1,4 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
+(Symbol as any).dispose ??= Symbol("Symbol.dispose");
+(Symbol as any).asyncDispose ??= Symbol("Symbol.asyncDispose");
+/* eslint-enable @typescript-eslint/no-explicit-any */