避免 RXJS 中的嵌套订阅

Question

我有以下代码：

someService.subscribeToChanges().subscribe(value => {
    const newValue = someArray.find(val => val.id === value.id)
    if (newValue) {
       if (value.status === 'someStatus') { 
          someArray.findIndex((a) => a.id === value.id)
          someArray.splice(index, 1);
       } else { 
          newValue.status = value.status;
       }
    } else { 
        if (value.someProp === 'abc') {
          someService.doSomeAsyncStuff().subscribe(data => {
            someService.handleData(data);
          });
        }
    }
 });

我想保持相同的逻辑，同时避免第二次订阅。

这是我尝试过的代码

sub$ = someService.subscribeToChanges().pipe(switchMap(value => {
  const newValue = someArray.find(val => val.id === value.id)
  if (newValue) {  
       if (value.status === 'someStatus') { 
          someArray.findIndex((a) => a.id === value.id)
          someArray.splice(index, 1);
       } else { 
          newValue.status = value.status;
       }
  } else { 
    return iif(() => value.someProp === 'ABC', someService.doSomeAsyncStuff);
  }
}));

sub$.subscribe(data => someService.handleData(data))

如果iif 条件为真，则此方法有效，但如果为假，它只会停止整个流，而我希望它继续流。

Answer 1

你可以把它们分成三个条件

sub$ = someService.subscribeToChanges().pipe(switchMap(value => {
 const newValue = someArray.find(val => val.id === value.id)
 if (newValue) {
   if (value.status === 'someStatus') { 
      someArray.findIndex((a) => a.id === value.id)
      someArray.splice(index, 1);
   } else { 
      newValue.status = value.status;
   }
 return of(value); 
 }

    if(value.someProp === 'ABC')
      return someService.doSomeStuff()

    return of(value)
  })
)

Answer 2

嵌套订阅是 rxjs 中的 antipattern。所以你尽量避免它是好的。我建议您仍然使用两个订阅，但不要嵌套。因为你有不同的副作用（someService.handleData 和 doSomeStuff）

出于重用原因，使用 someArray id 比较器创建您的 subscribeTochanges id

const subscribeToChangesIdEqualsSomeArrayId$ = someService.subscribeToChanges().pipe(
  map(value => someArray.find(val => val.id === value.id)
);

创建你的 doSomeStuff Observable

const doSomeStuff$ = subscribeToChangesIdEqualsSomeArrayId$.pipe(
  filter(newValue => newValue)
);

创建您的 handleData Observable

const handleData$ = combineLatest([
  subscribeToChangesIdEqualsSomeArrayId$,
  someService.subscribeToChanges()
]).pipe(
  filter(([newValue, value]) => !newValue && value.someProp === 'abc'),
  switchMap(() => someService.doSomeStuff())
);

最终使用

doSomeStuff$.subscribe(() => doSomeStuff());
handleData$.subscribe(data => someService.handleData(data));

仅供参考：我使用了您在问题中使用的所有外部和内部（从管道角度）变量名称。为了阅读，我可能会调整它们:)

Answer 3

将所有逻辑保持在一条链中总是更好，因此您可以像这样进行操作。由于您不需要newValue 更进一步的链条，它非常简单：

someService.subscribeToChanges()
  .pipe(
    switchMap(value => {
      const newValue = someArray.find(val => val.id === value.id);
      if (!newValue && value.someProp === 'abc') {
        return someService.doSomeStuff();
      }
      return of(undefined);
    }),
  )
  .subscribe(data => {
    if (data === undefined) { // When data is set means we ran the second Observable
      doSomeStuff();
    } else {
      someService.handleData(data);
    }
  });

如果someService.doSomeStuff() 是undefined 的有效响应，那么我会用{ response: data } 之类的对象包装它的响应，这样您就永远知道它来自哪里。

在真正有必要之前尽量避免进行多次订阅，因为这会带来更多你必须注意的事情（比如订阅顺序、共享 Observables 等等），而且以自上而下的顺序阅读 RxJS 链总是更容易，例如这是同步代码。