如何构建Observable包装类，在每个发射值前后执行自定义操作？

我看了你写的初始代码，核心需求是给Observable的每个发射值加上前置（BEFORE）和后置（AFTER）操作对吧？你的思路方向是对的，但有几个关键问题没打通——比如数据源没关联、操作顺序的逻辑没串起来。我给你两种可行的实现方案，按需选择：

方案1：直接Pipe链式处理（最简洁） #

如果不需要额外的多播或中间控制逻辑，直接在原始Observable的pipe里添加两个tap操作就行，顺序很重要：

import { Observable, tap } from 'rxjs';

class Wrapper<T> {
  data$: Observable<T>;

  constructor(wrapped$: Observable<T>) {
    this.data$ = wrapped$.pipe(
      // 每个值被订阅者接收**之前**执行前置操作
      tap(_ => console.log("BEFORE")),
      // 每个值被订阅者接收**之后**执行后置操作
      tap({
        next: _ => console.log("AFTER")
      })
    );
  }
}

测试一下效果：

import { Subject } from 'rxjs';

const source$ = new Subject<string>();
const wrapper = new Wrapper(source$);

// 订阅包装后的Observable
wrapper.data$.subscribe(val => console.log(`收到值：${val}`));

source$.next("Hello RxJS");
// 输出顺序：
// BEFORE
// 收到值：Hello RxJS
// AFTER

方案2：用Subject做中间层（适合多播/额外控制） #

如果你需要让多个订阅者共享同一个数据流，或者要在中间层做额外的逻辑处理，可以用Subject作为中转：

import { Observable, Subject, tap } from 'rxjs';

class Wrapper<T> {
  private _dataSubject = new Subject<T>();
  // 对外暴露的数据流，先执行前置操作
  data$ = this._dataSubject.pipe(
    tap(_ => console.log("BEFORE"))
  );

  constructor(wrapped$: Observable<T>) {
    // 订阅原始Observable，把值传给Subject，然后执行后置操作
    wrapped$.pipe(
      tap(val => this._dataSubject.next(val)),
      tap(_ => console.log("AFTER"))
    ).subscribe();
  }
}

为什么你的初始代码没生效？ #

1. 你的wrapped$只是加了AFTER的tap，但没有订阅它，所以这个pipe里的逻辑永远不会执行；
2. _dataSubject没有被任何数据源触发，所以data$不会发射任何值；
3. 操作顺序的逻辑没有串联起来——前置操作应该在值到达订阅者前执行，后置在之后。

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