fix(request): refactored stream to transfer

Author: StephanGerbeth

packages/operators/src/request.js

@@ -3,15 +3,15 @@ import { concatMap, from, throwError } from 'rxjs';
 import { cache } from './cache';
 import { resolveBlob, resolveJSON, resolveText } from './response';
 import { retryWhenRequestError } from './retry';
-import { bypassStream } from './stream/bypassStream';
+import { interceptTransfer } from './transfer/interceptTransfer';
 
 export const request = ({ retry, cache: cacheOptions, stats } = {}) => {
   return source =>
     source.pipe(
-      bypassStream(stats?.upload),
+      interceptTransfer(stats?.upload),
       tryRequest(),
       retryWhenRequestError(retry),
-      bypassStream(stats?.download),
+      interceptTransfer(stats?.download),
       cache(cacheOptions)
       //
     );

packages/operators/src/request.test.js

@@ -9,11 +9,11 @@ import { test, describe, beforeEach, expect, vi, afterAll, beforeAll } from 'vit
 
 import { log, logResult } from './log.js';
 import { resolveBlob, resolveJSON } from './response.js';
-import ElapsedTime from './stream/stats/ElapsedTime.js';
-import EstimateTime from './stream/stats/EstimateTime.js';
-import Progress from './stream/stats/Progress.js';
-import TransferRate from './stream/stats/TransferRate.js';
-import { MBYTE, SECOND } from './stream/stats/utils.js';
+import Bandwidth from './transfer/stats/Bandwidth.js';
+import ElapsedTime from './transfer/stats/ElapsedTime.js';
+import Progress from './transfer/stats/Progress.js';
+import TimeEstimate from './transfer/stats/TimeEstimate.js';
+import { MBYTE, SECOND } from './transfer/stats/utils.js';
 
 describe.skip('request', () => {
   let testScheduler;
@@ -209,11 +209,11 @@ describe('test', () => {
     const progress = Progress.create();
     progress.subscribe({ next: e => console.log('DOWNLOAD', e) });
 
-    const byteRate = TransferRate.create(MBYTE, SECOND);
-    byteRate.subscribe({ next: e => console.log('RATE', e) });
+    const bandwidth = Bandwidth.create(MBYTE, SECOND);
+    bandwidth.subscribe({ next: e => console.log('RATE', e) });
 
-    const estimateTime = EstimateTime.create(SECOND);
-    estimateTime.subscribe({ next: e => console.log('ESTIMATE', e) });
+    const timeEstimate = TimeEstimate.create(SECOND);
+    timeEstimate.subscribe({ next: e => console.log('ESTIMATE', e) });
 
     const elapsedTime = ElapsedTime.create(SECOND);
     elapsedTime.subscribe({ next: e => console.log('ELAPSED', e) });
@@ -233,7 +233,7 @@ describe('test', () => {
 
     const value = await lastValueFrom(
       of(req).pipe(
-        request({ stats: { download: [progress, byteRate, estimateTime, elapsedTime] } }),
+        request({ stats: { download: [progress, bandwidth, timeEstimate, elapsedTime] } }),
         resolveBlob()
         //
       )

packages/operators/src/stream/bypassStream.js renamed to packages/operators/src/transfer/interceptTransfer.js

@@ -2,28 +2,28 @@ import { concatMap, from, map, of } from 'rxjs';
 
 import { readBytes } from './utils';
 
-export const bypassStream = (reworkers = [], chunkSize = 60 * 1024) => {
+export const interceptTransfer = (reworkers = [], chunkSize = 60 * 1024) => {
   return source =>
     source.pipe(
       concatMap(requestResponse => {
         if (reworkers.length) {
           return of(requestResponse).pipe(
-            objectToStream(),
+            sourceToStream(),
             interceptStream(reworkers, chunkSize),
-            streamToObject(requestResponse)
+            streamToSource(requestResponse)
           );
         }
         return of(requestResponse);
       })
     );
 };
 
-const objectToStream = () => {
+const sourceToStream = () => {
   return source =>
     source.pipe(concatMap(reqResp => objectToStreamMap.get(reqResp.constructor)(reqResp)));
 };
 
-const streamToObject = reqResp => {
+const streamToSource = reqResp => {
   return source =>
     source.pipe(concatMap(stream => streamToObjectMap.get(reqResp.constructor)(stream, reqResp)));
 };

packages/operators/src/stream/bypassStream.test.js renamed to packages/operators/src/transfer/interceptTransfer.test.js

@@ -5,9 +5,9 @@ import { of, Subject } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
 import { beforeAll, beforeEach, describe, expect, test, vi } from 'vitest';
 
-import { bypassStream } from './bypassStream';
+import { interceptTransfer } from './interceptTransfer';
 
-describe('bypass stream', () => {
+describe('intercept transfer', () => {
   let testScheduler;
 
   beforeAll(() => {
@@ -19,21 +19,21 @@ describe('bypass stream', () => {
     testScheduler = new TestScheduler((actual, expected) => expect(actual).deep.equal(expected));
   });
 
-  test('bypass response stream', async () => {
+  test('intercept response', async () => {
     const blob = new Blob(
       [fs.readFileSync(`${__dirname}/../../fixtures/videos/demo.mp4`)],
       'video/mp4'
     );
 
-    const bypass = new Subject();
+    const intercept = new Subject();
 
     const triggerVal = {
       a: new Response(blob, { status: 200 })
     };
 
     // testScheduler.run(({ cold, expectObservable }) => {
     //   expectObservable(bypass).toBe('a');
-    //   expectObservable(cold('a|', triggerVal).pipe(bypassStream([bypass])));
+    //   expectObservable(cold('a|', triggerVal).pipe(interceptTransfer([bypass])));
     // });
   });
 });

packages/operators/src/stream/stats/TransferRate.js renamed to packages/operators/src/transfer/stats/Bandwidth.js

@@ -2,10 +2,10 @@ import { tap } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
 import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
 
-import TransferRate from './TransferRate';
+import Bandwidth from './Bandwidth';
 import { KBIT } from './utils';
 
-describe('TransferRate', () => {
+describe('Bandwidth', () => {
   let testScheduler;
 
   beforeEach(() => {
@@ -18,7 +18,7 @@ describe('TransferRate', () => {
     vi.restoreAllMocks();
   });
 
-  test('calc transfer rate', async () => {
+  test('calc bandwidth', async () => {
     const time = Date.now();
 
     const triggerVal = {
@@ -37,11 +37,11 @@ describe('TransferRate', () => {
       e: 15.625
     };
 
-    const transferRate = TransferRate.create(KBIT);
+    const bandwidth = Bandwidth.create(KBIT);
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(transferRate).toBe('--a-b-c-d-e|', expectedVal);
-      expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(transferRate)));
+      expectObservable(bandwidth).toBe('--a-b-c-d-e|', expectedVal);
+      expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(bandwidth)));
     });
   });
 });

packages/operators/src/stream/stats/TransferRate.test.js renamed to packages/operators/src/transfer/stats/Bandwidth.test.js

@@ -7,7 +7,7 @@ export default {
     return new Subject().pipe(
       calcReceivedStats(),
       calcEstimatedTime(),
-      convertEstimedTimeTo(timeUnit),
+      convertEstimatedTimeTo(timeUnit),
       concatWith(of(0)),
       distinctUntilChanged()
     );
@@ -19,6 +19,6 @@ const calcEstimatedTime = () => {
     source.pipe(map(({ value, total, period }) => Math.ceil((total - value) * (period / value))));
 };
 
-const convertEstimedTimeTo = timeRatio => {
+const convertEstimatedTimeTo = timeRatio => {
   return source => source.pipe(map(value => value / timeRatio));
 };

packages/operators/src/stream/stats/ElapsedTime.js renamed to packages/operators/src/transfer/stats/ElapsedTime.js

@@ -2,10 +2,10 @@ import { tap } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
 import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
 
-import EstimateTime from './EstimateTime';
+import TimeEstimate from './TimeEstimate';
 import { SECOND } from './utils';
 
-describe('EstimateTime', () => {
+describe('TimeEstimate', () => {
   let testScheduler;
 
   beforeEach(() => {
@@ -37,11 +37,11 @@ describe('EstimateTime', () => {
       e: 0
     };
 
-    const estimateTime = EstimateTime.create();
+    const timeEstimate = TimeEstimate.create();
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(estimateTime).toBe('--a-b-c-d-e|', expectedVal);
-      expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(estimateTime)));
+      expectObservable(timeEstimate).toBe('--a-b-c-d-e|', expectedVal);
+      expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(timeEstimate)));
     });
   });
 
@@ -64,11 +64,11 @@ describe('EstimateTime', () => {
       e: 0
     };
 
-    const estimateTimeSecond = EstimateTime.create(SECOND);
+    const timeEstimateSecond = TimeEstimate.create(SECOND);
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(estimateTimeSecond).toBe('--a-b-c-d-e|', expectedVal);
-      expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(estimateTimeSecond)));
+      expectObservable(timeEstimateSecond).toBe('--a-b-c-d-e|', expectedVal);
+      expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(timeEstimateSecond)));
     });
   });
 });