ing

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@overflow/rpc-js",
-  "version": "0.0.4",
+  "version": "0.0.5",
   "description": "TypeScript library setup for multiple compilation targets using tsc and webpack",
   "main": "./bundles/index.umd.js",
   "module": "./esm5/index.js",

src/client/client_rwc.ts

@@ -4,7 +4,7 @@ import { Message } from '../core/type';
 export interface ClientRWC {
   connect(queryString?: string): void;
   disconnect(): void;
-  connectionStatus(): Observable<boolean> | undefined;
+  connectionStatus(): Observable<boolean>;
 
   read(): Observable<Message>;
   write(data: Message): void;

src/client/rwc/ws/websocket_rwc.ts

@@ -17,7 +17,7 @@ export interface WebSocketClientRWCConfig {
 }
 
 export class WebSocketClientRWC implements ClientRWC {
-  private wsSubject: RxWebSocketSubject<Message> | undefined;
+  private wsSubject: RxWebSocketSubject<Message>;
   private _rxConfig: RxWebSocketSubjectConfig<Message>;
   private resSubject: Subject<Message> | undefined;
 
@@ -29,15 +29,8 @@ export class WebSocketClientRWC implements ClientRWC {
     this._rxConfig.serializer = (value: Message) => value;
     this._rxConfig.deserializer = (e: MessageEvent) => e.data;
     this._rxConfig.binaryType = 'arraybuffer';
-  }
 
-  public connect(queryString?: string): void {
-    if (undefined === this.wsSubject) {
-      if (undefined !== queryString) {
-        this._rxConfig.queryString = queryString;
-      }
-      this.wsSubject = new RxWebSocketSubject(this._rxConfig);
-    }
+    this.wsSubject = new RxWebSocketSubject(this._rxConfig);
     this.wsSubject.subscribe(
       (value: Message) => {
         if (undefined !== this.resSubject) {
@@ -50,23 +43,37 @@ export class WebSocketClientRWC implements ClientRWC {
         }
       },
       () => {
-        console.log('sss');
+        console.log('Cannot connect to server');
       },
     );
   }
 
-  public disconnect(): void {
-    if (undefined !== this.wsSubject) {
-      this.wsSubject.disconnect();
+  public connect(queryString?: string): void {
+    if (undefined !== queryString) {
+      this.wsSubject.rxConfig.queryString = queryString;
     }
+
+    if (this.wsSubject.isConnected()) {
+      console.log('connected already');
+
+      return;
+    }
+
+    this.wsSubject.connect();
   }
 
-  public connectionStatus(): Observable<boolean> | undefined {
-    if (undefined !== this.wsSubject) {
-      return this.wsSubject.connectionStatus;
+  public disconnect(): void {
+    if (!this.wsSubject.isConnected()) {
+      console.log('not connected');
+
+      return;
     }
 
-    return undefined;
+    this.wsSubject.disconnect();
+  }
+
+  public connectionStatus(): Observable<boolean> {
+    return this.wsSubject.connectionStatus;
   }
 
   public read(): Observable<Message> {
@@ -78,8 +85,6 @@ export class WebSocketClientRWC implements ClientRWC {
   }
 
   public write(data: Message): void {
-    if (undefined !== this.wsSubject) {
-      this.wsSubject.send(data);
-    }
+    this.wsSubject.send(data);
   }
 }

src/dom/rx_websocket_subject.ts

@@ -17,19 +17,26 @@ export class RxWebSocketSubject<T> extends Subject<T> {
   private connectionObserver: Observer<boolean> | undefined;
   public connectionStatus: Observable<boolean>;
 
+  public rxConfig: RxWebSocketSubjectConfig<T>;
+
   private readonly reconnectInterval: number;
   private readonly reconnectRetry: number;
 
-  constructor(private _rxConfig: RxWebSocketSubjectConfig<T>) {
+  private _isConnected: boolean;
+
+  constructor(_rxConfig: RxWebSocketSubjectConfig<T>) {
     super();
 
+    this.rxConfig = _rxConfig;
+    this._isConnected = false;
+
     this.reconnectInterval =
-      undefined !== this._rxConfig.reconnectInterval
-        ? this._rxConfig.reconnectInterval
+      undefined !== this.rxConfig.reconnectInterval
+        ? this.rxConfig.reconnectInterval
         : 5000;
     this.reconnectRetry =
-      undefined !== this._rxConfig.reconnectRetry
-        ? this._rxConfig.reconnectRetry
+      undefined !== this.rxConfig.reconnectRetry
+        ? this.rxConfig.reconnectRetry
         : 10;
 
     this.connectionStatus = new Observable((observer: Observer<boolean>) => {
@@ -39,9 +46,10 @@ export class RxWebSocketSubject<T> extends Subject<T> {
       distinctUntilChanged(),
     );
 
-    this.wsSubjectConfig = Object.assign({}, this._rxConfig);
+    this.wsSubjectConfig = Object.assign({}, this.rxConfig);
     this.wsSubjectConfig.closeObserver = {
       next: (_e: CloseEvent) => {
+        this._isConnected = false;
         this.socket = undefined;
         if (undefined !== this.connectionObserver) {
           this.connectionObserver.next(false);
@@ -51,12 +59,12 @@ export class RxWebSocketSubject<T> extends Subject<T> {
     this.wsSubjectConfig.openObserver = {
       next: (_e: Event) => {
         if (undefined !== this.connectionObserver) {
+          this._isConnected = true;
           this.connectionObserver.next(true);
         }
       },
     };
 
-    this.connect();
     this.connectionStatus.subscribe(isConnected => {
       if (
         !this.reconnectionObservable &&
@@ -67,11 +75,11 @@ export class RxWebSocketSubject<T> extends Subject<T> {
     });
   }
 
-  private connect(): void {
+  public connect(): void {
     this.wsSubjectConfig.url =
-      undefined !== this._rxConfig.queryString
-        ? urljoin(this._rxConfig.url, this._rxConfig.queryString)
-        : this._rxConfig.url;
+      undefined !== this.rxConfig.queryString
+        ? urljoin(this.rxConfig.url, this.rxConfig.queryString)
+        : this.rxConfig.url;
     this.socket = new WebSocketSubject(this.wsSubjectConfig);
     this.socket.subscribe(
       m => {
@@ -85,6 +93,13 @@ export class RxWebSocketSubject<T> extends Subject<T> {
     );
   }
 
+  /**
+   * isConnected
+   */
+  public isConnected(): boolean {
+    return this._isConnected;
+  }
+
   private reconnect(): void {
     this.reconnectionObservable = interval(this.reconnectInterval).pipe(
       takeWhile((_v, index) => {