Remove duplicte notification (#768)

* Remove duplicte notification

* on listener close, capture the unsolicited close event

Author: rg911

src/infrastructure/Listener.ts

@@ -15,7 +15,7 @@
  */
 
 import { Observable, of, Subject } from 'rxjs';
-import { catchError, filter, map, mergeMap, share, switchMap } from 'rxjs/operators';
+import { catchError, distinctUntilChanged, filter, map, mergeMap, share, switchMap } from 'rxjs/operators';
 import { BlockInfoDTO } from 'symbol-openapi-typescript-fetch-client';
 import * as WebSocket from 'ws';
 import { UnresolvedAddress } from '../model';
@@ -74,6 +74,8 @@ export class Listener implements IListener {
      */
     private uid: string;
 
+    private SIGINT = false;
+
     /**
      * Constructor
      * @param url - Listener websocket server url. default: rest-gateway's url with ''/ws'' suffix. (e.g. http://localhost:3000/ws).
@@ -120,6 +122,18 @@ export class Listener implements IListener {
                 this.webSocket.onerror = (err: Error): void => {
                     reject(err);
                 };
+                this.webSocket.onclose = (closeEvent?: any): void => {
+                    if (this.SIGINT) {
+                        return;
+                    }
+                    if (closeEvent) {
+                        reject({
+                            client: this.uid,
+                            code: closeEvent.code,
+                            reason: closeEvent.reason,
+                        });
+                    }
+                };
                 this.webSocket.onmessage = (msg: any): void => {
                     const message = JSON.parse(msg.data as string);
                     this.handleMessage(message, resolve);
@@ -232,6 +246,7 @@ export class Listener implements IListener {
             this.webSocket &&
             (this.webSocket.readyState === this.webSocket.OPEN || this.webSocket.readyState === this.webSocket.CONNECTING)
         ) {
+            this.SIGINT = true;
             this.webSocket.close();
         }
     }
@@ -339,6 +354,11 @@ export class Listener implements IListener {
                 return this.messageSubject.asObservable().pipe(
                     filter((listenerMessage) => listenerMessage.channelName === channel),
                     filter((listenerMessage) => listenerMessage.message instanceof Transaction),
+                    distinctUntilChanged((prev, curr) => {
+                        const currentHash = (curr.message as Transaction).transactionInfo!.hash;
+                        const previousHash = (prev.message as Transaction).transactionInfo!.hash;
+                        return (currentHash && previousHash && previousHash === currentHash) || !currentHash || !previousHash;
+                    }),
                     switchMap((_) => {
                         const transactionObservable = of(_.message as T).pipe(
                             filter((transaction) => this.filterHash(transaction, transactionHash)),
@@ -421,7 +441,8 @@ export class Listener implements IListener {
                     filter((_) => typeof _.message === 'string'),
                     filter((_) => subscribers.includes(_.channelParam.toUpperCase())),
                     map((_) => _.message as string),
-                    filter((_) => !transactionHash || _.toUpperCase() == transactionHash.toUpperCase()),
+                    filter((_) => !transactionHash || _.toUpperCase() === transactionHash.toUpperCase()),
+                    distinctUntilChanged(),
                 );
             }),
         );