Merge pull request #123 from scorebreaker/memory-leak

fix: cleanup getXudClient$

Author: Karl Ranna

src/opendex/xud/client.spec.ts

@@ -2,23 +2,30 @@ import { XudClient } from '../../proto/xudrpc_grpc_pb';
 import { testConfig } from '../../test-utils';
 import { errors } from '../errors';
 import { getXudClient$ } from './client';
+import { take } from 'rxjs/operators';
 
 jest.mock('../../proto/xudrpc_grpc_pb');
 jest.mock('../../proto/xudrpc_pb');
 
 describe('getXudClient$', () => {
   test('success', done => {
-    expect.assertions(2);
+    expect.assertions(3);
     const config = testConfig();
     const xudClient$ = getXudClient$(config);
-    xudClient$.subscribe(() => {
-      expect(XudClient).toHaveBeenCalledTimes(1);
-      expect(XudClient).toHaveBeenCalledWith(
-        `${config.OPENDEX_RPC_HOST}:${config.OPENDEX_RPC_PORT}`,
-        expect.any(Object),
-        expect.any(Object)
-      );
-      done();
+
+    xudClient$.pipe(take(1)).subscribe({
+      next: client => {
+        expect(XudClient).toHaveBeenCalledTimes(1);
+        expect(XudClient).toHaveBeenCalledWith(
+          `${config.OPENDEX_RPC_HOST}:${config.OPENDEX_RPC_PORT}`,
+          expect.any(Object),
+          expect.any(Object)
+        );
+        setImmediate(() => {
+          expect(client.close).toHaveBeenCalledTimes(1);
+          done();
+        });
+      },
     });
   });
 

src/opendex/xud/client.ts

@@ -20,7 +20,9 @@ const getXudClient$ = (config: Config): Observable<XudClient> => {
       options
     );
     subscriber.next(client);
-    subscriber.complete();
+    return () => {
+      client.close();
+    };
   }).pipe(
     catchError(error => {
       if (error.code === 'ENOENT') {

src/utils.ts

@@ -1,5 +1,5 @@
 import moment from 'moment';
-import { Observable, Subject, timer, merge } from 'rxjs';
+import { Observable, Subject } from 'rxjs';
 import { take, tap } from 'rxjs/operators';
 import { curry } from 'ramda';
 
@@ -24,12 +24,7 @@ export const getStartShutdown$ = (): Observable<unknown> => {
   const shutdown$ = new Subject();
   process.on('SIGINT', () => shutdown$.next());
   process.on('SIGTERM', () => shutdown$.next());
-  const ONE_MINUTE = 1000 * 60;
-  const ONE_HOUR = ONE_MINUTE * 60;
-  const restart$ = timer(ONE_HOUR).pipe(
-    tap(() => console.log('Restarting Arby to reduce memory usage.'))
-  );
-  return merge(shutdown$.asObservable(), restart$).pipe(take(1));
+  return shutdown$.asObservable().pipe(take(1));
 };
 
 const debugObservable = (prefix: string, source: Observable<any>) => {