fix(operators): marble testing

packages/operators/package.json

@@ -19,6 +19,8 @@
   },
   "dependencies": {
     "@rxjs-collection/observables": "*",
+    "ansi-colors": "^4.1.3",
+    "consola": "^3.2.3",
     "fast-equals": "5.0.1",
     "rxjs": "7.8.1"
   },

packages/operators/src/log.js

@@ -1,20 +1,30 @@
-import { Observable } from 'rxjs';
+import { bgGreen } from 'ansi-colors';
+import debug from 'debug';
+import { connectable, finalize, Observable, Subject } from 'rxjs';
 
-export const log = (active = true) => {
-  if (active) {
+export const enableLog = tag => {
+  debug.enable(tag);
+};
+
+export const log = tag => {
+  var logger = debug(tag);
+  logger.log = console.log.bind(console);
+  var error = debug(`${tag}:error`);
+
+  if (debug.enabled(tag)) {
     return source => {
       return new Observable(observer => {
         return source.subscribe(
           val => {
-            console.log(val);
+            logger(val);
             observer.next(val);
           },
           err => {
-            console.error(err);
+            error(err);
             observer.error(err);
           },
           () => {
-            console.log('%ccomplete', 'color: green');
+            logger(bgGreen.bold('Complete!'));
             observer.complete();
           }
         );
@@ -24,3 +34,15 @@ export const log = (active = true) => {
     return source => source;
   }
 };
+
+export const logResult = (tag, observable) => {
+  return new Promise(done => {
+    connectable(
+      observable.pipe(
+        log(tag),
+        finalize(() => done())
+      ),
+      { connector: () => new Subject() }
+    ).connect();
+  });
+};

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

@@ -1,16 +1,16 @@
 import { concatAll, concatMap, delay, from, map, of, toArray } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
-import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
+import { afterAll, afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
 
-import { log } from '../log';
+import { log, logOutput, logResult } from '../log';
 import { resolveJSON } from './response';
 
-describe('auto pagination - mocked', function () {
+describe('auto pagination - mocked', () => {
   const testScheduler = new TestScheduler((actual, expected) => {
     expect(actual).to.eql(expected);
   });
 
-  beforeEach(function () {
+  beforeEach(() => {
     vi.doMock('./request', importOriginal => ({
       request: () => source => source.pipe(concatMap(({ v, t }) => of(v).pipe(delay(t))))
     }));
@@ -25,6 +25,10 @@ describe('auto pagination - mocked', function () {
     vi.doUnmock('./request');
   });
 
+  afterAll(() => {
+    vi.resetModules();
+  });
+
   test('classic testing', async () => {
     const { autoPagination } = await import('./autoPagination');
 
@@ -76,47 +80,41 @@ describe('auto pagination - mocked', function () {
           autoPagination({
             resolveRoute: (conf, resp) =>
               ((!resp || resp.next) && [triggerVal[resp?.next || 'a']]) || []
-          })
+          }),
+          log('marble:result')
         )
       ).toBe('---a----b--cd---e----', expectedVal);
     });
   });
 });
 
-describe.skip('auto pagination - demo', function () {
-  beforeEach(function () {
-    vi.resetModules();
-  });
-
-  test('sample testing', async function () {
+describe('auto pagination - demo', () => {
+  test('sample testing', async () => {
     const { autoPagination } = await import('./autoPagination');
 
-    return new Promise(done => {
-      return of(new URL('https://dummyjson.com/products'))
-        .pipe(
-          autoPagination({
-            resolveRoute: async (url, resp) => {
-              const data = (await resp?.json()) || { skip: -10, limit: 10 };
-
-              if (!data.total || data.total > data.skip + data.limit) {
-                const newUrl = new URL(`${url}`);
-                newUrl.searchParams.set('skip', data.skip + data.limit);
-                newUrl.searchParams.set('limit', data.limit);
-                newUrl.searchParams.set('select', 'title,price');
-                return newUrl;
-              }
+    await logResult(
+      'demo',
+      of(new URL('https://dummyjson.com/products')).pipe(
+        autoPagination({
+          resolveRoute: async (url, resp) => {
+            const data = (await resp?.json()) || { skip: -10, limit: 10 };
+
+            if (!data.total || data.total > data.skip + data.limit) {
+              const newUrl = new URL(`${url}`);
+              newUrl.searchParams.set('skip', data.skip + data.limit);
+              newUrl.searchParams.set('limit', data.limit);
+              newUrl.searchParams.set('select', 'title,price');
+              return newUrl;
             }
-          }),
-          log(false),
-          resolveJSON(),
-          log(false),
-          map(({ products }) => products),
-          concatAll()
-        )
-        .subscribe({
-          next: e => console.log(e),
-          complete: () => done()
-        });
-    });
+          }
+        }),
+        log('demo:response'),
+        resolveJSON(),
+        log('demo:response:json'),
+        map(({ products }) => products),
+        log('demo:response:result'),
+        concatAll()
+      )
+    );
   });
 });

packages/operators/src/request/cache.js

@@ -4,11 +4,9 @@ export const cache = ttl => {
   return source =>
     source.pipe(
       share({
-        // TODO: check if a buffer size is neccessary
         connector: () => new ReplaySubject(),
-        // resetOnError: false,
         resetOnComplete: () => timer(ttl),
-        resetOnRefCountZero: false
+        resetOnRefCountZero: () => timer(ttl)
       })
     );
 };

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

@@ -1,21 +1,49 @@
 import fetchMock from 'fetch-mock';
-import { defer, delay, from, interval, map, mapTo, of, tap, throttleTime } from 'rxjs';
+import { defer, delay, map, of, tap } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
 import { afterEach, beforeEach, describe, expect, test } from 'vitest';
 
 import { cache } from './cache';
 import { requestText } from './request';
 
-describe('cache - mocked', function () {
-  beforeEach(function () {
+describe('cache - mocked', () => {
+  const testScheduler = new TestScheduler((actual, expected) => {
+    expect(actual).deep.equal(expected);
+  });
+
+  beforeEach(() => {
     //
   });
 
-  afterEach(function () {
+  afterEach(() => {
     //
   });
 
-  test.skip('cache resetted after 100ms', async function () {
+  test('marble testing', () => {
+    const initial = new Response('initial', { status: 200 });
+    const updated = new Response('updated', { status: 200 });
+    const orderedResponses = [initial, updated];
+
+    testScheduler.run(({ cold, expectObservable }) => {
+      const stream = cold('a-----------', {
+        a: () => orderedResponses.shift()
+      }).pipe(
+        map(fn => fn()),
+        cache(2)
+      );
+
+      const unsubA = '-^!';
+      expectObservable(stream, unsubA).toBe('-a', { a: initial }, new Error());
+
+      const unsubB = '----^!';
+      expectObservable(stream, unsubB).toBe('----a', { a: initial }, new Error());
+
+      const unsubC = '---------^--!';
+      expectObservable(stream, unsubC).toBe('---------a', { a: updated }, new Error());
+    });
+  });
+
+  test('cache resetted after 100ms', async () => {
     let counter = 0;
     const a = of(counter).pipe(
       tap(e => console.log('U', e)),
@@ -39,8 +67,8 @@ describe('cache - mocked', function () {
   });
 });
 
-describe('cache', function () {
-  beforeEach(function () {
+describe('cache', () => {
+  beforeEach(() => {
     let counter = 0;
     fetchMock.mockGlobal().get(
       'https://httpbin.org/my-url-fast',
@@ -53,11 +81,11 @@ describe('cache', function () {
     );
   });
 
-  afterEach(function () {
+  afterEach(() => {
     fetchMock.unmockGlobal();
   });
 
-  test('cache resetted after 100ms', async function () {
+  test('cache resetted after 100ms', async () => {
     const a = of('https://httpbin.org/my-url-fast').pipe(
       requestText(),
       tap(() => console.log('CHECK')),

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

@@ -1,16 +1,16 @@
 import { concatAll, concatMap, delay, from, map, of, tap, toArray } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
-import { afterEach, beforeAll, beforeEach, describe, expect, test, vi } from 'vitest';
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test, vi } from 'vitest';
 
-import { log } from '../log';
+import { log, logResult } from '../log';
 import { resolveJSON } from './response';
 
-describe('concurrent request - mocked', function () {
+describe('concurrent request - mocked', () => {
   const testScheduler = new TestScheduler((actual, expected) => {
     expect(actual).to.eql(expected);
   });
 
-  beforeEach(function () {
+  beforeEach(() => {
     vi.doMock('./request', importOriginal => ({
       request: () => source => source.pipe(concatMap(({ v, t }) => of(v).pipe(delay(t))))
     }));
@@ -20,6 +20,10 @@ describe('concurrent request - mocked', function () {
     vi.doUnmock('./request');
   });
 
+  afterAll(function () {
+    vi.resetModules();
+  });
+
   test('classic testing', async () => {
     const { concurrentRequest } = await import('./concurrentRequest');
 
@@ -63,15 +67,12 @@ describe('concurrent request - mocked', function () {
   });
 });
 
-describe.skip('concurrent request - demo', function () {
-  beforeAll(function () {
-    vi.resetModules();
-  });
-
+describe('concurrent request - demo', () => {
   test('sample testing', async () => {
     const { concurrentRequest } = await import('./concurrentRequest');
 
-    await new Promise(done => {
+    await logResult(
+      'demo',
       of(
         new URL('https://dummyjson.com/products?limit=10&skip=0&select=title,price'),
         new URL('https://dummyjson.com/products?limit=10&skip=10&select=title,price'),
@@ -82,19 +83,15 @@ describe.skip('concurrent request - demo', function () {
         new URL('https://dummyjson.com/products?limit=10&skip=60&select=title,price'),
         new URL('https://dummyjson.com/products?limit=10&skip=70&select=title,price'),
         new URL('https://dummyjson.com/products?limit=10&skip=80&select=title,price')
+      ).pipe(
+        concurrentRequest(4),
+        log('demo:response'),
+        resolveJSON(),
+        log('demo:response:json'),
+        map(({ products }) => products),
+        log('demo:response:result'),
+        concatAll()
       )
-        .pipe(
-          concurrentRequest(4),
-          log(false),
-          resolveJSON(),
-          log(false),
-          map(({ products }) => products),
-          concatAll()
-        )
-        .subscribe({
-          next: e => console.log(e),
-          complete: () => done()
-        });
-    });
+    );
   });
 });