fix(rxjs): WebSocketSubject multiplex no longer tries to send unsubsc…

…ribe messages to closed sockets
ReactiveX · Feb 29, 2024 · 9a05f17 · 9a05f17
1 parent 63fff0f
commit 9a05f17
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 3 deletions.
diff --git a/packages/rxjs/spec/observables/dom/webSocket-spec.ts b/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', () => {

diff --git a/packages/rxjs/src/internal/observable/dom/WebSocketSubject.ts b/packages/rxjs/src/internal/observable/dom/WebSocketSubject.ts
@@ -1,4 +1,4 @@
-import type { Subscriber, Subscription} from '@rxjs/observable';
+import type { Subscriber, Subscription } from '@rxjs/observable';
 import { Observable, operate } from '@rxjs/observable';
 import type { NextObserver } from '../../types.js';
 
@@ -219,8 +219,11 @@ 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 +233,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();
+          },
         })
       );
     });