fix(core): Independent abort signals

Resolves #41

.github/workflows/ci.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       matrix:
         nodejs: [18, 19, 20, 21]
-        rxjs: [5, 6, 7]
+        rxjs: [6, 7]
         axios: ["1.0", 1.1, 1.2, 1.3, 1.4, 1.6]
 
     steps:

.mocharc.json

@@ -4,7 +4,8 @@
   "recursive": true,
   "require": [
     "ts-node/register",
-    "test/hooks.ts"
+    "./test/hooks.ts",
+    "./test/setup.ts"
   ],
   "spec": [
     "test/**/*.test.*"

package.json

@@ -36,7 +36,8 @@
   },
   "packageManager": "[email protected]",
   "devDependencies": {
-    "@assertive-ts/core": "^2.0.0",
+    "@assertive-ts/core": "^2.1.0",
+    "@assertive-ts/sinon": "^1.0.0",
     "@eslint/compat": "^1.1.0",
     "@eslint/eslintrc": "^3.1.0",
     "@eslint/js": "^9.5.0",

src/lib/RxjsAxios.ts

@@ -39,11 +39,6 @@ interface Interceptors {
   response: AxiosInterceptorManager<AxiosResponse<unknown>>;
 }
 
-interface Abortable {
-  controller: AbortController;
-  signal: AbortSignal;
-}
-
 const logger = pino({ browser: { asObject: false } });
 
 export class RxjsAxios {
@@ -95,64 +90,44 @@ export class RxjsAxios {
     config: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.request<T, R, D>({ ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.request<T, R, D>({ ...reqConfig, signal }));
   }
 
   public get<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
     url: string,
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.get<T, R, D>(url, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.get<T, R, D>(url, { ...reqConfig, signal }));
   }
 
   public delete<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
     url: string,
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.delete<T, R, D>(url, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.delete<T, R, D>(url, { ...reqConfig, signal }));
   }
 
   public head<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
     url: string,
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.head<T, R, D>(url, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.head<T, R, D>(url, { ...reqConfig, signal }));
   }
 
   public options<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
     url: string,
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.options<T, R, D>(url, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.options<T, R, D>(url, { ...reqConfig, signal }));
   }
 
   public post<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
@@ -161,12 +136,8 @@ export class RxjsAxios {
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.post<T, R, D>(url, data, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.post<T, R, D>(url, data, { ...reqConfig, signal }));
   }
 
   public put<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
@@ -175,12 +146,8 @@ export class RxjsAxios {
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.put<T, R, D>(url, data, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.put<T, R, D>(url, data, { ...reqConfig, signal }));
   }
 
   public patch<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
@@ -189,12 +156,8 @@ export class RxjsAxios {
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.patch<T, R, D>(url, data, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.patch<T, R, D>(url, data, { ...reqConfig, signal }));
   }
 
   public postForm<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
@@ -203,12 +166,8 @@ export class RxjsAxios {
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.postForm<T, R, D>(url, data, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.postForm<T, R, D>(url, data, { ...reqConfig, signal }));
   }
 
   public putForm<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
@@ -217,12 +176,8 @@ export class RxjsAxios {
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.putForm<T, R, D>(url, data, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.putForm<T, R, D>(url, data, { ...reqConfig, signal }));
   }
 
   public patchForm<T, R extends AxiosResponse<T> = AxiosResponse<T>, D = unknown>(
@@ -231,12 +186,8 @@ export class RxjsAxios {
     config?: AxiosRequestConfig<D>,
   ): Observable<R> {
     const reqConfig = this.validateConfig(config);
-    const { controller, signal } = this.makeCancellable();
 
-    return observify(
-      () => this.axios.patchForm<T, R, D>(url, data, { ...reqConfig, signal }),
-      controller,
-    );
+    return observify(signal => this.axios.patchForm<T, R, D>(url, data, { ...reqConfig, signal }));
   }
 
   private validateConfig<D>(config?: AxiosRequestConfig<D>): AxiosRequestConfig<D> | undefined {
@@ -257,14 +208,4 @@ export class RxjsAxios {
 
     return config;
   }
-
-  private makeCancellable(): Abortable {
-    const controller = new AbortController();
-    const signal = controller.signal;
-
-    return {
-      controller,
-      signal,
-    };
-  }
 }

src/lib/observify.ts

@@ -2,16 +2,17 @@ import { Observable } from "rxjs";
 
 import type { AxiosResponse } from "./RxjsAxios";
 
-export function observify<T, R extends AxiosResponse<T>>(
-  makeRequest: () => Promise<R>,
-  controller: AbortController,
-): Observable<R> {
+type RequestFn<R> = (signal: AbortSignal) => Promise<R>;
+
+export function observify<T, R extends AxiosResponse<T>>(makeRequest: RequestFn<R>): Observable<R> {
   return new Observable(subscriber => {
-    makeRequest()
+    const controller = new AbortController();
+
+    makeRequest(controller.signal)
       .then(response => subscriber.next(response))
       .catch((error: unknown) => subscriber.error(error))
       .finally(() => subscriber.complete());
 
-    return { unsubscribe: () => controller.abort() };
+    return () => controller.abort();
   });
 }

test/helpers/mocks.ts

@@ -1,6 +1,6 @@
 import { HttpResponse, PathParams, http } from "msw";
 
-interface User {
+export interface User {
   id?: number;
   lastname: string;
   name: string;

test/lib/RxjsAxios.test.ts

@@ -2,10 +2,12 @@ import { TypeFactories, expect } from "@assertive-ts/core";
 import originalAxios from "axios";
 import FormData from "form-data";
 import { Observable } from "rxjs";
+import { map, repeat } from "rxjs/operators";
 import Sinon from "sinon";
 
 import { axios } from "../../src";
 import { RxjsAxios } from "../../src/lib/RxjsAxios";
+import { User } from "../helpers/mocks";
 
 describe("[Unit] RxjsAxios.test.ts", () => {
   describe(".of", () => {
@@ -150,6 +152,26 @@ describe("[Unit] RxjsAxios.test.ts", () => {
         done();
       });
     });
+
+    context("when the data is fetched repeatedly", () => {
+      it("returns the data on each rquest", done => {
+        const ref = { count: 0 };
+
+        axios.get<User[]>("http://localhost:8080/users")
+          .pipe(
+            map(({ data }) => data),
+            repeat({ count: 3, delay: 200 }),
+          )
+          .subscribe(users => {
+            ref.count++;
+            expect(users).not.toBeEmpty();
+
+            if (ref.count >= 3) {
+              done();
+            }
+          });
+      });
+    });
   });
 
   describe(".delete", () => {

test/lib/observify.test.ts

@@ -16,17 +16,12 @@ const RESPONSE: AxiosResponse<string> = {
 
 const REQUEST_ERROR = new AxiosError("Something went wrong", "Bad Request");
 
-const controller = new AbortController();
-
 describe("[Unit] observify.test.ts", () => {
   describe(".observify", () => {
     context("when the request promise is resolved", () => {
       context("and the observable is not unsubscribed", () => {
         it("sets the axios response on the next value and completes the observable", done => {
-          const observable = observify(
-            () => Promise.resolve<AxiosResponse<unknown>>(RESPONSE),
-            controller,
-          );
+          const observable = observify(() => Promise.resolve<AxiosResponse<unknown>>(RESPONSE));
 
           observable.subscribe({
             complete: done,
@@ -38,35 +33,31 @@ describe("[Unit] observify.test.ts", () => {
 
       context("and the observable is unsubscribed", () => {
         it("does not set the next value and cancels the request", done => {
-          const abort = Sinon.spy(controller, "abort");
-          const observable = observify(
-            () => delay(10).then(() => RESPONSE),
-            controller,
-          );
+          const spy = Sinon.spy();
+          const observable = observify(signal => {
+            signal.addEventListener("abort", spy);
+            return delay(10).then(() => RESPONSE);
+          });
 
           const subscription = observable.subscribe({
             complete: done,
             error: error => expect(error).not.toBePresent(),
             next: response => expect(response).not.toBePresent(),
           });
 
+          expect(spy).toNeverBeCalled();
           subscription.unsubscribe();
 
-          delay(20)
-            .then(() => Sinon.assert.calledOnce(abort))
-            .then(done)
-            .catch(done);
+          expect(spy).toBeCalledOnce();
+          done();
         });
       });
     });
 
     context("when the request promise is rejected", () => {
       context("and the observable is not unsubscribed", () => {
         it("sets the axios error on the error value and completes the observable", done => {
-          const observable = observify(
-            () => Promise.reject(REQUEST_ERROR),
-            controller,
-          );
+          const observable = observify(() => Promise.reject(REQUEST_ERROR));
 
           observable.subscribe({
             complete: done,
@@ -81,24 +72,23 @@ describe("[Unit] observify.test.ts", () => {
 
       context("and the observable is unsubscribed", () => {
         it("does not set the error value and cancels the request", done => {
-          const abort = Sinon.spy(controller, "abort");
-          const observable = observify(
-            () => delay(10).then(() => Promise.reject(REQUEST_ERROR)),
-            controller,
-          );
+          const spy = Sinon.spy();
+          const observable = observify(signal => {
+            signal.addEventListener("abort", spy);
+            return delay(10).then(() => Promise.reject(REQUEST_ERROR));
+          });
 
           const subscription = observable.subscribe({
             complete: done,
             error: error => expect(error).not.toBePresent(),
             next: response => expect(response).not.toBePresent(),
           });
 
+          expect(spy).toNeverBeCalled();
           subscription.unsubscribe();
 
-          delay(20)
-            .then(() => Sinon.assert.calledOnce(abort))
-            .then(done)
-            .catch(done);
+          expect(spy).toBeCalledOnce();
+          done();
         });
       });
     });

test/setup.ts

@@ -0,0 +1,4 @@
+import { usePlugin } from "@assertive-ts/core";
+import { SinonPlugin } from "@assertive-ts/sinon";
+
+usePlugin(SinonPlugin);