从 Observable<value> 对象中获取值答案

【问题标题】：get a value from a Observable<value> object从 Observable<value> 对象中获取值
【发布时间】：2019-02-05 08:22:57
【问题描述】：

我正在编写一个服务函数，它应该在哪里回复产品的最新价格。我连接到我的服务器 - 使用 websocket 并使用 Rxjs observable 获得 DataStream，现在服务层中的函数应该只回复值而不是 Observable 对象。

我正在尝试的示例代码

public getRealTimePrice(id: string): number {

 let ws = new $WebSocket('/price?Id=' + id + '&connectionId=' + this.randomUUID());
    this.priceWebsocket = ws;

    this.priceWebsocket.getDataStream()
      .map(msg => JSON.parse(msg.data).price)
      .subscribe(responseNumber => {
        return responseNumber // by the time execution reaches here the context is changed.
      });
    // I am not returning anything on the function level - which results in a compilation error as per typescript.
  }

但由于很明显的原因，我无法返回数字 - 因为它在订阅回调中，上下文发生变化并且函数是异步的

如何等待 Observable 的回复，然后只返回 observable 中收到的值。

【问题讨论】：

标签： javascript rxjs observable

【解决方案1】：

做到这一点的唯一真正方法是返回一个承诺。因为调用是异步的，所以你不能只返回值。这样的事情会起作用：

public getRealTimePrice(id: string): Promise<number> {
    return new Promise((resolve, reject) => {
        let ws = new $WebSocket('/price?Id=' + id + '&connectionId=' + this.randomUUID());
        this.priceWebsocket = ws;

        this.priceWebsocket.getDataStream()
        .map(msg => JSON.parse(msg.data).price)
        .subscribe(responseNumber => {
            resolve(responseNumber);
        });
    });
}

那么你可以这样称呼它：

getRealTimePrice().then((data:number) = {
    console.log(data);
});

【讨论】：

我正在 Angular 中制作 FormArray - 因此需要从服务器的实时服务更新输入值之一。因此，我需要一个返回值的函数，而不是我见过的等待函数调用的 Promise 或 Observable - 如果有人可以提供它的用法示例。
我之前没有使用过angular的await功能，但是看起来很简单。我会添加另一个答案，但我没有测试过代码，所以不确定它是否会工作。另外值得注意的是，将有一种方法可以使用返回承诺的函数来完成您需要的操作，但如果没有您的代码，我将无法提供帮助。

【解决方案2】：

理想情况下，您应该在 getRealTimePrice 方法中从对 getDataStream() 的调用中返回 Observable，并在预期数据的位置进行订阅。

【讨论】：

【解决方案3】：

我建议像这样使用 toPromise：

public getRealTimePrice(id: string): Promise<number> {
    let ws = new $WebSocket('/price?Id=' + id + '&connectionId=' + this.randomUUID());
    this.priceWebsocket = ws;

    return this.priceWebsocket.getDataStream()
        .map(msg => JSON.parse(msg.data).price)
        .take(1)
        .toPromise()
}

注意 take(1) 运算符，它将获取第一个值并取消订阅 DataStream。

【讨论】：

【解决方案4】：

我不确定 Angular 的 await 功能，但查看文档看起来会是这样的：

public getRealTimePrice(id: string): Promise<number> {

    let ws = new $WebSocket('/price?Id=' + id + '&connectionId=' + this.randomUUID());
    this.priceWebsocket = ws;

    this.priceWebsocket.getDataStream()
      .map(msg => JSON.parse(msg.data).price)
      .subscribe(responseNumber => {
        return responseNumber
      })
      .take(1)
      .toPromise();
}

然后你可以使用 await 关键字调用它：

async getAsyncValue(id: string) {
    const value = <number>await this.getRealTimePrice(id);
    console.log(`number: ${value}`);
}

请注意，函数以“async”关键字为前缀，如果函数中包含“await”关键字，则需要这样做。

“async”关键字意味着在“getRealTimePrice()”函数的promise被解决（或拒绝）之前不会命中console.log。

【讨论】：