refactor(WebSocketSubject): directly inherit Subject

ReactiveX · Nov 22, 2023 · ae005b4 · ae005b4
1 parent f6e54f0
commit ae005b4
Showing 1 changed file with 18 additions and 2 deletions.
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 { Subject, AnonymousSubject } from '../../Subject.js';
+import { Subject } from '../../Subject.js';
 import { Subscriber, Observable, Subscription, operate } from '../../Observable.js';
 import { ReplaySubject } from '../../ReplaySubject.js';
 import { Observer, NextObserver } from '../../types.js';
@@ -151,14 +151,18 @@ const WEBSOCKETSUBJECT_INVALID_ERROR_OBJECT =
 
 export type WebSocketMessage = string | ArrayBuffer | Blob | ArrayBufferView;
 
-export class WebSocketSubject<T> extends AnonymousSubject<T> {
+export class WebSocketSubject<T> extends Subject<T> {
   private _config!: WebSocketSubjectConfig<T>;
 
   /** @internal */
   _output!: Subject<T>;
 
   private _socket: WebSocket | null = null;
 
+  private destination: Observer<T> | undefined = undefined;
+
+  private _source: Observable<T> | undefined = undefined;
+
   constructor(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) {
     super();
     if (urlConfigOrSource instanceof Observable) {
@@ -334,6 +338,18 @@ export class WebSocketSubject<T> extends AnonymousSubject<T> {
     };
   }
 
+  next(value: T) {
+    this.destination?.next?.(value);
+  }
+
+  error(err: any) {
+    this.destination?.error?.(err);
+  }
+
+  complete() {
+    this.destination?.complete?.();
+  }
+
   /** @internal */
   protected _subscribe(subscriber: Subscriber<T>): Subscription {
     const { _source } = this;