rxjs Observable - 订阅导致循环和条件中断/无法读取未定义的属性“订阅”

Question

我是 rxjs 的新手，似乎无法为我正在尝试做的事情找到正确的运算符。

在我的示例中，我有一个数组，我需要用另一个 observable 的结果来填充，一旦我在该数组中通过订阅调用获得了足够的结果，我想中断并返回该数组。

// for reference, there is a class variable called keys

    result = getResults(data, index).subscribe((result: any[]) => { 
          doSomethingWith(result);
    });

    getResults(data: any[], index: number) : Observable<any[]> {
       obsFunctionThatGetsMeData(keys[index]).subscribe(result => 

         data = data.concat(result);

          if(data.length >= NEEDED_NUM_DATA) {
              return Observable.of(data);

          } else {

            /* need to call obsFunctionThatGetsMeData(keys[index++]) and do the same logic as above. */

          }
       );
    }

我知道将订阅放在订阅中是不好的做法，这只是我正在寻找的想法。我知道 takeWhile 在某个条件下有效，但是如果该条件失败，我不知道如何进行额外的调用。有谁知道哪种运算符最适合这种行为？

谢谢！

• obsFunctionThatGetsMeData 返回 Observable

解决了我自己的问题 使用递归和 switchmap

getResults(data: any[], index: number): Observable<any> {
   return obsFunctionThatGetsMeData(keys[index]).pipe(
        switchMap(result => {
            if (result) {
                data= data.concat(result);

                if (data.length < NEEDED_NUM_DATA && index < keys.length) {
                    return getResults(data, ++index);
                } else {
                    return Observable.of(data);
                }
            }
        }));
}

Answer 1

我不是 RxJS 方面的专家，但我认为这样的事情应该是可能的：

Rx.Observable.from(obsFunctionThatGetsMeData)
  .take(NEEDED_NUM_DATA)
  .subscribe(doSomethingWith);

public take(count: number): Observable 只发射第一个计数 源 Observable 发出的值。

这只是一个想法，并不假装是一个可行的解决方案。

Answer 2

看起来 obsFunctionThatGetsMeData 发出一次（如 HTTP 请求），并且您希望将这些请求的 NEEDED_NUM_DATA 数量的结果收集到数组中。 它还应该附加到你已经在 data 参数中的任何内容？

试试这个：

getResults( data: any[], index: number ): Observable<any[]>
{
    return Observable.range( index, NEEDED_NUM_DATA )
        .concatMap( i => obsFunctionThatGetsMeData( keys[i] ) )
        .reduce(( acc: any[], value ) => acc.concat( value ), data );
}

Answer 3

Promises 是你的朋友。 （因为您使用的是单个值）。

尽管您认为您正在处理结果流（可观察对象），但实际上您一次只处理一个值。每个值都在循环中独立操作，将结果推送到数组中。所以NEEDED_NUM_DATA自变量合并成一个集合。

这段代码更容易合理化并实现您的目标：

var NEEDED_NUM_DATA = 4;
var collection = [];
async function getResults() {
  let url = 'https://jsonplaceholder.typicode.com/posts/';
  let i = 1;
  while(collection.length <= NEEDED_NUM_DATA) {
    let result = await getsMeData(url + i++);
    collection.push(result);
  }
}
async function getsMeData(url) {
  const response = await fetch(url);
  const json = await response.json();
  return json;
}
getResults()
.then(()=>console.log(collection));