Behaviorsubject和of的行为完全不同

问题描述

我正在尝试建立一个可观察变量的处理链，其中第一步必须是一个主题（我需要调用.next（））。使用of（）有效，但返回一个Observable，使用BehaviorSubject应该具有相似的效果，但是它不起作用：with的observable创建的完美起作用，并且订阅将数据通过管道并返回修改的数据，而使用行为主题将数据保留在主题中，订阅将永远不会获取数据。

示例：

getProcessed(processed: string = null,identifier = 'default'): Observable<any> {
  const bs = new BehaviorSubject(this.start.data);
  this.localFilterSub.set(identifier,bs);
  this.localFilterObs.set(identifier,bs.asObservable());
  this.localFilterSet.set(identifier,{});
  this.process(processed,identifier);
  return this.localFilterObs.get(identifier);
}

process(name: string,identifier = 'default') {
  this.localFilterObs.set(identifier,this.doProcess(name,identifier));
}


private doProcess(name: string,identifier = 'default'): Observable<any>|Subject<any> {
  if (name) {
    const inst = new Op();

    const obss = [];
    obss.push(this.localFilterObs.get(identifier));
    obss.push(inst.getExternal());

    return forkJoin(obss).pipe(
      tap((data) => {
        console.log(name,data);
      }),map((data) => {
        return inst?.run(data);
        // this.done.push(name);
      }),tap((data) => {
        console.log(name,);
  } 
}

我真的不明白我在做什么错。

解决方法

forkJoin在所有可观察对象完成之前不会发出。会立即完成，但是行为主题要等到您致电完成后才能完成。使用combinateLatest，它会在所有可观察对象都发出后发出。

const { BehaviorSubject,of,forkJoin } = rxjs;

const bs$ = new BehaviorSubject('bs');

o$ = of('of')

forkJoin([bs$,o$]).subscribe(res => { console.log(res); });

console.log('Nothing yet as bs$ not complete');

setTimeout(() => { bs$.complete(); },2000);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.2/rxjs.umd.min.js"></script>

但是CombineLatest会立即发出

const { BehaviorSubject,combineLatest } = rxjs;

const bs$ = new BehaviorSubject('bs');

o$ = of('of')

combineLatest([bs$,o$]).subscribe(res => { console.log(res); });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.2/rxjs.umd.min.js"></script>