如何使用emit（）和sync（）从Storm spout输出元组流？

Question

（xpost github问题）

这是我目前为止的尝试。我想我有一些配置错误，因为我知道我的wsemitter正在发射future事件，但我的Storm UI显示没有喷出发射。

我想也许我不应该在spout函数中绑定监听器？

该函数是否被多次调用？（看起来像是...参见https://github.com/rallysoftware/node-storm/blob/master/lib/spout.js#L4)

var storm = require('node-storm');
var wsEmitter = require('./wsEmitter.js')();
wsEmitter.init();  // subscribe to websocket

var futuresSpout = storm.spout(function(sync) {
  var self = this;
  console.log('subscribing to ws');
  wsEmitter.on('future', function(data){       // websocket data arrived
    self.emit([data]);
    sync();
  });
})
.declareOutputFields(["a"]);

Answer 1

结果我有两个问题。首先，我的拓扑没有执行，因为我的一个bolt（未显示）无法设置.declareoutputfields()。

其次，我需要延迟喷口的发射，直到主管使用NextTick()请求一个发射。我通过缓冲所有传入的消息来做到这一点,直到主管调用spout：

module.exports = (function(){
  var storm = require('node-storm');

  var wsEmitter = require('./wsEmitter.js')();
  wsEmitter.init();

  var queue = [];
  var queueEmpty = true;

  wsEmitter.on('thing', function(data){
    var trade = JSON.parse(data);
    trade.timeReported = new Date().valueOf();
    queue.push(trade);
    queueEmpty = false;
  });

  return storm.spout(function(sync) {
    var self = this;
    setTimeout(function(){
      if(!queueEmpty){
        self.emit([queue.shift()]);
        queueEmpty =
        ( queue.length === 0
        ? true
        : false )
      }
      sync();
    }, 100);

  })
  .declareOutputFields(['trade'])
})()