RxJS v7中如何扩展ReplaySubject实现subscribe触发回调？

问题背景 #

想要在Observable类实例被调用subscribe时触发自定义回调，最初尝试继承ReplaySubject并重写subscribe方法，但在RxJS v7+ TypeScript环境下出现类型不匹配错误：

Property 'subscribe' in type 'CallbackReplaySubject<T>' is not assignable to the same property in base type 'ReplaySubject<T>'.
  Type '(observer: Partial<Observer<T>>) => Unsubscribable' is not assignable to type '{ (observerOrNext?: Partial<Observer<T>> | ((value: T) => void) | undefined): Subscription; (next?: ((value: T) => void) | null | undefined, error?: ((error: any) => void) | ... 1 more ... | undefined, complete?: (() => void) | ... 1 more ... | undefined): Subscription; }'.
    Type 'Unsubscribable' is missing the following properties from type 'Subscription': closed, _parentage, _finalizers, add, and 4 more.

同时RxJS禁止修改下划线标记的内部方法，此前v6版本通过重写_subscribe的方案也不再适用，委托模式因Observable是类而非接口难以直接实现。

可行方案 #

方案1：封装委托类（完全兼容Subject接口） #

通过封装ReplaySubject实例，转发所有公共方法，在自定义的subscribe中先触发回调再调用原实例的subscribe，确保类型完全匹配：

import { ReplaySubject, Observable, Observer, Subscription } from 'rxjs';

export class CallbackReplaySubject<T> {
  private readonly subject: ReplaySubject<T>;

  constructor(bufferSize = 1, windowTime?: number) {
    this.subject = new ReplaySubject(bufferSize, windowTime);
  }

  next(value: T): void {
    this.subject.next(value);
  }

  error(err: any): void {
    this.subject.error(err);
  }

  complete(): void {
    this.subject.complete();
  }

  subscribe(
    observerOrNext?: Partial<Observer<T>> | ((value: T) => void),
    error?: ((err: any) => void) | null,
    complete?: (() => void) | null
  ): Subscription {
    this.triggerSubscriptionCallback();
    return this.subject.subscribe(observerOrNext, error, complete);
  }

  asObservable(): Observable<T> {
    return new Observable<T>(observer => {
      this.triggerSubscriptionCallback();
      return this.subject.subscribe(observer);
    });
  }

  private triggerSubscriptionCallback(): void {
    // 这里编写订阅触发时的自定义逻辑
    console.log('新订阅已触发');
  }
}

优势：完全保留ReplaySubject的所有公共API，外部使用方式与原生Subject完全一致，类型安全。

方案2：利用defer操作符轻量化实现 #

如果不需要扩展Subject的next/error/complete等方法，可使用defer操作符在每次订阅时执行回调，返回目标ReplaySubject：

import { ReplaySubject, defer, Observable } from 'rxjs';

function createCallbackReplaySubject<T>(bufferSize = 1, windowTime?: number): Observable<T> {
  const subject = new ReplaySubject<T>(bufferSize, windowTime);

  return defer(() => {
    triggerSubscriptionCallback();
    return subject;
  });

  function triggerSubscriptionCallback() {
    // 订阅触发逻辑
    console.log('新订阅已触发');
  }
}

// 使用示例
const mySubject = createCallbackReplaySubject<number>();
mySubject.subscribe(val => console.log(val));
mySubject.next(123);

优势：无需创建自定义类，代码更简洁，适合仅需监听订阅事件、不需要直接操作Subject的场景。

方案说明 #

• 原继承方案失败的核心原因：RxJS v7中subscribe方法有多重重载签名，且返回值为Subscription而非Unsubscribable，重写时未覆盖所有重载导致类型不匹配。
• RxJS官方不建议重写subscribe或修改内部下划线方法，上述方案均遵循官方设计规范，避免依赖内部实现细节。

内容的提问来源于stack exchange，提问作者netmikey