使用 RxJS 进行速率限制

Question

我现在正在发现 RxJS，我的第一次尝试是尝试对 API 请求进行速率限制。

不知怎的，我遗漏了一些东西，输出只是“未定义”。

我做错了什么？

const Rx = require('rx');
const request = require('request');

function f() {
  return Rx.Observable.from(arguments);
}

function expand(condensedId) {
  console.log('requesting', condensedId)
  return f(request(INDEX_URL + '/' + condensedId));
}

const INDEX_URL = 'http://jsonplaceholder.typicode.com/posts';

var source = f([1,2,3,4,5,6,7])
  .windowWithTimeOrCount(5000, 2)//rate limitation, 2 every 5 seconds
  .flatMap(condensed => expand(condensed))
  .map(entry => entry.title);

var subscription = source.subscribe(
  function (x) {
    console.log('title: %s', x);
  },
  function (err) {
    console.log('Error: %s', err);
  },
  function () {
    console.log('Completed');
  });

Answer 1

Rx.Observable.from 需要一个可迭代的，我不认为对 request() 的响应是一个可迭代的。您可以将返回 Promise 或 Observable 的函数传递给 flatMap，它会返回一个流，该流将发出已解析的数据。

因此，让我们使用request-promise 代替request 并在expand 函数中返回一个Promise。另外，让我们使用 cheerio 库来提取 html 标题：

const Rx = require('rx');
const request = require('request-promise');

// HTML parsing library
const cheerio = require('cheerio');

function f() {
  return Rx.Observable.from(arguments);
}

const INDEX_URL = 'http://jsonplaceholder.typicode.com/posts';

// Return an Observable of resolved responses
function expand(condensedId$) {
  return condensedId$.flatMap(id => request(INDEX_URL + '/' + id));
}

var source = f([1,2,3,4,5,6,7])
  .windowWithTimeOrCount(5000, 2)//rate limitation, 2 every 5 seconds
  .flatMap(condensed => expand(condensed))
  .map(body => {
    const $ = cheerio.load(body);
    return $('title').text();
   });