angular - 你将如何从函数返回一个 Observable

Question

我有这个功能:

  getData(): Observable<ServerResponse> {
    const data = {} as ServerResponse;

    data.client = this.getClientData().pipe(
      map(response =>
        response.map(x => {
          return x.data.client;
        }),
      ),
    );

    data.otherData = this.otherData().pipe(
      map(response =>
        response.map(x => {
          return this.groupByTimePeriod(x.data.other, 'day', 'week');
        }),
      ),
    );
  }

函数需要返回 data 以及在函数内部分配给它的所有属性。你会怎么做？

如果我只是返回 data 它是行不通的，因为 this.getClientData() 和 this.otherData() 还没有正在完成。

Answer 1

那么，您在这里有多个问题。我将从最简单的开始。您如何从函数/对象中获取可观察对象？答案是通过观察 of :

return of(data);

但是您避开了一个更大的问题，即:您如何推迟返回数据直到子可观察对象发出它们的值？您正在寻找forkJoin .通过文档:

forkJoin will wait for all passed Observables to complete and then it will emit an array with last values from corresponding Observables. So if you pass n Observables to the operator, resulting array will have n values, where first value is the last thing emitted by the first Observable, second value is the last thing emitted by the second Observable and so on. That means forkJoin will not emit more than once and it will complete after that.

您还有其他几个问题。例如，您永远不会订阅 this.getClientData() 或 this.otherData()。 Observables 是延迟执行的。您的可观察对象中的代码在订阅它之前不会执行。来自 the docs :

The code inside Observable.create(function subscribe(observer) {...}) represents an "Observable execution", a lazy computation that only happens for each Observer that subscribes.

它看起来好像您正在使用 pipe/map 试图在您的 data 对象上设置一个属性。但是您永远不会设置 data.client 或 data.other，因此它们将始终为空。

所以，将它们放在一起，这就是您的代码可能的样子，模拟服务器延迟以显示 forkJoin 等待两个 observable 的完成:

import { Injectable } from '@angular/core';
import { Observable, of, forkJoin } from 'rxjs';
import { delay } from 'rxjs/operators';

@Injectable({
    providedIn: 'root'
})
export class TestService {
    getData(): Observable<ServerResponse> {
        const allOperations = forkJoin(
            this.getClientData(),
            this.getOtherData()
        );

        const observable = Observable.create(function subscribe(observer) {
            // Wait until all operations have completed
            allOperations.subscribe(([clientData, otherData]) => {
                const data = new ServerResponse;
                // Update your ServerReponse with client and other data
                data.otherdata = otherData.other;
                data.client = clientData.client;
                // Now that data is 100% populated, emit to anything subscribed to getData().
                observer.next(data);
                observer.complete();
            });

        });
        // We return the observable, with the code above to be executed only once it is subscribed to
        return observable;
    }

    getClientData() : Observable<any> {
        return of({ client: 'Client 1' });
    }
    getOtherData(): Observable<any> {
        // Fake server latency
        return of({ other: 'Other data that takes a while to return from server...' })
            .pipe(delay(2000));
    }
}

export class ServerResponse {
    client: string;
    otherdata: string;
}

如果您调用 getData() 并订阅 observable，您将看到 forkJoin 按预期工作，我们必须等待 2子 Observable 完成和我们的 Observable 发出值的秒数:

this.testService.getData().subscribe(data => {
  console.log(data);
});

看来您可能是 RxJS/异步编程的新手。我建议阅读 the excellent introduction当你有机会的时候去 RxJs。一开始可能会很棘手，但通过练习，这将成为第二天性。