如何在 rxjs 中模拟“带有承诺的回调”函数参数？答案

【问题标题】：How to mimic a "callback with promise" function parameter in rxjs?如何在 rxjs 中模拟“带有承诺的回调”函数参数？
【发布时间】：2017-03-30 22:33:31
【问题描述】：

我是一个绝对的 rxjs 初学者。为了让我开始学习用 observables 进行思考，我需要通过代码示例来翻译概念。我觉得如果我能看到这个的代码，我就可以开始用其他概念自己做这个了。

我确实不想将一个承诺转换为一个可观察的，我想使用 Observable 进行一个新的实现，它的行为像一个承诺。我将如何使用 Observables 重写以下内容？

 constructor(){
    let makeMessage2 = function(){
        return new Promise(resolve, reject){
           setTimeout(()=>{
                  var r = Math.random();
                  resolve("message two plus random value: " + r );
           }, 1000);
        }
    }
    this.logMessageAndResultOfCallback("message one!", makeMessage2);
}
private sideEffect1:string = "";
private sideEffect2:string = "";

logMessageAndResultOfCallback( message1:string, callback:Function ){
    console.log(message1);
    this.sideEffect1 = message1;

    callback().then((message2)=>{
          console.log(message2);
          this.sideEffect2 = message2;
    }
}

我想我不明白的部分是如何定义“回调”函数，如何调用它。我知道我会等待完整或发出处理程序，例如makeMessage2().subscribe(message2 => console.log(message2));，但我不知道如何定义makeMessage2。

这可能是一个完全没有头绪的问题，但我已经阅读了大约 10 篇关于 rxjs 的不同介绍，但还没有完全得到点击。我只需要将这个场景映射到可观察的模式，我想我可以理解它。

基本上，我想定义一个可观察函数myObs()，它不会“立即执行”，而是在执行someMethod(message:string,obs:Observable) 时“执行”。当myObs 执行时，它应该在其中异步执行一些操作（例如获取HTTP 请求的结果），然后设置下一个值，然后触发一个完整（），以便我在someMethod 中定义的观察者可以处理完整并执行结果。

编辑：我不关心 rxjs 中的计时器或本机等效项，这只是为了模拟任何异步操作，例如从服务器获取数据。

【问题讨论】：

Observable.create 采用订阅者功能——对订阅者执行。这对你不起作用吗？你的问题不是很清楚。在 rxjs 文档站点上，有一个很好的向导，介绍了如何从不同的事物开始观察
“承诺回调模式”也很混乱..通常这些是异步代码的两种不同方法
所以，我最终做了自己的尝试。这看起来正确吗？
没错，这可能是两种异步行为。在我的用例中，它在“回调”完成计算后关闭 UI 项，但在用户操作之前不会调用回调。
是的，这就是我的意思。仅供参考，尽管您根本不需要 setTimeout，但对于这个示例，内置 rx 运算符 delayTime

标签： javascript rxjs reactive-programming observable

【解决方案1】：

您编写并希望“翻译”为可观察对象的代码可能无法正常工作。 callback是promise，不是函数，所以不能写callback()。

你也试过introduction吗？它适用于许多人。

要回答你的问题，你可以写

Rx.Observable.of(""message one!", "message two!")
  .map(console.log.bind(console)) // would be better to use `do` operator actually, for semantic reasons, but that works the same here
  .subscribe(noop, noop, noop)

或

Rx.Observable.of(""message one!", "message two!")
  .subscribe(console.log.bind(console), noop, noop)

其中noop 是一个不执行任何操作的函数，即function noop(){}

简而言之，您的流发出数据，该数据流经一系列运算符，数据流由.subscribe 启动。在您的情况下，您在订阅方面没有什么有趣的事情可做，因为您所做的只是记录。

Rxjs 流实际上是基于回调的。您想查看this answer 以了解它。

【讨论】：

哎呀，这意味着要创建一个返回承诺的函数。编辑的代码。我不认为这是我要找的。预期的用例是一个具有一些初始副作用（记录 message1）的函数，并且需要使用异步解析来调用“回调”。该回调的解析需要调用与回调函数内部无关的行为/副作用。我不只是想打印两条消息。
不，我需要副作用。你的例子不能这样做。我只是让我的例子有具体的副作用来说明。
我添加了我的尝试。这似乎符合我建议的模式？

【解决方案2】：

我在 this 指南的帮助下解决了这个问题。

import {Observable} from 'rxjs';
var makeMessage2 = Observable.create(observer => {
  // Yield a single value and complete
  setTimeout(function(){
     let r = Math.random();
     observer.next("message two plus random value: " + r );
     observer.complete();
  }, 1000);
  return () => console.log('disposed')
});
logMessageAndResultOfCallback( "some message one", makeMessage2);


logMessageAndResultOfCallback( message1:string, callback:Observeable ){
    console.log(message1);
    this.sideEffect1 = message1;

    var subscription = callback.subscribe(
       (value)=>{this.sideEffect2 = value;},
       (e) =>{ console.log('onError: %s', e)},
       () => {console.log(this.sideEffect2);});
      subscription.dispose();
}

【讨论】：