fix(transfer): optimized stats workflow

Author: StephanGerbeth

packages/operators/src/transfer/stats/Bandwidth.js

@@ -1,21 +1,27 @@
-import { map, Subject } from 'rxjs';
+import { concat, delay, EMPTY, map, of, Subject, switchMap } from 'rxjs';
 
 import { calcReceivedStats, MBIT, SECOND } from './utils';
 
 export default {
-  create: (byteUnit = MBIT, timeUnit = SECOND) => {
+  create: (byteRatio = MBIT, timeRatio = SECOND) => {
     return new Subject().pipe(
       calcReceivedStats(),
-      calcAverageByteLengthPerTimeUnit(timeUnit),
-      calcTransferRate(byteUnit)
+      calcBandwidth(byteRatio, timeRatio)
+      //
     );
   }
 };
 
-const calcAverageByteLengthPerTimeUnit = timeRatio => {
-  return source => source.pipe(map(({ value, period }) => (value / period) * timeRatio));
+const calcBandwidth = (byteRatio, timeRatio) => {
+  return source =>
+    source.pipe(
+      switchMap(stats => {
+        let noBandwidth = stats.value === stats.total ? EMPTY : of(0).pipe(delay(500));
+        return concat(calcTransmittableBytes(stats, byteRatio, timeRatio), noBandwidth);
+      })
+    );
 };
 
-const calcTransferRate = byteRatio => {
-  return source => source.pipe(map(bytes => bytes * byteRatio));
+const calcTransmittableBytes = (stats, byteRatio, timeRatio) => {
+  return of(stats).pipe(map(({ value, period }) => (value / period) * byteRatio * timeRatio));
 };

packages/operators/src/transfer/stats/Bandwidth.test.js

@@ -40,8 +40,36 @@ describe('Bandwidth', () => {
     const bandwidth = Bandwidth.create(KBIT);
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(bandwidth).toBe('--a-b-c-d-e|', expectedVal);
       expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(bandwidth)));
+      expectObservable(bandwidth).toBe('--a-b-c-d-e|', expectedVal);
+    });
+  });
+
+  test('bandwidth outage', async () => {
+    const time = Date.now();
+
+    const triggerVal = {
+      a: { value: new TextEncoder().encode('abcd'), total: 20, period: time },
+      b: { value: new TextEncoder().encode('edgh'), total: 20, period: time },
+      c: { value: new TextEncoder().encode('ijkl'), total: 20, period: time },
+      d: { value: new TextEncoder().encode('mnop'), total: 20, period: time },
+      e: { value: new TextEncoder().encode('qrst'), total: 20, period: time }
+    };
+
+    const expectedVal = {
+      a: 15.625,
+      b: 0,
+      c: 0.10364842454394693,
+      d: 0.15495867768595042,
+      e: 0.20593080724876442,
+      f: 0.256568144499179
+    };
+
+    const bandwidth = Bandwidth.create(KBIT);
+
+    testScheduler.run(({ cold, expectObservable }) => {
+      expectObservable(cold('--a 600ms b-c-d-e|', triggerVal).pipe(tap(bandwidth)));
+      expectObservable(bandwidth).toBe('--a 499ms b 100ms c-d-e-f|', expectedVal);
     });
   });
 });

packages/operators/src/transfer/stats/ElapsedTime.test.js

@@ -48,8 +48,8 @@ describe('ElapsedTime', () => {
     const elapsedTime = ElapsedTime.create();
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(elapsedTime).toBe('a-b-c-d-e-f-g-h-i|', expectedVal);
       expectObservable(cold('a-b-c-d-e-f-g-h-i|', triggerVal).pipe(tap(elapsedTime)));
+      expectObservable(elapsedTime).toBe('a-b-c-d-e-f-g-h-i|', expectedVal);
     });
   });
 
@@ -83,8 +83,8 @@ describe('ElapsedTime', () => {
     const elapsedTimeSecond = ElapsedTime.create(SECOND);
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(elapsedTimeSecond).toBe('a-b-c-d-e-f-g-h-i|', expectedVal);
       expectObservable(cold('a-b-c-d-e-f-g-h-i|', triggerVal).pipe(tap(elapsedTimeSecond)));
+      expectObservable(elapsedTimeSecond).toBe('a-b-c-d-e-f-g-h-i|', expectedVal);
     });
   });
 });

packages/operators/src/transfer/stats/Progress.test.js

@@ -39,8 +39,8 @@ describe('Progress', () => {
     const progress = Progress.create();
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(progress).toBe('a-b-c-d-e-f-g-h-i|', expectedVal);
       expectObservable(cold('a-b-c-d-e-f-g-h-i|', triggerVal).pipe(tap(progress)));
+      expectObservable(progress).toBe('a-b-c-d-e-f-g-h-i|', expectedVal);
     });
   });
 });

packages/operators/src/transfer/stats/TimeEstimate.js

@@ -1,24 +1,40 @@
-import { concatWith, distinctUntilChanged, map, of, Subject } from 'rxjs';
+import {
+  concat,
+  concatWith,
+  delay,
+  distinctUntilChanged,
+  EMPTY,
+  map,
+  of,
+  Subject,
+  switchMap
+} from 'rxjs';
 
 import { calcReceivedStats, MSECOND } from './utils';
 
 export default {
-  create: (timeUnit = MSECOND) => {
+  create: (timeRatio = MSECOND) => {
     return new Subject().pipe(
       calcReceivedStats(),
-      calcEstimatedTime(),
-      convertEstimatedTimeTo(timeUnit),
+      calcTimeEstimate(timeRatio),
       concatWith(of(0)),
       distinctUntilChanged()
     );
   }
 };
 
-const calcEstimatedTime = () => {
+const calcTimeEstimate = timeRatio => {
   return source =>
-    source.pipe(map(({ value, total, period }) => Math.ceil((total - value) * (period / value))));
+    source.pipe(
+      switchMap(stats => {
+        let noEstimation = stats.value === stats.total ? EMPTY : of(Infinity).pipe(delay(500));
+        return concat(calcEstimation(stats, timeRatio), noEstimation);
+      })
+    );
 };
 
-const convertEstimatedTimeTo = timeRatio => {
-  return source => source.pipe(map(value => value / timeRatio));
+const calcEstimation = (stats, timeRatio) => {
+  return of(stats).pipe(
+    map(({ value, total, period }) => Math.ceil((total - value) * (period / value)) / timeRatio)
+  );
 };

packages/operators/src/transfer/stats/TimeEstimate.test.js

@@ -18,7 +18,7 @@ describe('TimeEstimate', () => {
     vi.restoreAllMocks();
   });
 
-  test('calc estimate time - millisecond', async () => {
+  test('time estimation - millisecond', async () => {
     const time = Date.now();
 
     const triggerVal = {
@@ -40,12 +40,12 @@ describe('TimeEstimate', () => {
     const timeEstimate = TimeEstimate.create();
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(timeEstimate).toBe('--a-b-c-d-e|', expectedVal);
       expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(timeEstimate)));
+      expectObservable(timeEstimate).toBe('--a-b-c-d-e|', expectedVal);
     });
   });
 
-  test('calc estimate time - second', async () => {
+  test('time estimation - second', async () => {
     const time = Date.now();
 
     const triggerVal = {
@@ -67,8 +67,36 @@ describe('TimeEstimate', () => {
     const timeEstimateSecond = TimeEstimate.create(SECOND);
 
     testScheduler.run(({ cold, expectObservable }) => {
-      expectObservable(timeEstimateSecond).toBe('--a-b-c-d-e|', expectedVal);
       expectObservable(cold('--a-b-c-d-e|', triggerVal).pipe(tap(timeEstimateSecond)));
+      expectObservable(timeEstimateSecond).toBe('--a-b-c-d-e|', expectedVal);
+    });
+  });
+
+  test('time estimation - outage', async () => {
+    const time = Date.now();
+
+    const triggerVal = {
+      a: { value: new TextEncoder().encode('abcd'), total: 20, period: time },
+      b: { value: new TextEncoder().encode('edgh'), total: 20, period: time },
+      c: { value: new TextEncoder().encode('ijkl'), total: 20, period: time },
+      d: { value: new TextEncoder().encode('mnop'), total: 20, period: time },
+      e: { value: new TextEncoder().encode('qrst'), total: 20, period: time }
+    };
+
+    const expectedVal = {
+      a: 0.008,
+      b: Infinity,
+      c: 0.905,
+      d: 0.404,
+      e: 0.152,
+      f: 0
+    };
+
+    const timeEstimateSecond = TimeEstimate.create(SECOND);
+
+    testScheduler.run(({ cold, expectObservable }) => {
+      expectObservable(cold('--a 600ms b-c-d-e|', triggerVal).pipe(tap(timeEstimateSecond)));
+      expectObservable(timeEstimateSecond).toBe('--a 499ms b 100ms c-d-e-f|', expectedVal);
     });
   });
 });