Allow replaySubject to unsubscribe (#373)

* Allow replaySubject to unsubscribe

* Cleanup types, cache observable is not a subject,

* ReplaySubject now unused

* Cleanup CodeClimate

* Cleanup helpers

packages/rpc-rx/src/cached.spec.js

@@ -75,17 +75,4 @@ describe('createCachedObservable', () => {
       observable2
     ).not.toEqual(observable1);
   });
-
-  it('functions as an subject', (done) => {
-    const subject = creator(123);
-
-    subject.subscribe((value) => {
-      if (value) {
-        expect(value).toEqual('test');
-        done();
-      }
-    });
-
-    subject.next('test');
-  });
 });

packages/rpc-rx/src/index.ts

@@ -2,18 +2,19 @@
 // This software may be modified and distributed under the terms
 // of the ISC license. See the LICENSE file for details.
 
-import { RpcInterface$Section } from '@polkadot/rpc-core/types';
+import { RpcInterface$Section, RpcInterface$Section$Method } from '@polkadot/rpc-core/types';
 import { ProviderInterface } from '@polkadot/rpc-provider/types';
 import { RpcRxInterface, RpcRxInterface$Events, RpcRxInterface$Section } from './types';
 
 import EventEmitter from 'eventemitter3';
-import { BehaviorSubject, ReplaySubject, Observable, Subscriber, from } from 'rxjs';
+import { BehaviorSubject, Observable, Subscriber, from } from 'rxjs';
+import { publishReplay, refCount } from 'rxjs/operators';
 import Rpc from '@polkadot/rpc-core/index';
 import { isFunction, isUndefined } from '@polkadot/util';
 
 type CachedMap = {
   [index: string]: {
-    [index: string]: ReplaySubject<any>
+    [index: string]: Observable<any>
   }
 };
 
@@ -99,7 +100,7 @@ export default class RpcRx implements RpcRxInterface {
       }, ({} as RpcRxInterface$Section));
   }
 
-  private createObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array<any>) => Observable<any> | ReplaySubject<any> {
+  private createObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array<any>) => Observable<any> {
     if (isFunction(section[name].unsubscribe)) {
       return this.createCachedObservable(subName, name, section);
     }
@@ -114,43 +115,30 @@ export default class RpcRx implements RpcRxInterface {
       );
   }
 
-  private createCachedObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array<any>) => ReplaySubject<any> {
+  private createCachedObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array<any>) => Observable<any> {
     if (!this._cacheMap[subName]) {
       this._cacheMap[subName] = {};
     }
 
-    return (...params: Array<any>): ReplaySubject<any> => {
+    return (...params: Array<any>): Observable<any> => {
       const paramStr = JSON.stringify(params);
 
       if (!this._cacheMap[subName][paramStr]) {
-        this._cacheMap[subName][paramStr] = this.createSubject(name, params, section);
+        this._cacheMap[subName][paramStr] = this.createReplay(name, params, section, subName, paramStr);
       }
 
       return this._cacheMap[subName][paramStr];
     };
   }
 
-  private createSubject (name: string, params: Array<any>, section: RpcInterface$Section, unsubCallback?: () => void): ReplaySubject<any> {
-    const subject = new ReplaySubject(1);
-
-    Observable
+  private createReplay (name: string, params: Array<any>, section: RpcInterface$Section, subName: string, paramStr: string): Observable<any> {
+    return Observable
       .create((observer: Subscriber<any>): Function => {
         try {
           const fn = section[name];
-          const subscribe = fn(...params, this.createSubjectCallback(observer));
+          const subscribe = fn(...params, this.createReplayCallback(observer));
 
-          return (): void => {
-            subscribe
-              .then((subscriptionId: number) =>
-                fn.unsubscribe(subscriptionId)
-              )
-              .then(() =>
-                isFunction(unsubCallback) && unsubCallback()
-              )
-              .catch((error) => {
-                console.error('Unsubscribe failed', error);
-              });
-          };
+          return this.createReplayUnsub(fn, subscribe, subName, paramStr);
         } catch (error) {
           console.error(error);
 
@@ -159,12 +147,13 @@ export default class RpcRx implements RpcRxInterface {
           };
         }
       })
-      .subscribe(subject);
-
-    return subject;
+      .pipe(
+        publishReplay(1),
+        refCount()
+      );
   }
 
-  private createSubjectCallback (observer: Subscriber<any>) {
+  private createReplayCallback (observer: Subscriber<any>) {
     let cachedResult: any;
 
     return (result: any) => {
@@ -182,4 +171,19 @@ export default class RpcRx implements RpcRxInterface {
       observer.next(cachedResult);
     };
   }
+
+  private createReplayUnsub (fn: RpcInterface$Section$Method, subscribe: Promise<number>, subName: string, paramStr: string): () => void {
+    return (): void => {
+      subscribe
+        .then((subscriptionId: number) =>
+          fn.unsubscribe(subscriptionId)
+        )
+        .then(() => {
+          delete this._cacheMap[subName][paramStr];
+        })
+        .catch((error) => {
+          console.error('Unsubscribe failed', error);
+        });
+    };
+  }
 }

packages/rpc-rx/src/replay.js

@@ -10,12 +10,11 @@ jest.mock('@polkadot/rpc-provider/ws', () => class {
 
 const RpcRx = require('./index').default;
 
-describe('subject', () => {
+describe('replay', () => {
   const params = [123, false];
   let api;
   let update;
   let section;
-  let subscription;
   let observable;
 
   beforeEach(() => {
@@ -37,13 +36,7 @@ describe('subject', () => {
       subMethod
     };
 
-    observable = api.createSubject('subMethod', params, section);
-  });
-
-  afterEach(() => {
-    if (subscription) {
-      subscription.unsubscribe();
-    }
+    observable = api.createReplay('subMethod', params, section);
   });
 
   it('subscribes via the api section', (done) => {
@@ -61,7 +54,7 @@ describe('subject', () => {
   });
 
   it('returns the observable value', (done) => {
-    subscription = observable.subscribe((value) => {
+    observable.subscribe((value) => {
       if (value) {
         expect(value).toEqual('test');
         done();
@@ -70,4 +63,18 @@ describe('subject', () => {
 
     update('test');
   });
+
+  it('unsubscribes as required', (done) => {
+    section.subMethod.unsubscribe = async () => {
+      done();
+    };
+
+    let subscription = observable.subscribe((value) => {
+      if (value) {
+        subscription.unsubscribe();
+      }
+    });
+
+    update('test');
+  });
 });