fix(rxjs/webSocket): No longer tries to send unsub when the socket source disconnects

Author: benlesh

packages/rxjs/spec/observables/dom/webSocket-spec.ts

@@ -669,7 +669,7 @@ describe('webSocket', () => {
       });
       socket.triggerClose({ wasClean: true });
 
-      expect(results).to.deep.equal(['A next', 'A unsub', 'B next', 'B next', 'B next', 'B unsub', 'B complete']);
+      expect(results).to.deep.equal(['A next', 'A unsub', 'B next', 'B next', 'B next', 'B complete']);
     });
 
     it('should not close the socket until all subscriptions complete', () => {

packages/rxjs/src/internal/observable/dom/WebSocketSubject.ts

@@ -1,4 +1,4 @@
-import type { Subscriber, Subscription} from '../../Observable.js';
+import type { Subscriber, Subscription } from '../../Observable.js';
 import { Observable, operate } from '../../Observable.js';
 import type { NextObserver } from '../../types.js';
 
@@ -219,8 +219,13 @@ export class WebSocketSubject<In, Out = In> extends Observable<Out> {
   multiplex(subMsg: () => In, unsubMsg: () => In, messageFilter: (value: Out) => boolean) {
     return new Observable<Out>((destination) => {
       this.next(subMsg());
+
+      let isUnsub = true;
+
       destination.add(() => {
-        this.next(unsubMsg());
+        if (isUnsub) {
+          this.next(unsubMsg());
+        }
       });
       this.subscribe(
         operate({
@@ -230,6 +235,14 @@ export class WebSocketSubject<In, Out = In> extends Observable<Out> {
               destination.next(x);
             }
           },
+          error: (err) => {
+            isUnsub = false;
+            destination.error(err);
+          },
+          complete: () => {
+            isUnsub = false;
+            destination.complete();
+          },
         })
       );
     });