将来自 Twitter 的流式数据与 Meteor 一起使用

Question

到目前为止，我已经能够从 Twitter 上下载流式实时数据。我如何使用这些数据？我正在尝试将其插入集合中，但出现此错误：

错误：Meteor 代码必须始终在 Fiber 中运行。尝试使用 Meteor.bindEnvironment 包装传递给非 Meteor 库的回调。

我尝试用纤维包裹我的代码，但它不起作用/或我没有包裹代码的正确部分。另外，我不确定这是否是在 Meteor 中使用流数据的正确方法。

Posts = new Meteor.Collection('posts');

if (Meteor.isClient) {
  Meteor.call("tweets", function(error, results) {
    console.log(results); //results.data should be a JSON object
  });
}

if (Meteor.isServer) {    
  Meteor.methods({
    tweets: function(){

      Twit = new TwitMaker({
        consumer_key: '...',
        consumer_secret: '...',
        access_token: '...',
        access_token_secret: '...'
      });

      sanFrancisco = [ '-122.75', '36.8', '-121.75', '37.8' ];

      stream = Twit.stream('statuses/filter', { locations: sanFrancisco });


      stream.on('tweet', function (tweet) {
        userName = tweet.user.screen_name;
        userTweet = tweet.text;
        console.log(userName + " says: " + userTweet);
        Posts.insert({post: tweet})

      })  
    }    
  })  
}

Answer 1

我写了一篇关于Building Twitter Monitoring Apps with MeteorJS from Scratch 的长文，包括 Meteor.bindEnvironment 部分，摘录如下。

var Twit = Meteor.npmRequire(‘twit’);
var conf = JSON.parse(Assets.getText(‘twitter.json’)); 
var T = new Twit({
 consumer_key: conf.consumer.key, 
 consumer_secret: conf.consumer.secret,
 access_token: conf.access_token.key, 
 access_token_secret: conf.access_token.secret

// 
// filter the twitter public stream by the word ‘iwatch’. 
//

var stream = T.stream(‘statuses/filter’, { track: conf.keyword })

stream.on(‘tweet’, Meteor.bindEnvironment(function (tweet) {
 console.log(tweet);
 Tweets.insert(tweet);
}))

只添加了两个函数：

Meteor.bindEnvironment()

这个函数帮助我们将一个函数绑定到所有环境变量的当前值。

玩得开心！

Answer 2

这对我有用。本质是从 Twit 回调中调用 Meteor.bindEnvironment。

Meteor.methods({
    consumeTwitter: function () {

    var Twit = Meteor.npmRequire('twit');

    var T = new Twit({
        consumer_key:         'xxx', // API key
        consumer_secret:      'yyy', // API secret
        access_token:         'xxx',
        access_token_secret:  'xxx'
    });

    //  search twitter for all tweets containing the word 'banana'
    var now = new Date().getTime();

    var wrappedInsert = Meteor.bindEnvironment(function(tweet) {
      Tweets.insert(tweet);
    }, "Failed");


    T.get('search/tweets',
        {
            q: 'banana since:2011-11-11',
            count: 4
        },
        function(err, data, response) {
          var statuses = data['statuses'];

          for(var i in statuses) {
             wrappedInsert(statuses[i]);
          }
        }
    )}
});

Answer 3

改变数据库的代码需要在纤程中运行，这就是错误所在。在来自 Meteor 以外的库的回调中运行的代码不会（必然）在 Fiber 中运行，因此您需要包装回调函数以确保它在 Fiber 中运行，或者至少是在 Fiber 中运行的部分与数据库交互。
Meteor.bindEnvironment 目前没有文档记录，但它通常被认为是包装回调的最可靠方法。错误所说的 Meteor.bindEnvironment 在这里定义以供参考： https://github.com/meteor/meteor/blob/master/packages/meteor/dynamics_nodejs.js#L63

这样的事情可能是完成这项工作的最简单方法：

tweets: function() {
  ...

  // You have to define this wrapped function inside a fiber .
  // Meteor.methods always run in a fiber, so we should be good here. 
  // If you define it inside the callback, it will error out at the first
  // line of Meteor.bindEnvironment.

  var wrappedInsert = Meteor.bindEnvironment(function(tweet) {
    Posts.insert(tweet);
  }, "Failed to insert tweet into Posts collection.");

  stream.on('tweet', function (tweet) {
    var userName = tweet.user.screen_name;
    var userTweet = tweet.text;
    console.log(userName + " says: " + userTweet);
    wrappedInsert(tweet);
  });
}