// reflector.js
//
var parseurl = require( './parse-url' ),
persistence = require( './persistence' ),
helpers = require( './helpers' ),
fs = require( 'fs' );
function parseSocketUrl( socket ) {
try
{
var referer = require('url')
.parse(socket.handshake.url)
.query;
referer = require('querystring')
.parse(referer)
.pathname;
var namespace = referer;
if(!namespace) return null;
if (namespace[namespace.length - 1] != "/")
namespace += "/";
return parseurl.Process( namespace );
}
catch (e)
{
return null;
}
}
function GetLoadForSocket( processedURL ) {
if ( processedURL[ 'private_path' ] ) {
return persistence.GetLoadInformation( processedURL );
}
return { 'save_name': undefined, 'save_revision': undefined, 'explicit_revision': undefined, 'application_path': undefined };
}
//Get the instance ID from the handshake headers for a socket
function GetNamespace( processedURL ) {
if ( ( processedURL[ 'instance' ] ) && ( processedURL[ 'public_path' ] ) ) {
return helpers.JoinPath( processedURL[ 'public_path' ], processedURL[ 'application' ], processedURL[ 'instance' ] );
}
return undefined;
}
function GetNow( ) {
return new Date( ).getTime( ) / 1000.0;
}
function OnConnection( socket ) {
var processedURL = parseSocketUrl( socket );
if (processedURL == null) {
setInterval(function() {
var address = socket.conn.request.headers.host;
var obj = {};
for (var prop in global.instances) {
obj[prop] = {
"instance":address + prop,
"clients": Object.keys(global.instances[prop].clients).length
};
}
var json = JSON.stringify(obj);
socket.emit('getWebAppUpdate', json);
}, 3000);
// socket.on('getWebAppUpdate', function(msg){
// });
return
}
//get instance for new connection
var namespace = GetNamespace( processedURL );
if ( namespace == undefined ) {
return;
}
//prepare for persistence request in case that's what this is
var loadInfo = GetLoadForSocket( processedURL );
var saveObject = persistence.LoadSaveObject( loadInfo );
//if it's a new instance, setup record
if( !global.instances[ namespace ] ) {
global.instances[ namespace ] = { };
global.instances[ namespace ].clients = { };
global.instances[ namespace ].pendingList = [ ];
global.instances[ namespace ].start_time = undefined;
global.instances[ namespace ].pause_time = undefined;
global.instances[ namespace ].rate = 1.0;
global.instances[ namespace ].setTime = function( time ) {
this.start_time = GetNow( ) - time;
this.pause_time = undefined;
this.rate = 1.0;
};
global.instances[ namespace ].isPlaying = function( ) {
if ( ( this.start_time != undefined ) && ( this.pause_time == undefined ) ) {
return true;
}
return false
};
global.instances[ namespace ].isPaused = function( ) {
if ( ( this.start_time != undefined ) && ( this.pause_time != undefined ) ) {
return true;
}
return false
};
global.instances[ namespace ].isStopped = function( ) {
if ( this.start_time == undefined ) {
return true;
}
return false;
};
global.instances[ namespace ].getTime = function( ) {
if ( this.isPlaying( ) ) {
return ( GetNow( ) - this.start_time ) * this.rate;
} else if ( this.isPaused( ) ) {
return ( this.pause_time - this.start_time ) * this.rate;
}
else {
return 0.0;
}
};
global.instances[ namespace ].play = function( ) {
if ( this.isStopped( ) ) {
this.start_time = GetNow( );
this.pause_time = undefined;
} else if ( this.isPaused( ) ) {
this.start_time = this.start_time + ( GetNow( ) - this.pause_time );
this.pause_time = undefined;
}
};
global.instances[ namespace ].pause = function( ) {
if ( this.isPlaying( ) ) {
this.pause_time = GetNow( );
}
};
global.instances[ namespace ].stop = function( ) {
if ( ( this.isPlaying( ) ) || ( this.isPaused( ) ) ) {
this.start_time = undefined;
this.pause_time = undefined;
}
};
global.instances[ namespace ].setTime( 0.0 );
if ( saveObject ) {
if ( saveObject[ "queue" ] ) {
if ( saveObject[ "queue" ][ "time" ] ) {
global.instances[ namespace ].setTime( saveObject[ "queue" ][ "time" ] );
}
}
}
global.instances[ namespace ].state = { };
var log;
function generateLogFile() {
try {
if ( !fs.existsSync( './/log/' ) ) {
fs.mkdir( './/log/', function ( err ) {
if ( err ) {
console.log ( err );
}
})
}
log = fs.createWriteStream( './/log/' + namespace.replace( /[\\\/]/g, '_' ), { 'flags': 'a' } );
} catch( err ) {
console.log( 'Error generating Node Server Log File\n');
}
}
global.instances[ namespace ].Log = function ( message, level ) {
if( global.logLevel >= level ) {
if ( !log ) {
generateLogFile();
}
log.write( message + '\n' );
global.log( message + '\n' );
}
};
global.instances[ namespace ].Error = function ( message, level ) {
var red, brown, reset;
red = '\u001b[31m';
brown = '\u001b[33m';
reset = '\u001b[0m';
if ( global.logLevel >= level ) {
if ( !log ) {
generateLogFile();
}
log.write( message + '\n' );
global.log( red + message + reset + '\n' );
}
};
//keep track of the timer for this instance
global.instances[ namespace ].timerID = setInterval( function ( ) {
var message = { parameters: [ ], time: global.instances[ namespace ].getTime( ) };
for ( var i in global.instances[ namespace ].clients ) {
var client = global.instances[ namespace ].clients[ i ];
if ( ! client.pending ) {
client.emit( 'message', message );
}
}
if(global.instances[ namespace ]){
if ( global.instances[ namespace ].pendingList.pending ) {
global.instances[ namespace ].pendingList.push( message );
}
}
}, 50 );
}
//add the new client to the instance data
global.instances[ namespace ].clients[ socket.id ] = socket;
socket.pending = true;
//Get the descriptor for the `clients.vwf` child.
var clientDescriptor = GetClientDescriptor( socket );
// The time for the setState message should be the time the new client joins, so save that time
var setStateTime = global.instances[ namespace ].getTime( );
// If this client is the first, it can just load the application, and mark it not pending
if ( Object.keys( global.instances[ namespace ].clients ).length === 1 ) {
if ( saveObject ) {
socket.emit( 'message', {
action: "setState",
parameters: [ saveObject ],
time: global.instances[ namespace ].getTime( )
} );
}
else {
var instance = namespace;
//Get the state and load it.
//Now the server has a rough idea of what the simulation is
socket.emit( 'message', {
action: "createNode",
parameters: [ "http://vwf.example.com/clients.vwf" ],
time: global.instances[ namespace ].getTime( )
} );
socket.emit( 'message', {
action: "createNode",
parameters: [
( processedURL.public_path === "/" ? "" : processedURL.public_path ) + "/" + processedURL.application,
"application"
],
time: global.instances[ namespace ].getTime( )
} );
}
socket.pending = false;
//xapi.logClient( saveObject, loadInfo[ 'application_path' ], loadInfo[ 'save_name' ], namespace, clientDescriptor.properties || {}, true, true );
}
else { //this client is not the first, we need to get the state and mark it pending
if ( ! global.instances[ namespace ].pendingList.pending ) {
var firstclient = Object.keys( global.instances[ namespace ].clients )[ 0 ];
firstclient = global.instances[ namespace ].clients[ firstclient ];
firstclient.emit( 'message', {
action: "getState",
respond: true,
time: global.instances[ namespace ].getTime( )
} );
global.instances[ namespace ].Log( 'GetState from Client', 2 );
global.instances[ namespace ].pendingList.pending = true;
}
socket.pending = true;
}
//Create a child in the application's 'clients.vwf' global to represent this client.
var clientNodeMessage = {
action: "createChild",
parameters: [ "http://vwf.example.com/clients.vwf", socket.id, clientDescriptor ],
time: global.instances[ namespace ].getTime( )
};
// Send messages to all the existing clients (that are not pending),
// telling them to create a new node under the "clients" parent for the new client
for ( var i in global.instances[ namespace ].clients ) {
var client = global.instances[ namespace ].clients[ i ];
if ( !client.pending ) {
client.emit ( 'message', clientNodeMessage );
}
}
if ( global.instances[ namespace ].pendingList.pending ) {
global.instances[ namespace ].pendingList.push( clientNodeMessage );
}
socket.on( 'message', function ( msg ) {
//need to add the client identifier to all outgoing messages
try {
var message = JSON.parse( msg );
}
catch ( e ) {
console.error( "Error on socket message: ", e );
return;
}
message.client = socket.id;
message.time = global.instances[ namespace ].getTime( );
if ( message.result === undefined ) {
//distribute message to all clients on given instance
for ( var i in global.instances[ namespace ].clients ) {
var client = global.instances[ namespace ].clients[ i ];
//just a regular message, so push if the client is pending a load, otherwise just send it.
if ( ! client.pending ) {
client.emit( 'message', message );
}
}
if ( global.instances[ namespace ].pendingList.pending ) {
global.instances[ namespace ].pendingList.push( message );
}
} else if ( message.action == "getState" ) {
//distribute message to all clients on given instance
for ( var i in global.instances[ namespace ].clients ) {
var client = global.instances[ namespace ].clients[ i ];
//if the message was get state, then fire all the pending messages after firing the setState
if ( client.pending ) {
global.instances[ namespace ].Log( 'Got State', 2 );
var state = message.result;
global.instances[ namespace ].Log( state, 2 );
client.emit( 'message', { action: "setState", parameters: [ state ], time: setStateTime } );
client.pending = false;
for ( var j = 0; j < global.instances[ namespace ].pendingList.length; j++ ) {
client.emit( 'message', global.instances[ namespace ].pendingList[ j ] );
}
//xapi.logClient( state, undefined, undefined, namespace, GetClientDescriptor( client ).properties || {}, true, false );
}
}
global.instances[ namespace ].pendingList = [ ];
} else if ( message.action === "execute" ) {
var evaluation = socket.pendingEvaluations && socket.pendingEvaluations.shift();
if ( evaluation ) {
evaluation.resolve( message.result );
clearTimeout( evaluation.timeout );
}
}
} );
// When a client disconnects, go ahead and remove the instance data
socket.on( 'disconnect', function ( ) {
// Remove the disconnecting client
var leavingClient = global.instances[ namespace ].clients[ socket.id ];
global.instances[ namespace ].clients[ socket.id ] = null;
delete global.instances[ namespace ].clients[ socket.id ];
if ( leavingClient.pendingEvaluations ) {
leavingClient.pendingEvaluations.forEach( function( evaluation ) {
evaluation.reject( new Error( "connection closed" ) );
clearTimeout( evaluation.timeout );
} );
}
// Notify others of the disconnecting client. Delete the child representing this client in the application's `clients.vwf` global.
var clientMessage = { action: "deleteChild", parameters: [ "http://vwf.example.com/clients.vwf", socket.id ], time: global.instances[ namespace ].getTime( ) };
for ( var i in global.instances[ namespace ].clients ) {
var client = global.instances[ namespace ].clients[ i ];
if ( ! client.pending ) {
client.emit ( 'message', clientMessage );
}
}
if ( global.instances[ namespace ].pendingList.pending ) {
global.instances[ namespace ].pendingList.push( clientMessage );
}
// If it's the last client, delete the data and the timer
if ( Object.keys( global.instances[ namespace ].clients ).length == 0 ) {
clearInterval( global.instances[ namespace ].timerID );
delete global.instances[ namespace ];
// xapi.logClient( undefined, loadInfo[ 'application_path' ], loadInfo[ 'save_name' ], namespace, clientDescriptor.properties || {}, false, true );
} else {
// xapi.logClient( undefined, loadInfo[ 'application_path' ], loadInfo[ 'save_name' ], namespace, clientDescriptor.properties || {}, false, false );
}
} );
}
function Evaluate( namespace, node, expression ) {
return new Promise( function( resolve, reject ) {
var firstClientID = Object.keys( global.instances[ namespace ].clients )[ 0 ];
var firstClient = global.instances[ namespace ].clients[ firstClientID ];
if ( firstClient ) {
firstClient.pendingEvaluations = firstClient.pendingEvaluations || [];
firstClient.pendingEvaluations.push( {
resolve: resolve,
reject: reject,
timeout: setTimeout( function() { reject( new Error( "timeout" ) ) }, 1000 ),
} );
firstClient.emit( "message", { node: node, action: "execute", parameters: [ expression ], respond: true, time: global.instances[ namespace ].getTime() } );
} else {
reject( new Error( "no clients are connected" ) );
}
} );
}
/// Get a descriptor for the `clients.vwf` child for a new client. An authenticator may set a
/// descriptor in the session at `session.vwf.client`. If the authenticator doesn't provide a
/// descriptor, use an empty node inheriting from `client.vwf`.
function GetClientDescriptor( socket ) {
// socket.io doesn't provide access to the request and the session, but we do have the cookies.
// Create a mock request and run it through the session middleware to recreate the session. This
// creates a session object at `mockRequest.session`.
var mockRequest = {
headers: { cookie: socket.handshake.headers.cookie },
connection: {},
session: {},
};
var mockResponse = {
getHeader: function() {},
setHeader: function() {},
};
sessionStack.forEach( function( middleware ) {
middleware( mockRequest, mockResponse, function() {} );
} );
// Get the descriptor from `vwf.client` in the session.
var descriptor = ( mockRequest.session.vwf || {} ).client || {};
// Set the default prototype.
if ( ! descriptor.extends ) {
descriptor.extends = "http://vwf.example.com/client.vwf";
}
return descriptor;
}
/// Middleware stack to parse a cookie session from `req.headers.cookie` into `req.session`.
var sessionStack = [
//cookieParser(),
//cookieSession( { secret: config.get( 'session.secret' ) } ),
];
function GetInstances() {
return global.instances;
}
exports.OnConnection = OnConnection;
exports.Evaluate = Evaluate;
exports.GetInstances = GetInstances;