递归 Observable 调用不返回数据

Question

我需要 RxJS 专业人士的帮助 :)

我尝试通过 http 请求从 REST API 递归加载数据。 递归调用工作正常，但是当我订阅最终的 Observable（由 GetTemperatures 返回）时，订阅中没有返回数据。

似乎没有数据在调用链中传回。

这里出了什么问题？

GetTemperatures().subscribe((data: MeasureData) => {

    // add data to a chart, etc...

})

GetTemperatures(): Observable<MeasureData> {

    const l_startDate = new Date(2019, 0, 1);

    var l_httpParams = new HttpParams()
    .set('device_id', this._deviceId)
    .set('module_id', this._moduleId)
    .set('scale', '1hour')
    .set('type', 'Temperature')
    .set('date_begin', Math.floor(l_startDate.getTime() / 1000).toString())
    .set('real_time', 'true')
    .set('optimize', 'true');

    return this._http.post<MeasureDataInternal>(this._getMeasureUrl, l_httpParams)
    .pipe(
        map((data: MeasureDataInternal): MeasureData => this.transformMeasureData(data)),
        flatMap((data: MeasureData) => {
            return this.recursiveLoadData(data);
        })
    );
}

recursiveLoadData(data: MeasureData): Observable<MeasureData> {

    // search until now minus 1,5 hours
    const endDate = new Date(Date.now() - (1.5 * 60 * 60 * 1000));

    console.error('RECURSIVE begin: ' + data.value[0].date + ' end: ' + data.value[data.value.length - 1].date);

    // check if complete
    if (data.value[data.value.length - 1].date.getTime() >= endDate.getTime()) {
      console.error('recursive ENDs here');
      return EMPTY;
    }

    var l_httpParams = new HttpParams()
    .set('device_id', this._deviceId)
    .set('module_id', this._moduleId)
    .set('scale', '1hour')
    .set('type', 'Temperature')
    .set('date_begin', Math.floor(data.value[data.value.length - 1].date.getTime() / 1000).toString())
    .set('real_time', 'true')
    .set('optimize', 'true');

    return this._http.post<MeasureDataInternal>(this._getMeasureUrl, l_httpParams)
      .pipe(
        map((data2: MeasureDataInternal): MeasureData => this.transformMeasureData(data2)),
        flatMap((data2: MeasureData) => {
          return this.recursiveLoadData(data2);
        })
      )

}

Answer 1

我不知道您真正想要完成什么，但是递归中的每个新步骤除了将您带到下一步之外没有任何作用。因此，您需要包含您希望每个步骤执行的操作。

这不是特定于流的，一般递归也是如此。

一般递归

这实际上与常规递归函数的工作方式没有任何不同。假设您正在递归地将数组中的数字相加，您需要将数组的尾部添加到第一个值。如果你只是在一个较小的数组上递归而不将你弹出的数字相加，你会得到基本情况的值。

这将返回数组的最后一个值（数组的最后一个值是基本情况）：

recursiveAdd(array){
  if(array.length === 1) return array[0];
  return recursiveAdd(array.shift());
}

这会添加数组：

recursiveAdd(array){
  if(array.length === 1) return array[0];
  return array[0] + recursiveAdd(array.shift());
}

在这个简单的例子中，+ 操作数在递归的每一步都完成了工作。没有它，数组就不会被求和。当然，我可以做任何事情。从 1000 中减去数组，平均数组中的数字，从值构建一个对象。 任何东西。

在进行递归调用之前，您必须做一些事情。除非您所追求的是基本案例的价值（在您的情况下，是一个空流）

使用流递归

当您将一个值mergeMap 到一个流中时，您不会同时传递该值。

from([69,70,71]).pipe(
  mergeMap(val => from([
    String.fromCharCode(val),
    String.fromCharCode(val),
    String.fromCharCode(val)
  ]))
).subscribe(console.log);

输出

e e e f f f g g g

请注意输出不包含任何数字？当您mergeMap 时，您将值映射到流中。如果您希望要映射的值成为流的一部分，则必须以某种方式包含它们。这与一般递归相同。

因此，这里有两个示例，它们都将您的数据包含在返回的流中。它们非常基础，但希望您能从中获得一些理解并加以应用。

这会将返回的 Steam 转换为包含您的数据作为其第一个值（当然是递归的）

return this._http.post<MeasureDataInternal>(this._getMeasureUrl, l_httpParams)
.pipe(
  map((data: MeasureDataInternal): MeasureData => 
    this.transformMeasureData(data)
  ),
  mergeMap((data: MeasureData) => 
    this.recursiveLoadData(data).pipe(
      startWith(data)
    )
  )
);

这会创建一个数据流、一个递归调用流，并将这两个流合并在一起。

return this._http.post<MeasureDataInternal>(this._getMeasureUrl, l_httpParams)
.pipe(
  map((data: MeasureDataInternal): MeasureData => 
    this.transformMeasureData(data)
  ),
  mergeMap((data: MeasureData) => 
    merge (
      of(data),
      this.recursiveLoadData(data)
    )
  )
);