Articles of rxjs

RxJS – 使用zip并行化使HTTP请求不工作的延迟Observable – 您提供了一个无效的对象,其中stream是预期的

我正在运行一个Node.js API端点,它必须在它的一个端点中对几个HTTP调用进行并列。 以星球大战API为例。 我通过id获取用户信息。 它包含一系列的电影,我想并行检索所有这些电影信息。 得到人的信息后,我创build了一个观察数组,每个人得到一个电影。 我build立了一个zip操作员来组成所有得到的电影的结果。 关键是,在订阅后,我看到n次,每个电影一个, console.log(filmData)信息,这是正确的。 但它似乎与console.log("*********************************************************************"); 不叫。 下一个,错误,完整的callback都不是。 为什么会这样? client.get("https://swapi.co/api/people/"+uid, args, (data, response) => { var filmsIt = data.films; for(var i in filmsIt){ var observable = Rx.Observable.defer(function () { client.get(filmsIt[i], args, (filmData, filmResponse) => { console.log(filmData); return filmData; }); }); observables.push(observable); } var observableFinal = Rx.Observable.zip(…observables, function() { console.log("*********************************************************************"); }).subscribe( function […]

节点本地mongodb驱动程序连接池问题

我使用Node.js + RxJS + MongoDB作为socket.io服务器。 经过一定数量的请求后,我的连接池到DB变得非常大。 所以文件描述符永远不会被释放,服务器closures。 对于db查询我使用下面的代码: /* @flow */ var { Observable } = require('rx'); var client = require('mongodb'); var { assign } = require('lodash'); var __DEV__ = process.env.NODE_ENV !== 'production'; var URL = 'my database url'; class QueryBuilder { _db$: Observable; _selectors: Object; constructor(db$: Observable, selectors?: Object) { this._db$ = db$; if […]

什么是“反应”的方式来逐行读取文件

我正在学习使用RxJS的反应式编程,遇到需要逐行读取文件的情况。 其实我解决它使用一个解决scheme喜欢: https://gist.github.com/yvele/447555b1c5060952a279 它的工作原理,但我需要使用一些正常的JS代码将缓冲区的stream转换为线的stream。 (使用上面例子中的“readline”模块) 我想知道是否还有其他的方法可以将Observable of Buffer转换为Observable的行,使用RxJS操作符,喜欢下面的例子。 var Rx = require('rx'); var fs = require('fs'); var lines = Rx.Observable .fromEvent(rl, 'data') // emits buffers overtime // some transforms … .subscribe( (line) => console.log(line), // emit string line by line err => console.log("Error: %s", err), () => console.log("Completed") );

在Node.js中将三个不同的函数映射到Observable

我是Rxjs的新手。 如果可能,我想遵循最佳做法。 我正在执行三个不同的函数在可观察的返回相同的数据。 在“数据stream”概念之后,我一直认为我需要将这个可观察对象分成三个stream,并继续进行。 这是我的代码,所以我可以停止抽象地说话: // NotEmptyResponse splits the stream in 2 to account based on whether I get an empty observable back. let base_subscription = RxNode.fromStream(siteStream).partition(NotEmptyResponse); // Success Stream to perform further actions upon. let successStream = base_subscription[0]; // The Empty stream for error reporting let failureStream = base_subscription[1]; //Code works up until this […]

将节点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去嵌套callback

我想做一些事情: Rx.Observable.of(userToken) .flatMap(verifyToken) .flatMap(getUserInformation) .flatMap(createUser) .flatMap(signNewToken) .subcribe({ next: result => useResult(result), error: error => handleError(error) }) 我试图避免的是一个callback混乱。 在我的代码中,像verifyToken这样的verifyToken是Observables,我想链接它们。 这种模式是否正确? 因为现在只要其中一个内部做observer.error(new Error('problem')) ,链崩溃,我的error handling程序不会被调用。 我怎样才能改善这个?

RxJS – 等待5个事件发生,然后再继续

我有一个简单的RxJS5观察像这样: function foo(){ return Rx.Observable.create(obs => { obs.next(); }); } 可观察的是这样创build的: const x = foo(); function bar(){ return someObs() .flatMap(() => x.wait(5)) // wait for x to fire 5 times .map(v => ({z:v})); } 我想要做的就是等待X个可观察对象中的5个事件发生,然后再继续。 我不认为这个运营商是我想要的,所以我只是叫运营商“ wait ” 我怎样才能用RxJS5做到这一点?

RxJS具有asynchronous订户function的Observable

我试图做一些感觉应该是直截了当的事情,但certificate令人惊讶的困难。 我有一个订阅RabbitMQ队列的函数。 具体来说,这里是Channel.consume函数: http ://www.squaremobius.net/amqp.node/channel_api.html#channel_consume 它返回一个承诺,这个承诺用一个订阅ID来解决 – 稍后需要取消订阅 – 并且还有一个callback参数来在邮件从队列中取出时调用。 当我想取消订阅队列时,我需要在这里使用Channel.cancel函数取消使用者: http ://www.squaremobius.net/amqp.node/channel_api.html#channel_cancel。 这需要先前返回的订阅ID。 我想把所有这些东西包装在观察者订阅时订阅队列的Observable中,并且当observable退订时取消订阅。 然而,由于调用的“双重asynchronous”性质(我的意思是说,它们同时具有callback和返回承诺),这样做有些困难。 理想情况下,我想写的代码是: return new Rx.Observable(async (subscriber) => { var consumeResult = await channel.consume(queueName, (message) => subscriber.next(message)); return async () => { await channel.cancel(consumeResult.consumerTag); }; }); 然而,这是不可能的,因为这个构造函数不支持asynchronous订阅函数或者拆卸逻辑。 我一直无法找出这一个。 我在这里错过了什么? 为什么这么难? 干杯,Alex

如何/何时从Observable.create返回的函数执行(rxjs)

我从' https://chrisnoring.gitbooks.io/rxjs-5-ultimate/content/observable-anatomy.html '有下面的代码: const Observable = require('rxjs/Observable').Observable; require('rxjs/add/observable/of'); require('rxjs/add/operator/map'); let stream = Observable.create((observer) => { let i = 0; let id = setInterval(() => { observer.next(i++); }, 500); return function () { // Line 11 clearInterval(id); }; }) let subscription = stream.subscribe((value) => { console.log('Value: ', value); }) setTimeout(() => { subscription.unsubscribe(); }, 1500); 这个程序的输出如下。 […]

NodeJS – 响应stream

我使用Sails.js构build了一个使用NodeJS的简单API端点。 当有人访问我的API端点时,服务器开始等待数据,每当出现新数据时,他都使用套接字广播它。 每个客户应该根据他的用户input接收他自己的数据stream。 var Cap = require('cap').Cap; collect: function (req, res) { var iface = req.param("ip"); var c = new Cap(), device = Cap.findDevice(ip); c.on('data', function(myData) { sails.sockets.blast('message', {"host": myData}); }); }); 响应没有完成(我从来没有发送res.json() – 实际上发生的事情是浏览器不断加载 – 但上述function)。 2问题: 我正尝试从我的客户端(使用RxJS)订阅和取消订阅此API端点。 当我订阅时,我开始通过套接字接收数据,但我不能退订API端点(浏览器期望完成请求)。 每个客户端都应该根据请求IP参数订阅他自己的套接字空间(请参阅更新后的代码)。 目前它把这个信息传达给每个人。 如何用Sails.js创build一个类似Stream / Service的API端点,并根据他的input向每个用户发送新的数据? 我的目标是能够从每个客户端订阅/取消订阅此API端点。