fix(operators): fix cache log

Author: StephanGerbeth

packages/operators/src/cache.test.js

@@ -24,8 +24,8 @@ describe('cache', () => {
       const stream = cold('a', { a: () => triggerVal.shift() }).pipe(
         map(fn => fn()),
         log('operators:cache:default:input'),
-        cache({ ttl: 2 }),
-        log('operators:cache:default:output')
+        cache({ ttl: 2 })
+        // log('operators:cache:default:output')
       );
 
       const unsubA = '-^!';

packages/operators/src/log.js

@@ -1,6 +1,6 @@
 import { bgGreen } from 'ansi-colors';
 import debug from 'debug';
-import { connectable, finalize, Subject, tap } from 'rxjs';
+import { connectable, finalize, Observable, Subject, tap } from 'rxjs';
 
 import { pipeWhen } from './when';
 
@@ -17,17 +17,14 @@ export const log = tag => {
     source.pipe(
       pipeWhen(
         () => debug.enabled(tag),
-        source =>
-          source.pipe(
-            tap({
-              subscribe: () => logger('subscribed'),
-              unsubscribe: () => logger('unsubscribed'),
-              finalize: () => logger('finalize'),
-              next: val => logger(val),
-              error: err => error(err),
-              complete: () => logger(bgGreen.bold('complete!'))
-            })
-          )
+        tap({
+          subscribe: () => logger('subscribed'),
+          unsubscribe: () => logger('unsubscribed'),
+          finalize: () => logger('finalize'),
+          next: val => logger(val),
+          error: err => error(err),
+          complete: () => logger(bgGreen.bold('complete!'))
+        })
       )
     );
 };

packages/operators/src/retry.js

@@ -17,7 +17,7 @@ export const retryWhenRequestError = ({
     return source.pipe(
       pipeWhen(
         resp => (retryableStatuses && retryableStatuses.includes(resp.status)) || !resp.ok,
-        source => source.pipe(concatMap(() => throwError(() => new Error('invalid request'))))
+        concatMap(() => throwError(() => new Error('invalid request')))
       ),
       retry({ count, delay: () => determineDelayWhenOnline(timeout, ++counter) })
     );

packages/operators/src/when.js

@@ -1,6 +1,8 @@
 import { merge, partition, share } from 'rxjs';
 
-export const pipeWhen = (condition, fn) => source => {
-  const [success, fail] = partition(source.pipe(share()), condition);
-  return merge(fn(success), fail);
+export const pipeWhen = (condition, ...operators) => {
+  return source => {
+    const [success, fail] = partition(source.pipe(share()), condition);
+    return merge(success.pipe(...operators), fail);
+  };
 };

packages/operators/src/when.test.js

@@ -32,7 +32,7 @@ describe('when', () => {
         log('operators:when:pipe:input'),
         pipeWhen(
           v => !(v % 2),
-          s => s.pipe(map(v => v * v))
+          map(v => v * v)
         ),
         log('operators:when:pipe:output')
       );