Push Earthquake Early Warning through WebSocket.

streaming.js

/**
 * Mitsuaki Ishimoto
 * Push a message of Earthquake Early Warning
 * Using WebSocket-Node
 * @see: https://github.com/Worlize/WebSocket-Node
/*=================================================================*/
var sys   = require('sys');
var eew   = require('../eew.js');
var https = require('https');

var delivers = {};


/**
 * extend default WebSocketServer for broadcast
 */
var WebSocketServer = require('websocket').server;
WebSocketServer.prototype.broadcastUTF = function(data){
    this.connections.forEach(function(connection, i, ary){
        if(connection.connected){
            connection.sendUTF(data);
        }
    });
};


/**
 * websocket server for end users
 */
var server = require('http').createServer(
    function(request, response) {
        //console.log((new Date()) + " Received request for " + request.url);
        response.writeHead(404);
        response.end();
    }
);
server.listen(
    eew.env['port'],
    function() {
        console.log((new Date()) + " Server is listening on port 80");
    }
);

wsServer = new WebSocketServer({
    "httpServer"            : server,
    "autoAcceptConnections" : true
});

wsServer.on(
    'connect',
    function(connection){
        //console.log((new Date()) + " Connection accepted.");
        connection.sendUTF('{"status" : "accepted"}');
    }
);


/**
 * http request to twitter
 */
var auth = 'Basic ' + new Buffer(eew.env['username'] + ':' + eew.env['password']).toString('base64');
var options = {
    "host"          : "stream.twitter.com",
    "port"          : 443,
    "path"          : '/1/statuses/filter.json?follow=16052553',
    "method"        : 'GET',
    "Authorization" : auth
};

var request = https.request(
    options,
    function(response) {
        console.log("statusCode: ", response.statusCode);
        console.log("headers: ", response.headers);
        sys.puts('response');
        var chunk = '';
        response.on(
            'data',
            function(data){
                sys.puts("data:" + data);
                sys.puts("data.length:" + data.length);
                try{
                    data = "" + data;
                    data = data.replace(/\n|\r/gm, '').replace(/^\s*/, '');
                    sys.puts("data:" + data);
                    sys.puts("data.length:" + data.length);
                    var json = JSON.parse(data);
                    if(json['user']['id'] === 16052553){
                        delivers['id_str' + json['id_str']] = true;
                        wsServer.broadcastUTF(data);
                    }
                    chunk = '';
                }
                catch(e){
                    chunk += data;
                    sys.puts("e:" + e);
                    sys.puts("chunk:" + chunk);
                    sys.puts("chunk.length:" + chunk.length);
                    try{
                        var json = JSON.parse(chunk);
                        if(json['user']['id'] === 16052553){
                            delivers['id_str' + json['id_str']] = true;
                            wsServer.broadcastUTF(chunk);
                        }
                        chunk = '';
                    }
                    catch(e2){
                        wsServer.broadcastUTF(" ");
                        if(chunk.length > 10000){
                            chunk = '';
                        }
                    }
                }
            }
        );
        response.on(
            'end',
            function(){
                sys.puts('response::twitter - response - end');
                throw new Error('response::twitter - response - end');
            }
        );
        response.on(
            'close',
            function(){
                sys.puts('response::twitter - response - close');
                throw new Error('response::twitter - response - close');
            }
        );
    }
);
request["_headers"]["Authorization"] = auth;
request["_headerNames"]["Authorization"] = 'Authorization';
request.end();



/**
 * connection to other servers
 */
/*
var WebSocketClient = require('websocket').client;
var OtherServerConnection = function(host){
    sys.puts('called-otherserverconnections');
    this.host = host;
    this.client;
    var self = this;
    this.construct = function(){
         self.client = new WebSocketClient();
         self.client.on(
             'connect',
             function(conn){
                 conn.on(
                     'error',
                     function(){
                         setTimeout(function(){
                             self.construct(self.host);
                         }, 5000);
                     }
                 );
                 conn.on(
                     'close',
                     function(){
                         setTimeout(function(){
                             self.construct(self.host);
                         }, 5000);
                     }
                 );
                 conn.on(
                     'message',
                     function(msg){
                         //wsServer.broadcastUTF(msg.utf8Data);
                         sys.puts("others-d:" + msg.utf8Data);
                         try{
                             var json = JSON.parse(msg.utf8Data);
                             if(!(delivers['id_str' + json['id_str']] === true)){
                                 sys.puts('deliver-info-from-others');
                                 delivers['id_str' + json['id_str']] = true;
                                 wsServer.broadcastUTF(msg.utf8Data);
                             }
                         }
                         catch(e){// parse error
                             //wsServer.broadcastUTF(' ');// unlimited echo loop
                         }
                     }
                 );
             }
         );
         self.client.on(
             'connectFailed',
             function(){
                 sys.puts('connectFailed');
                 setTimeout(function(){
                     self.construct(self.host);
                 }, 5000);
             }
         );
         self.client.connect('ws://' + self.host);
    }
    self.construct();
}
var conns = [];
for(var i = 0; i < eew.env['others'].length; i++){
    conns[i] = new OtherServerConnection(eew.env['others'][i]);
}
*/