如何从内部具有 Observable 订阅的函数返回值？

Question

我不知道如何从 Observable 中提取值以由存在 Observable 的函数返回。我只需要从中返回一个值即可。

有效的当前版本

function getValueFromObservable() {
    this.store.subscribe(
        (data:any) => {
            console.log(data)
        }
    )
}
getValueFromObservable()

我需要这个工作，函数返回值，然后：

function getValueFromObservable() {
    this.store.subscribe(
        (data:any) => {
            return data
        }
    )
}
console.log(getValueFromObservable())

我在这里做错了什么？

Answer 1

这不是使用Observable的完全正确的想法

在组件中，您必须声明将包含一个对象的类成员（您将在组件中使用的东西）

export class MyComponent {
  name: string = "";
}

然后Service 将返回一个Observable：

getValueFromObservable():Observable<string> {
    return this.store.map(res => res.json());
}

Component 应该做好准备，以便能够从中检索值：

OnInit(){
  this.yourServiceName.getValueFromObservable()
    .subscribe(res => this.name = res.name)
}

您必须将 Observable 中的值分配给变量：

您的模板将使用变量name：

<div> {{ name }} </div>

另一种使用Observable的方式是通过async管道http://briantroncone.com/?p=623

注意：如果这不是您所要求的，请更新您的问题并提供更多详细信息

Answer 2

编辑：更新代码以反映在最新版本的 RXJS 中对管道工作方式所做的更改。所有运算符（以我的示例为例）现在都包装在 pipe() 运算符中。

我意识到这个问题是很久以前的事了，你现在肯定有一个合适的解决方案，但对于任何寻找这个问题的人，我建议用 Promise 来解决它以保持异步模式。

更详细的版本是创建一个新的 Promise：

function getValueFromObservable() {
    return new Promise(resolve=>{
        this.store.pipe(
           take(1) //useful if you need the data once and don't want to manually cancel the subscription again
         )
         .subscribe(
            (data:any) => {
                console.log(data);
                resolve(data);
         })
    })
}

在接收端，您将“等待”承诺通过以下方式解决：

getValueFromObservable()
   .then((data:any)=>{
   //... continue with anything depending on "data" after the Promise has resolved
})

一个更精简的解决方案是使用 RxJS 的 .toPromise() 代替：

function getValueFromObservable() {
    return this.store.pipe(take(1))
       .toPromise()   
}

接收方当然和上面一样。

Answer 3

可以从任何位置检索可观察值。源序列首先被 推送 到一个特殊的 observer 上，该观察者可以在其他地方发射。这是通过 Reactive Extensions (RxJS) 中的 Subject 类 实现的。

var subject = new Rx.AsyncSubject();  // store-last-value method

将值存储到观察者。

subject.next(value); // store value
subject.complete(); // publish only when sequence is completed

要从其他地方检索值，请像这样订阅观察者：

subject.subscribe({
  next: (response) => {
      //do stuff. The property name "response" references the value
  }
});

Subjects 既是 Observables 又是 Observers。还有其他的Subject types，比如 BehaviourSubject 和 ReplaySubject 用于其他使用场景。

别忘了导入 RxJS。

var Rx = require('rxjs');

Answer 4

如果您想预订阅将返回的同一个 Observable，只需使用

.do():

function getValueFromObservable() {
    return this.store.do(
        (data:any) => {
            console.log("Line 1: " +data);
        }
    );
}

getValueFromObservable().subscribe(
        (data:any) => {
            console.log("Line 2: " +data)
        }
    );

Answer 5

例如这是我的 html 模板：

<select class="custom-select d-block w-100" id="genre" name="genre"
                  [(ngModel)]="film.genre"
                  #genreInput="ngModel"
                  required>
            <option value="">Choose...</option>
            <option *ngFor="let genre of genres;" [value]="genre.value">{{genre.name}}</option>
          </select>

这是与我的组件中的模板绑定的字段：

  // Genres of films like action or drama that will populate dropdown list.
  genres: Genre[];

我从服务器动态获取电影类型。为了与服务器通信，我创建了FilmService

这是与服务器通信的方法：

 fetchGenres(): Observable<Genre[]> {
    return this.client.get(WebUtils.RESOURCE_HOST_API + 'film' + '/genre') as Observable<Genre[]>;
  }

为什么这个方法返回 Observable<Genre[]> 而不是 Genre[]？

JavaScript 是async，它不会等待一个方法在一个昂贵的过程之后返回值。昂贵的意思是一个需要时间来返回价值的过程。就像从服务器获取数据一样。所以你必须返回 Observable 的引用并订阅它。

例如在我的组件中：

ngOnInit() {
    this.filmService.fetchGenres().subscribe(
      val => this.genres = val
    );
  }

Answer 6

问题是数据是在 observable 中捕获的，我可以通过控制台记录它。我想通过调用它所在的函数来返回该值和 console.log 或来自不同文件的任何内容。

看起来您正在可观察对象内寻找“当前值”getter，当它发射时和发射后。

Subject 和 Observable 没有这样的东西。当一个值被发出时，它被传递给它的订阅者，Observable 就完成了。

您可以使用 BehaviorSubject 存储最后发出的值并立即将其发送给新订阅者。

它还有一个getValue()方法来获取当前值；

进一步阅读：

RxJS BehaviorSubject

How to get current value of RxJS Subject or Observable?

Answer 7

虽然前面的答案可能以某种方式起作用，但我认为如果您想继续使用 observables，使用 BehaviorSubject 是正确的方法。

例子：

    this.store.subscribe(
        (data:any) => {
            myService.myBehaviorSubject.next(data)
        }
    )

在服务中：

let myBehaviorSubject = new BehaviorSubjet(value);

在component.ts中：

this.myService.myBehaviorSubject.subscribe(data => this.myData = data)

我希望这会有所帮助！

Answer 8

在单线程、异步、面向承诺、反应式趋势的 javascript 世界中，async/await 是命令式程序员最好的朋友：

(async()=>{

    const store = of("someValue");
    function getValueFromObservable () {
        return store.toPromise();
    }
    console.log(await getValueFromObservable())

})();

如果store 是多个值的序列：

  const aiFrom = require('ix/asynciterable').from;
  (async function() {

     const store = from(["someValue","someOtherValue"]);
     function getValuesFromObservable () {
        return aiFrom(store);
     }
     for await (let num of getValuesFromObservable()) {
       console.log(num);
     }
  })();

Answer 9

function getValueFromObservable() {
    this.store.subscribe(
        (data:any) => {
            return data
        }
    )
}
console.log(getValueFromObservable())

在上述情况下，console.log 在 promise 解决之前运行，因此不显示任何值，请将其更改为 follow

function getValueFromObservable() {
    return this.store
}

getValueFromObservable()
 .subscribe((data: any) => {
    // do something here with data
    console.log(data);
});

另一种解决方案是当您需要 getValueFromObservable 中的数据以使用 of 运算符返回可观察对象并订阅该函数时。

 function getValueFromObservable() {
        return this.store.subscribe((data: any) => {
            // do something with data here
            console.log(data);
            //return again observable.
            return of(data);
       })
    }

    getValueFromObservable()
     .subscribe((data: any) => {
        // do something here with data
        console.log(data);
    });

Answer 10

体面的方法是从函数返回 observable 并在需要的地方订阅它，因为 observable 是惰性的，它们只有在订阅时才会开始发出值。

这里我有一个更有趣的事件驱动解决方案，我最初用来玩弄它。以下示例通过使用 nodejs 的“events”模块来实现。 您可以将它与存在类似模块的其他框架一起使用（注意：语法和样式可能会根据使用的模块而改变）。

var from =require("rxjs").from;
var map = require("rxjs/operators").map;
var EventEmitter = require("events");

function process(event) {
    from([1,2,3]).pipe(
        map(val => `The number is:: ${val}`)
    ).subscribe((data) => {
       event.emit("Event1", data); //emit value received in subscribe to the "Event1" listener
    });
}

function main() {
   class Emitter extends EventEmitter{};
    var event = new Emitter(); //creating an event
    event.on("Event1", (data)=>{ //listening to the event of name "Event1" and callback to log returned result
        console.log(data); //here log, print, play with the data you receive
    });
    process(event); //pass the event to the function which returns observable.
}

main(); //invoke main function

这只是一个示例，展示了我们可以通过发射和监听的方法从不同地方传递数据的想法。这也称为事件驱动代码。