fix(stream): avoid event listener leak

Author: RobertCraigie

src/lib/BetaMessageStream.ts

@@ -181,22 +181,30 @@ export class BetaMessageStream implements AsyncIterable<BetaMessageStreamEvent>
     options?: RequestOptions,
   ): Promise<void> {
     const signal = options?.signal;
+    let abortHandler: (() => void) | undefined;
     if (signal) {
       if (signal.aborted) this.controller.abort();
-      signal.addEventListener('abort', () => this.controller.abort());
+      abortHandler = this.controller.abort.bind(this.controller);
+      signal.addEventListener('abort', abortHandler);
     }
-    this.#beginRequest();
-    const { response, data: stream } = await messages
-      .create({ ...params, stream: true }, { ...options, signal: this.controller.signal })
-      .withResponse();
-    this._connected(response);
-    for await (const event of stream) {
-      this.#addStreamEvent(event);
-    }
-    if (stream.controller.signal?.aborted) {
-      throw new APIUserAbortError();
+    try {
+      this.#beginRequest();
+      const { response, data: stream } = await messages
+        .create({ ...params, stream: true }, { ...options, signal: this.controller.signal })
+        .withResponse();
+      this._connected(response);
+      for await (const event of stream) {
+        this.#addStreamEvent(event);
+      }
+      if (stream.controller.signal?.aborted) {
+        throw new APIUserAbortError();
+      }
+      this.#endRequest();
+    } finally {
+      if (signal && abortHandler) {
+        signal.removeEventListener('abort', abortHandler);
+      }
     }
-    this.#endRequest();
   }
 
   protected _connected(response: Response | null) {
@@ -497,20 +505,28 @@ export class BetaMessageStream implements AsyncIterable<BetaMessageStreamEvent>
     options?: RequestOptions,
   ): Promise<void> {
     const signal = options?.signal;
+    let abortHandler: (() => void) | undefined;
     if (signal) {
       if (signal.aborted) this.controller.abort();
-      signal.addEventListener('abort', () => this.controller.abort());
+      abortHandler = this.controller.abort.bind(this.controller);
+      signal.addEventListener('abort', abortHandler);
     }
-    this.#beginRequest();
-    this._connected(null);
-    const stream = Stream.fromReadableStream<BetaMessageStreamEvent>(readableStream, this.controller);
-    for await (const event of stream) {
-      this.#addStreamEvent(event);
-    }
-    if (stream.controller.signal?.aborted) {
-      throw new APIUserAbortError();
+    try {
+      this.#beginRequest();
+      this._connected(null);
+      const stream = Stream.fromReadableStream<BetaMessageStreamEvent>(readableStream, this.controller);
+      for await (const event of stream) {
+        this.#addStreamEvent(event);
+      }
+      if (stream.controller.signal?.aborted) {
+        throw new APIUserAbortError();
+      }
+      this.#endRequest();
+    } finally {
+      if (signal && abortHandler) {
+        signal.removeEventListener('abort', abortHandler);
+      }
     }
-    this.#endRequest();
   }
 
   /**

src/lib/MessageStream.ts

@@ -180,22 +180,30 @@ export class MessageStream implements AsyncIterable<MessageStreamEvent> {
     options?: RequestOptions,
   ): Promise<void> {
     const signal = options?.signal;
+    let abortHandler: (() => void) | undefined;
     if (signal) {
       if (signal.aborted) this.controller.abort();
-      signal.addEventListener('abort', () => this.controller.abort());
+      abortHandler = this.controller.abort.bind(this.controller);
+      signal.addEventListener('abort', abortHandler);
     }
-    this.#beginRequest();
-    const { response, data: stream } = await messages
-      .create({ ...params, stream: true }, { ...options, signal: this.controller.signal })
-      .withResponse();
-    this._connected(response);
-    for await (const event of stream) {
-      this.#addStreamEvent(event);
-    }
-    if (stream.controller.signal?.aborted) {
-      throw new APIUserAbortError();
+    try {
+      this.#beginRequest();
+      const { response, data: stream } = await messages
+        .create({ ...params, stream: true }, { ...options, signal: this.controller.signal })
+        .withResponse();
+      this._connected(response);
+      for await (const event of stream) {
+        this.#addStreamEvent(event);
+      }
+      if (stream.controller.signal?.aborted) {
+        throw new APIUserAbortError();
+      }
+      this.#endRequest();
+    } finally {
+      if (signal && abortHandler) {
+        signal.removeEventListener('abort', abortHandler);
+      }
     }
-    this.#endRequest();
   }
 
   protected _connected(response: Response | null) {
@@ -496,20 +504,28 @@ export class MessageStream implements AsyncIterable<MessageStreamEvent> {
     options?: RequestOptions,
   ): Promise<void> {
     const signal = options?.signal;
+    let abortHandler: (() => void) | undefined;
     if (signal) {
       if (signal.aborted) this.controller.abort();
-      signal.addEventListener('abort', () => this.controller.abort());
+      abortHandler = this.controller.abort.bind(this.controller);
+      signal.addEventListener('abort', abortHandler);
     }
-    this.#beginRequest();
-    this._connected(null);
-    const stream = Stream.fromReadableStream<MessageStreamEvent>(readableStream, this.controller);
-    for await (const event of stream) {
-      this.#addStreamEvent(event);
-    }
-    if (stream.controller.signal?.aborted) {
-      throw new APIUserAbortError();
+    try {
+      this.#beginRequest();
+      this._connected(null);
+      const stream = Stream.fromReadableStream<MessageStreamEvent>(readableStream, this.controller);
+      for await (const event of stream) {
+        this.#addStreamEvent(event);
+      }
+      if (stream.controller.signal?.aborted) {
+        throw new APIUserAbortError();
+      }
+      this.#endRequest();
+    } finally {
+      if (signal && abortHandler) {
+        signal.removeEventListener('abort', abortHandler);
+      }
     }
-    this.#endRequest();
   }
 
   /**