javascript - Promise 流

Question

我正在尝试 promise 流，但它似乎比我预期的更难。这是我的尝试:

'use strict'

const Promise = require('bluebird')
const Twitter = require('twitter')

const TwitterStream = module.exports = function TwitterStream (config) {
  // init Twitter Streaming API for OAuth
  this.stream = new Twitter({
    consumer_key: config.get('/twitter/consumerKey'),
    consumer_secret: config.get('/twitter/consumerSecret'),
    access_token_key: config.get('/twitter/accessTokenKey'),
    access_token_secret: config.get('/twitter/accessTokenSecret')
  })
  .stream('statuses/filter', {
    track: config.get('/twitter/track')
  })
}

TwitterStream.prototype.receive = function () {
  return new Promise((resolve, reject) => {
    this.stream.on('data', resolve).on('error', reject)
  })
}

TwitterStream.prototype.destroy = function () {
  this.stream.destroy()
}

主要问题是当我创建对象时

const stream = new TwitterStream(config)

stream.receive().then((data) => console.log(data))

当我执行时，仅读取一个对象。没有其他数据被传输。

  TwitterStream.prototype.receive = function () {
     return new Promise((resolve, reject) => {
       this.stream
         .on('data', (data) => resolve(data)
         .on('error', (error) => reject(error))
       })
    }

Answer 1

通过使用 Rx 扩展，这非常简单:

TwitterStream.prototype.receive = function () {
    return Rx.Observable.create((observer) => {
        this.stream
            .on('data', (data) => observer.onNext(data))
            .on('error', (err) => observer.onError(err));
    });
}

然后

const stream = new TwitterStream(config)

stream.receive().subscribe((data) => console.log(data));