RxJS 寻找类似“delayUntil”的东西

Question

我有两个函数，一个返回observableA，一个返回observableB。我希望 getter 以及 observableB 的订阅延迟到 observableA 完成（但当我订阅 observableB，observableA 已经完成时，也可能是这种情况）。

我已经尝试使用pipe 和skipUntil，但不幸的是，如果observableA 尚未完成，这只会阻止执行并且不会延迟它。

functionA() {
  this.observableA$ = getObservableA()
  this.observableA$.subscribe(_ => {
    // A: This line should execute *before* line B
  })
}

functionB() {
  this.observableB$ = getObservableB()  // This getter should execute *after* line A
  this.observableB$.subscribe(_ => {
    // B: This line should execute *after* line A
  })
}

// Two functions are called independently
functionA()
functionB()

如果能找到一种非常 RxJS-ish 的方式会很棒:)

---

更新 1：concat 的问题： 如前所述，这两个函数都是独立调用的，这将导致在使用concat 时重复执行observableA$，就像建议的那样。 functionB 中的订阅也会执行两次我不想要的。

functionA() {
  this.observableA$ = getObservableA()
  this.observableA$.subscribe(_ => {
    // A: This line should execute *before* line B
  })
}

functionB() {
  Observable.concat(
    this.observableA$,  // I get executed once again :(
    Observable.defer(() => getObservableB())  // To prevent earlier execution
  ).subscribe(() => {
    // B: This line should execute *after* line A
    console.log("I'm logged twice, once for each observable :(")
  })
}

// Two functions are called independently
functionA()
functionB()

---

更新 2：@Wilhelm Olejnik 通过使用额外的 BehaviorSubject 解决了这个问题

Answer 1

我遇到了类似的情况，我创建了一个服务 SyncService 来同步 observables。对我来说，ObsA 和 ObsB 来自不同的组件。 ComponentA 在加载自己的数据后得到了ComponentB 需要的一些初始化数据。

在ComponentA我订阅了ObsA（从getInitData()返回）并调用了同步服务：

public initialize() {
   this.apiSvc.getInitData().subscribe((initData) => {
      this.data = initData;
      this.syncService.setA(initData);
   }
}

然后，在ComponentB订阅ObsB（从getBData()返回），然后订阅同步服务：

public loadBData() {
    this.apiSvc.getBData().subscribe((dataB) => {
        this.syncService.setB(dataB).subscribe((dataA, dataB) => {
            this.doStuffWithAAndB(dataA, dataB);
        }
    }
}

最后，同步服务如下所示：

@Injectable
export class SyncService {
    private dataA: DataA = null;
    private dataB: DataB = null;
    private gotAEvent: new EventEmitter<DataA>();

    public setA(dataA: DataA) {
        this.dataA = dataA;
        if (this.dataB != null) {
            // ObsB was already resolved!
            this.gotAEvent.emit(dataA);
        }
    }

    public setB(dataB: DataB) {
        this.dataB = dataB;
        if (this.dataA != null) {
            return of({this.dataA, this.dataB});
        } else {
            return this.gotAEvent().map((dataA: DataA) => {
                return {dataA, dataB};
            }
        }
    }
}

Answer 2

如果我理解正确，您希望延迟 getObservableB 的执行，直到某些 observable 被分配给属性 observableA$。

也许使用一些Proxy 技巧是可行的，但我认为将observableA$ 更改为空初始化BehaviorSubject 更容易。然后你可以观察observableA$并创建observableB$什么时候会发出非空信号。

https://stackblitz.com/edit/rxjs-mm2edy

import { of, BehaviorSubject, timer } from 'rxjs';
import { filter, switchMap, mapTo, tap } from 'rxjs/operators';

const getObservableA = () => timer(100).pipe(
  tap(() => console.log('getObservableA')),
  mapTo('A')
);

const getObservableB = () => timer(100).pipe(
  tap(() => console.log('getObservableB')),
  mapTo('B')
);

class Test {
  // init observableA$ as BehaviorSubject with null state
  observableA$ = new BehaviorSubject(null);
  observableB$;


  functionA() {
    getObservableA().subscribe(val => {
      console.log(val)
      this.observableA$.next(val);      // notify subscribers that observableA$ is ready
    });
  }

  functionB() {
    this.observableB$ = this.observableA$.pipe(
      filter(value => value !== null),            // ignore observableA$ until initalized
      switchMap(value => getObservableB())
    )
    this.observableB$.subscribe(console.log)
  }
}

const test = new Test();

test.functionB();
setTimeout(() => test.functionA(), 500);

// getObservableA
// A
// getObservableB
// B