fix(fetch): polling completed and cleanup of operators

Author: StephanGerbeth

package-lock.json

@@ -44,7 +44,7 @@
     "eslint-plugin-prettier": "5.2.1",
     "eslint-plugin-security": "3.0.1",
     "eslint-plugin-vitest": "0.5.4",
-    "fetch-mock": "11.1.3",
+    "fetch-mock": "12.0.0",
     "happy-dom": "15.7.4",
     "husky": "9.1.6",
     "lint-staged": "15.2.10",

package.json

@@ -19,6 +19,7 @@
   },
   "dependencies": {
     "@rxjs-collection/observables": "*",
+    "fast-equals": "^5.0.1",
     "rxjs": "7.8.1"
   },
   "devDependencies": {

packages/operators/package.json

@@ -8,12 +8,12 @@ export const autoPagination = ({ resolveRoute }) => {
       concatMap(({ url }) => {
         return from(resolveRoute(url)).pipe(
           download(),
-          expand(resp => {
-            return from(resolveRoute(url, resp)).pipe(
+          expand(resp =>
+            from(resolveRoute(url, resp)).pipe(
               filter(url => !!url),
               download()
-            );
-          })
+            )
+          )
         );
       }),
       map(resp => resp.clone())

packages/operators/src/fetch/autoPagination.js

@@ -8,33 +8,29 @@ import { resolveJSON } from './resolve.js';
 
 describe('download operator', function () {
   beforeEach(function () {
-    fetchMock.get(
+    fetchMock.mockGlobal().get(
       'https://httpbin.org/my-url-fast',
-      new Response(JSON.stringify({ hello: 'fast world' }), {
-        status: 200,
-        headers: {
-          'Content-type': 'application/json'
-        }
-      }),
-      {
-        delay: 1000
-      }
+      () => {
+        return new Response(JSON.stringify({ hello: 'fast world' }), {
+          status: 200,
+          headers: { 'Content-type': 'application/json' }
+        });
+      },
+      { delay: 1000 }
     );
   });
 
   afterEach(function () {
-    fetchMock.restore();
+    fetchMock.unmockGlobal();
   });
 
   test('successfull download - indirect json resolve', () =>
     new Promise(done => {
       of('https://httpbin.org/my-url-fast')
         .pipe(download(), log(false), resolveJSON(), log(false))
         .subscribe({
-          next: data => {
-            expect(data).deep.equal({ hello: 'fast world' });
-          },
-          complete: e => done()
+          next: data => expect(data).deep.equal({ hello: 'fast world' }),
+          complete: () => done()
         });
     }));
 
@@ -43,9 +39,7 @@ describe('download operator', function () {
       of('https://httpbin.org/my-url-fast')
         .pipe(downloadJSON(), log(false))
         .subscribe({
-          next: data => {
-            expect(data).deep.equal({ hello: 'fast world' });
-          },
+          next: data => expect(data).deep.equal({ hello: 'fast world' }),
           complete: () => done()
         });
     }));

packages/operators/src/fetch/download.test.js

@@ -1,20 +1,14 @@
-import { concatMap, delay as delayOperator, EMPTY, expand, of } from 'rxjs';
+import { delay, expand, of } from 'rxjs';
 
+import { log } from '../log';
 import { download } from './download';
+import { distinctUntilResponseChanged } from './response';
 
-export const polling = ({ validateResult, delay = 1000 }) => {
+export const polling = (timeout = 1000) => {
   return source =>
     source.pipe(
-      concatMap(({ url }) => {
-        return of(url).pipe(
-          download(),
-          expand(data => {
-            if (validateResult(data)) {
-              return of(url).pipe(delayOperator(delay), download());
-            }
-            return EMPTY;
-          })
-        );
-      })
+      download(),
+      expand(resp => of(resp.url).pipe(delay(timeout), download())),
+      distinctUntilResponseChanged()
     );
 };

packages/operators/src/fetch/polling.js

@@ -1,27 +1,40 @@
-import { concatAll, map, of } from 'rxjs';
-import { beforeEach, describe, expect, test } from 'vitest';
+import fetchMock from 'fetch-mock';
+import { of, take } from 'rxjs';
+import { afterEach, beforeEach, describe, expect, test } from 'vitest';
 
+import { log } from '../log';
 import { polling } from './polling';
+import { resolveJSON } from './resolve';
 
 describe('polling', function () {
   beforeEach(function () {
-    //
+    let counter = 0;
+    fetchMock.mockGlobal().get('https://httpbin.org/my-url-fast', () => {
+      if (counter++ < 2) {
+        return new Response(JSON.stringify({ hello: 'fast world' }), {
+          status: 200,
+          headers: { 'Content-type': 'application/json' }
+        });
+      }
+      return new Response(JSON.stringify({ hello: 'faster world' }), {
+        status: 200,
+        headers: { 'Content-type': 'application/json' }
+      });
+    });
+  });
+
+  afterEach(function () {
+    fetchMock.unmockGlobal();
   });
 
   test('auto polling', async function () {
+    const expected = [{ hello: 'fast world' }, { hello: 'faster world' }];
+
     return new Promise(done => {
-      of({ url: new URL('https://dummyjson.com/products') })
-        .pipe(
-          polling({
-            validateResult: data => {
-              return data.total > data.skip + data.limit;
-            }
-          })
-          // map(({ data: { products } }) => products),
-          // concatAll()
-        )
+      of(new URL('https://httpbin.org/my-url-fast'))
+        .pipe(polling(), log(false), resolveJSON(), log(false), take(2))
         .subscribe({
-          next: e => console.log('aha'),
+          next: e => expect(e).deep.include(expected.shift()),
           complete: () => done()
         });
     });

packages/operators/src/fetch/polling.test.js

@@ -1,13 +1,11 @@
 import { concatMap } from 'rxjs';
 
-import { log } from '../log';
 import { networkRetry } from './retry';
 
 export const request = () => {
   return source =>
     source.pipe(
       concatMap(req => fetch(req)),
-      log(false),
       networkRetry()
     );
 };

packages/operators/src/fetch/request.js

@@ -8,7 +8,7 @@ import { request } from './request.js';
 describe('request observable with default operators', function () {
   beforeEach(function () {
     let counter = 0;
-    fetchMock.get(
+    fetchMock.mockGlobal().get(
       'https://httpbin.org/my-url-fast',
       () => {
         return new Response(JSON.stringify({ hello: 'fast world' }), {
@@ -21,17 +21,15 @@ describe('request observable with default operators', function () {
   });
 
   afterEach(function () {
-    fetchMock.restore();
+    fetchMock.unmockGlobal();
   });
 
   test('successfull request', () =>
     new Promise(done => {
       of('https://httpbin.org/my-url-fast')
         .pipe(request(), log(false))
         .subscribe({
-          next: resp => {
-            expect(resp).deep.includes({ ok: true });
-          },
+          next: resp => expect(resp).deep.includes({ ok: true }),
           complete: () => done()
         });
     }));

packages/operators/src/fetch/request.test.js

@@ -0,0 +1,37 @@
+import { of } from 'rxjs';
+import { afterEach, test, describe, beforeEach, expect } from 'vitest';
+
+import { log } from '../log';
+import { resolveJSON, resolveText } from './resolve';
+
+describe('resolver', function () {
+  beforeEach(function () {
+    //
+  });
+
+  afterEach(function () {
+    //
+  });
+
+  test('resolve json', () => {
+    return new Promise(done => {
+      of(new Response(JSON.stringify({ hello: 'world' })))
+        .pipe(resolveJSON(), log(false))
+        .subscribe({
+          next: e => expect(e).includes({ hello: 'world' }),
+          complete: () => done()
+        });
+    });
+  });
+
+  test('resolve text', () => {
+    return new Promise(done => {
+      of(new Response('hello world'))
+        .pipe(resolveText(), log(false))
+        .subscribe({
+          next: e => expect(e).toBe('hello world'),
+          complete: () => done()
+        });
+    });
+  });
+});