Tag: 被动编程的

将节点stream转换为Rx.js Observables: 我努力将节点stream转换为Rxjs Observables。当我尝试一个URL时，stream式传输本身就很好用。但是，当我尝试通过一系列URLS映射相同的函数时，我得到错误。我正在使用Rx.Node将stream转换为Observable。这是我目前正在尝试 // data_array is an array of 10 urls that I'm scraping data from. let parentStream = Rx.Observable.from(data_array); parentStream.map(createStream).subscribe(x => console.log(x), (e)=> console.log('Error', e), console.log('Complete')); function createStream(url){ return RxNode.fromStream(x(url, '#centercol ul li', [{name: 'a', link: 'a@href'}]).write().pipe(JSONStream.parse('*'))) } 但是这是输出 X 10（data_array中的URL数量） RefCountObservable { source: ConnectableObservable { source: AnonymousObservable { source: undefined, __subscribe: […]

为什么这个简单的rxjs例子不像预期的那样运行: 我想更深入地理解反应式编程的基础知识，所以我在这里通过示例： http : //reactivex.io/rxjs/manual/tutorial.html#creating-observables 我只是在文本编辑器中input它们并通过节点在terminal中运行它们。第一个例子没有产生任何输出（第二个工作，所以这不是加载库的问题）。我期待它输出“富”，但我什么都没有得到。为什么？ var myObservable = Rx.Subject.create(); myObservable.subscribe(value => console.log(value)); myObservable.next('foo');

Rx.js，订阅被调用undefined: 我正在使用Rx.js将来自AJAX调用的结果stream式传输到多个单元。但是，在观察者订阅MapObserver的时候，我遇到了更多的问题。当第一个用户将总是得到正确的数据，但其余的将不明确。 this.observable = new Rx.Subject(); observeMap = this.observable .map(createMarker.bind(this)); var s1 = observeMap.subscribe(console.log.bind(console, 1)); var s2 = observeMap.subscribe(console.log.bind(console, 2)); 请指教，谢谢！

RxJs避免外部状态，但仍然访问以前的值: 我正在使用RxJs来听一个amqp queu（不是真正相关的）。我有一个函数createConnection返回一个Observable发射新的连接对象。一旦我有连接，我希望每1000毫秒通过它发送消息，10个消息后，我想closures连接。我试图避免外部状态，但如果我不把连接存储在一个外部variables，我怎么能closures它？看到我从连接开始，然后flatMap和推消息，所以几个链后，我不再有连接对象。这不是我的stream量，但想象这样的事情： createConnection() .flatMap(connection => connection.createChannel()) .flatMap(channel => channel.send(message)) .do(console.log) .subscribe(connection => connection.close()) <— obviously connection isn't here 现在我明白这样做很愚蠢，但是现在如何访问连接呢？我当然可以从var connection = createConnection() 后来以某种方式join。但是，我该怎么做呢？我甚至不知道如何正确地提出这个问题。底线，我有一个可观察的，发出连接，打开连接后，我想要一个观察，每1000毫秒发出一个消息（带一个take(10) ），然后closures连接