Swap internal derives to use api.derive.* (#1553)

* Swap internal derives to use api.derive.*

* Rx vs normal decorate

* Small cleanups

* Res-use sessions

* Cleanup types

* Fix refCount lost subs

* Small cleanups

packages/rpc-core/src/rxjs/drr.ts

@@ -33,7 +33,7 @@ const NOOP = (): void => {};
  *
  * @ignore
  */
-export const drr = ({ skipChange, skipTimeout }: Options = {}): DrrResult => <T> (source$: Observable<T>): Observable<T> =>
+export const drr = ({ skipChange = false, skipTimeout = false }: Options = {}): DrrResult => <T> (source$: Observable<T>): Observable<T> =>
   source$.pipe(
     catchError(ERR),
     skipChange

packages/rpc-core/src/rxjs/refCountDelay.ts

@@ -2,54 +2,47 @@
 // This software may be modified and distributed under the terms
 // of the Apache-2.0 license. See the LICENSE file for details.
 
-// From https://medium.com/@volkeron/rxjs-unsubscribe-delay-218a9ab2672e (adapted for rxjs 6)
-// v6 inspiration from https://github.com/cartant/rxjs-etc/blob/7d7cd21233f67c364fd246ef72077bbe494dd5c2/source/operators/refCountDelay.ts
-
-import { asapScheduler, ConnectableObservable, MonoTypeOperatorFunction, NEVER, Observable, Subject, Subscription, timer, using } from 'rxjs';
-import { scan, switchMap, tap } from 'rxjs/operators';
+import { asapScheduler, ConnectableObservable, MonoTypeOperatorFunction, Observable, Subscription } from 'rxjs';
 
 const DELAY = 1750;
 
-// subscribe to a source, tracker and increment the counter
-function subscribe <T> (source: Observable<T>, counter: Subject<number>, tracker: Observable<number>): [Subscription, Subscription] {
-  const sourceConn = (source as ConnectableObservable<T>).connect();
-  const trackerConn = tracker.subscribe();
+function refCountDelayInner <T> (source: Observable<T>): Observable<T> {
+  let connected = 0; // 0 = disconnected, 1 = disconnecting, 2 = connecting, 3 = connected
+  let refCount = 0;
+  let con = Subscription.EMPTY;
+  let sched = Subscription.EMPTY;
 
-  counter.next(1);
+  return new Observable((ob) => {
+    source.subscribe(ob);
 
-  return [sourceConn, trackerConn];
+    if (refCount++ === 0) {
+      if (connected === 1) {
+        connected = 3;
+        sched.unsubscribe();
+      } else {
+        con = (source as ConnectableObservable<T>).connect();
+        connected = 3;
+      }
+    }
+
+    return (): void => {
+      if (--refCount === 0) {
+        if (connected === 2) {
+          connected = 0;
+          sched.unsubscribe();
+        } else {
+          // connected === 3
+          connected = 1;
+          sched = asapScheduler.schedule((): void => {
+            con.unsubscribe();
+            connected = 0;
+          }, DELAY);
+        }
+      }
+    };
+  });
 }
 
-// unsubscribe from both the tracker and the source
-function unsubscribe (sourceConn: Subscription | undefined, trackerConn: Subscription | undefined): void {
-  sourceConn && sourceConn.unsubscribe();
-  trackerConn && trackerConn.unsubscribe();
-}
-
-// keep track of references, but only unsubscribe after the configured delay
 export function refCountDelay <T> (): MonoTypeOperatorFunction<T> {
-  return (source: Observable<T>): Observable<T> => {
-    let sourceConn: Subscription | undefined;
-    let trackerConn: Subscription | undefined;
-    const counter = new Subject<number>();
-    const tracker = counter.pipe(
-      scan((total, change) => change + total, 0),
-      switchMap((count) =>
-        count === 0 // when we have a zero count, schedule a unsubscribe
-          ? timer(DELAY, asapScheduler).pipe(tap((): void => unsubscribe(sourceConn, trackerConn)))
-          : NEVER
-      )
-    );
-
-    return using(
-      () => {
-        [sourceConn, trackerConn] = subscribe(source, counter, tracker);
-
-        return {
-          unsubscribe: (): void => counter.next(-1)
-        };
-      },
-      (): Observable<T> => source
-    );
-  };
+  return refCountDelayInner;
 }