endress callback sample for Node.js++

bootstrap.js

/**
 * Bootstrap
 *
 * An asynchronous boostrap function that runs before your Sails app gets lifted.
 * This gives you an opportunity to set up your data model, run jobs, or perform some special logic.
 *
 * For more information on bootstrapping your app, check out:
 * http://sailsjs.org/#documentation
 */

var socketBind = function( room, onConnect, onDisconnect ){

  var io = sails.io;

  if( typeof(room) != "string" )
    room = "";

  if( onConnect == null )
    onConnect = function( appSocket, roomSocket, socket ){
      console.log("implements function on Connect")
    };

  var roomSocket = io.of( room )
    .on( "connection", function( socket ){

      setTimeout(function(){
        onConnect( io.sockets, roomSocket, socket );
      }, 100 );

      if( onDisconnect != null ){
        socket.on( "disconnect",function(){
          setTimeout(function(){
            onDisconnect( io.sockets, roomSocket, socket );
          }, 100);
        });
      }

    });

  return roomSocket;

}

module.exports.bootstrap = function (cb) {

  // By default: do nothing
  // This is a good place to subscribe a new socket to a room, inform other users that
  // someone new has come online, or any other custom socket.io logic

  var twitterService = sails.services["twitter"]();

  //chat room
  var twitterStreamRoom = "/twitter-stream";
  var twitterIo = socketBind( twitterStreamRoom,
    function onConnect( appSocket, roomSocket, socket ){

      roomSocket.emit("connected", {
        status: "OK"
      });

    }, function onDisconnect( appSocket, roomSocket, socket ){

    });

  //
  (function(){
    // keyword
    var search = "javascript OR node.js OR grunt jQuery OR #js OR #node";
    var spot =  "javascript,node.js,grunt,jQuery,js,node";

      //
//      twitterService.geoSearch( search, function( error, data ){
//
//        sails.log("geo search");
//        sails.log(error);
//        sails.log(data);
//
//        if( error ){
//          twitterIo.emit( "spot", {
//            status: "error",
//            error: error
//          });
//          return;
//        }
//        twitterIo.emit( "spot", data );
//
//      });

    //
    twitterService.geoStream( spot, {

        data: function( error, data ){

          if( error ){
            twitterIo.emit( "spot", {
              status: "error",
              error: error
            });
            return;
          }
          twitterIo.emit( "spot", data );
        },
        end: function(){},
        destroy: function(){}
      }
    );

  })();

  // total count
  var counterRoom = "/counter";
  socketBind( counterRoom,
    function onConnect( appSocket, roomSocket, socket ){

      var count = Object.keys(roomSocket.sockets).length;
      console.log( "application connect count:open is " + count );
      appSocket.emit( "count", { status: "OK", results: { count: socket.manager.rooms[counterRoom].length }});

    },
    function onDisconnect( appSocket, roomSocket, socket ){

      var count = Object.keys(roomSocket.sockets).length;
      console.log( "application connect count:close is " + count );
      appSocket.emit( "count", { status: "OK", results: { count: count }});

    });

  // It's very important to trigger this callback method when you are finished
  // with the bootstrap!  (otherwise your server will never lift, since it's waiting on the bootstrap)
  cb();
};

twitter.js

var TwitterRealtimeService = function( option ){

  var conf =  option || sails.config.twitter;

  var geocoder = require('geocoder');
  var twitter = require('ntwitter');
  var t = new twitter(conf);

  var s = {
    geoData: function(err, data){ /* geo data streaming */ },
    data: function(data){ /* data streaming */ },
    end: function(response) { /*data end */ },
    destroy: function(response) { /*data end */ }
  }

  var that = {};

  // geo 検索
  that.geoSearch = function( search, cb ){

    if ( typeof (cb) != "function" ) cb = function( err, response ){};

    var callback = function( error, data ){

      if( error )
        return cb( error, null );

      data.statuses.forEach(function(tweets){

        if( tweets != null && tweets.user ){
          var location = tweets.user.time_zone || tweets.user.location;

          if( location ){
            console.log( "location is " + location );
            var geocoder = require('geocoder');
            geocoder.geocode( location, cb );
          }
        }
      });

    }

    that.search( search, callback );

  };

  //検索
  that.search = function( search, cb ){

    if( typeof( search ) != "string" ){
      console.log( "search is String" );
      return;
    }
    if ( typeof (cb) != "function" ) cb = function( err, response ){};

    t.get("/search/tweets.json", {
        q: search,
        result_type: "popular",
        count: 50
      }, function( err, data ) {
        cb( err, data );
    });

  }

  // geo ストリーミング
  that.geoStream = function( search, event ){

    var cb =  event.data || s.geoData;

    event.data = function( data ){

      if( data != null && data.user )
      var location = data.user.time_zone || data.user.location;

      if( location ){
        console.log( "location is " + location );
        var geocoder = require('geocoder');
        geocoder.geocode( location, cb );
      }

    }

    that.stream( search, event );

  };

  //ストリーミング
  that.stream = function( search, event ){

    if( typeof( search ) != "string" ){
      console.log( "search is String" );
      return;
    }

    event.data = event.data || s.data;
    event.end = event.end || s.end;
    event.destroy = event.destroy || s.destroy;

    // Twitter Streaming APIを呼び出す
    t.stream("statuses/filter",
      { track: search },
      function(stream) {
        stream.on('data', event.data);
        stream.on("end", event.end );
        stream.on("destroy",event.destroy );
    });
  }

  return that;

}

module.exports = TwitterRealtimeService;