如何从函数创建 Observable？答案

【问题标题】：How to create Observable from function?如何从函数创建 Observable？
【发布时间】：2017-05-16 21:16:12
【问题描述】：

我想（同步地）调用一个函数，然后使用它的返回值作为初始发射（随后在结果 observable 上链接一些其他运算符）。

我想在订阅过程中调用这个函数，所以我不能只使用Observable.of(() => getSomeValue())。我见过bindCallback（以前是fromCallback），但我认为它不能用于这项任务（如果我错了，请纠正我）。我在 v4 文档中看到了 start 静态运算符，但显然它没有在 v5 中实现（并且没有迹象表明它正在开发中）。 RxJava 也有 fromCallable 操作符，可以做到这一点。

我唯一能想到的方法是这样的：

Observable.create((observer: Observer<void>) => {
  let val = getSomeValue();
  observer.next(val);
  observer.complete();
})

我认为就是这样做的。但这对于一个可能应该像Observable.fromFunction(() => getSomeValue()) 这样的简单事情来说似乎太复杂了，如果我想像start 运算符那样异步运行它怎么办？如何在当前版本的 RxJS 中做到这一点？

【问题讨论】：

你想达到什么目的？链接函数调用？
@ChirdeepTomar 是的，我正在尝试将函数用作种子并在其结果上链接可观察运算符。
通过订阅查看平面地图。 stackoverflow.com/questions/35268482/…
@ChirdeepTomar 我认为你没有仔细阅读我的问题......我的目标函数（我试图启动可观察链）没有返回可观察值，它只是返回普通值（比如string)，所以我显然不能在上面使用flatmap（或任何其他 Observable 运算符）。

标签： javascript rx-java rxjs system.reactive rxjs5

【解决方案1】：

我倾向于尽可能避免对Observable.create 的任何显式使用，因为通常它是错误的来源，不仅要管理您的事件发射，还要管理您的拆卸逻辑。

您可以改用Observable.defer。它接受一个返回Observable 或Observable-like 事物的函数（阅读：Promise、Array、Iterators）。因此，如果您有一个返回异步事物的函数，则很简单：

Observable.defer(() => doSomethingAsync());

如果您希望它与同步结果一起使用，请执行以下操作：

Observable.defer(() => Observable.of(doSomethingSync()));

注意：就像create 这将在每个订阅上重新运行该功能。这与 Observable.bindCallback 的结果不同，它存储函数调用结果而不重新执行函数。因此，如果您需要这种行为，则需要使用适当的 multicasting 运算符。

【讨论】：

【解决方案2】：

我在项目中使用的fromFunction$ 的实现：

function fromFunction$<T>(factory: () => T): Observable<T> {
    return Observable.create((observer: Subscriber<T>) => {
        try {
            observer.next(factory());
            observer.complete();
        } catch (error) {
            observer.error(error);
        }
    });
}

像这样使用：

fromFunction$(() => 0).subscribe((value) => console.log(`Value is '${value}'`), null, () => console.log('Completed'));
fromFunction$(() => [1, 2, 3]).subscribe((value) => console.log(`Value is '${value}'`), null, () => console.log('Completed'));
fromFunction$(() => { throw 'Something' }).subscribe(null, (error) => console.error(`Error: ${error}`));

给予：

Value is '0'
Completed

Value is '1,2,3'
Completed

Error: Something

直到存在这样的实现。

【讨论】：

【解决方案3】：

实际上我认为最好的选择是使用Observable.create，因为它是同步和异步初始值最通用的解决方案。

如果您确定要使用同步函数，则可以使用 startWith() 运算符（只有在 getSomeValue() 的返回值对所有观察者都应该相同时才有意义）。

使用 Observable.bindCallback 作为 Observable 源当然是可行的，但我个人建议避免使用它，因为它会使您的代码非常难以理解，而且通常没有必要，因为您可以只使用 Observable.create。

【讨论】：