WebSocket created during close of a previous one is never closed

[ChristofferGersen]

Bug Report

Current Behavior
When there are no more subscriptions to a WebSocketSubject, then the underlying WebSocket is closed. If you then create a new subscription on the WebSocketSubject, before the onclose of the previous WebSocket has fired, then the newly created WebSocket will never be closed.

If you combine this with the multiplex function, then some subscribe/unsubscribe messages will not be send as well.

Reproduction

1. Create WebSocketSubject
2. Create two subscriptions on WebSocketSubject
3. Wait a second
4. Unsubscribe both subscriptions and immediately create two new ones
5. Repeat steps 3 and 4
6. Unsubscribe both subscriptions

https://stackblitz.com/edit/typescript-n7jv6k

In case the stackblitz does not work, here is the code it contains:

Show imports and utility class

import { empty, interval, merge } from "rxjs";
import { webSocket } from "rxjs/webSocket";
import { switchMap, take } from "rxjs/operators";

function log(message: string) {
  console.log(message);
}

function error(err: any) {
  console.error(err);
}

class LoggingWebSocket implements WebSocket {
  private static nextSocketId = 0;
  readonly socketId = LoggingWebSocket.nextSocketId++;
  readonly CONNECTING: number = 0;
  readonly OPEN: number = 1;
  readonly CLOSING: number = 2;
  readonly CLOSED: number = 3;

  binaryType = "blob";
  readonly bufferedAmount: number = 0;
  readonly extensions: string = "";
  onclose: ((this: WebSocket, ev: CloseEvent) => any) | null = null;
  onerror: ((this: WebSocket, ev: Event) => any) | null = null;
  onmessage: ((this: WebSocket, ev: MessageEvent) => any) | null = null;
  onopen: ((this: WebSocket, ev: Event) => any) | null = null;
  readonly protocol: string;
  readyState: number = this.CONNECTING;
  constructor(
    public readonly url: string,
    public protocols: string | string[] = ""
  ) {
    this.protocol = Array.isArray(protocols) ? protocols[0] || "" : protocols;
    log("Created socket " + this.socketId);
    setTimeout(() => {
      this.readyState = this.OPEN;
      log("Opened socket " + this.socketId);
      this.dispatchEvent({ type: "open" });
    }, 100);
  }
  close(code?: number | undefined, reason?: string | undefined): void {
    this.readyState = this.CLOSING;
    log("Closing socket " + this.socketId);
    setTimeout(() => {
      this.readyState = this.CLOSED;
      log("Closed socket " + this.socketId);
      this.dispatchEvent({ type: "close", wasClean: true, code, reason });
    }, 100);
  }
  send(data: string | ArrayBuffer | Blob | ArrayBufferView): void {
    if (this.readyState === this.OPEN) {
      log("Socket " + this.socketId + " sends: " + data);
    } else {
      throw new Error(
        "Socket unable to send, because readyState=" + this.readyState
      );
    }
  }
  addEventListener(type: any, listener: any, options?: any) {
    throw new Error("Method not implemented.");
  }
  removeEventListener(type: any, listener: any, options?: any) {
    throw new Error("Method not implemented.");
  }
  dispatchEvent(evt: { type: string; [x: string]: any }): boolean {
    const handler = (this as any)["on" + evt.type];
    if (handler) {
      evt.target = this;
      handler.call(this, evt);
    }
    return true;
  }
}

const socket = webSocket<any>({
  url: "ws://localhost/ws",
  WebSocketCtor: LoggingWebSocket
});

function multiplex(id: number, index: number) {
  return socket.multiplex(
    () => "sub" + id + "." + index,
    () => "unsub" + id + "." + index,
    value => value === id + "." + index
  );
}

interval(1000)
  .pipe(
    take(4),
    switchMap(id => {
      if (id < 3) {
        return merge(multiplex(id, 1), multiplex(id, 2));
      }
      return empty();
    })
  )
  .subscribe(
    () => {},
    err => error(err),
    () => log("Complete")
  );

I would expect the log to show the following, but the lines with the strike-through are missing:

Created socket 0
Opened socket 0
Socket 0 sends: "sub0.1"
Socket 0 sends: "sub0.2"
Socket 0 sends: "unsub0.1"
Socket 0 sends: "unsub0.2"
Closing socket 0
Created socket 1
Closed socket 0
Opened socket 1
Socket 1 sends: "sub1.1"
Socket 1 sends: "sub1.2"
Socket 1 sends: "unsub1.1"
Socket 1 sends: "unsub1.2"
Closing socket 1
Created socket 2
Closed socket 1
Opened socket 2
Socket 2 sends: "sub2.1"
Socket 2 sends: "sub2.2"
Socket 2 sends: "unsub2.1"
Socket 2 sends: "unsub2.2"
Closing socket 2
Complete
Closed socket 2

Expected behavior
I expect all websockets to be closed in the end, and that all subscribe/unsubscribe messages are send.

Environment

• Runtime: Chrome v67.0.3396.99
• RxJS version: 6.2.2

Possible Solution
I did a little digging, and the problem is caused by _resetState being called twice within WebSocketSubject, when closing the first websocket. The first time it is called from the last unsubscribe (anonymous function inside _subscribe), and a second time from socket.onclose. The problem with that is that the second websocket has already been created, when the onclose of the first websocket fires. One possible solution is to add the following check around the _resetState call inside socket.onclose:

if (this._socket === e.target) {
  this._resetState();
}

[maknapp]

@benlesh This issue can be closed. (I used the wrong format to auto-close in the commit.)