tests/javascript/msgr-send-common.js - qpid-proton - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */

 /**
  * This file is essentially a "module" that is common to msgr-send.js and msgr-send.html.
  * It defines the Statistics and MessengerSend classes and if the environment is Node.js
  * it will import qpid-proton-messenger and export MessengerSend for use in msgr-send.js.
  * Because it's really a module/library trying to execute msgr-send-common.js won't
  * itself do anything terribly exciting.
  */

 var Statistics = function() {
     this.startTime = 0;
     this.latencySamples = 0;
     this.latencyTotal = 0;
     this.latencyMin = 0;
     this.latencyMax = 0;
 };

 Statistics.prototype.start = function() {
     this.startTime = +new Date();
 };

 Statistics.prototype.messageReceived = function(msg) {
     var ts = +msg.getCreationTime(); // The + gets the value of the returned Data Object.
     if (ts) {
         var l = +new Date() - ts;
         if (l) {
             this.latencyTotal += l;
             this.latencySamples += 1;
             if (this.latencySamples === 1) {
                 this.latencyMin = this.latencyMax = l;
             } else {
                 if (this.latencyMin > l) {
                     this.latencyMin = l;
                 }
                 if (this.latencyMax < l) {
                     this.latencyMax = l;
                 }
             }
         }
     }
 };

 Statistics.prototype.report = function(sent, received) {
     var seconds = (+new Date() - this.startTime)/1000;
     console.log("Messages sent: " + sent + " received: " + received);
     console.log("Total time: " + seconds + " seconds");
     if (seconds) {
         console.log("Throughput: " + (sent/seconds) + " msgs/sec sent");
         console.log("Throughput: " + (received/seconds) + " msgs/sec received");
     }

     if (this.latencySamples) {
         console.log("Latency (ms): " + this.latencyMin + " min " +
                                        this.latencyMax + " max " +
                                        (this.latencyTotal/this.latencySamples) + " avg");
     }
 };


 var MessengerSend = function(opts, callback) {
     //if (opts.verbose) {
         console.log("addresses = " + opts.addresses);
         console.log("messageCount = " + opts.messageCount);
         console.log("messageSize = " + opts.messageSize);
         console.log("recvCount = " + opts.recvCount);
         console.log("sendBatch = " + opts.sendBatch);
         console.log("outgoingWindow = " + opts.outgoingWindow);
         console.log("reportInterval = " + opts.reportInterval);
         console.log("getReplies = " + opts.getReplies);
         console.log("name = " + opts.name);
         console.log("verbose = " + opts.verbose);
         console.log();
     //}

     var stats = new Statistics();
     var targets = [];
     var running = true; // Used to avoid calling stop multiple times.
     var sent = 0;
     var received = 0;

     var message = new proton.Message();
     var replyMessage = new proton.Message();
     var messenger = new proton.Messenger(opts.name);

     // Retrieve replies and return the number of reply messages received.
     var processReplies = function() {
         var received = 0;
         if (opts.verbose) {
             console.log("Calling messenger.recv(" + opts.recvCount + ")");
         }
         messenger.recv(opts.recvCount);

         if (opts.verbose) {
             console.log("Messages on incoming queue: " + messenger.incoming());
         }
         while (messenger.incoming()) {
             messenger.get(replyMessage);
             received += 1;
             //console.log("Address: " + replyMessage.getAddress());
             //console.log("Content: " + replyMessage.body);
             stats.messageReceived(replyMessage);
         }
         return received;
     };

     // Send messages as fast as possible. This is analogous to the while loop in
     // the Python msgr-send.py but we wrap in a function in JavaScript so that
     // we can schedule on the JavaScript Event queue via setTimeout. This is needed
     // otherwise the JavaScript Event loop is blocked and no data gets sent.
     var sendData = function() {
         var delay = 0;
         while (opts.messageCount === 0 || (sent < opts.messageCount)) {
             // Check the amount of data buffered on the socket, if it's non-zero
             // exit the loop and call senData again after a short delay. This
             // will throttle the send rate if necessary.
             if (messenger.getBufferedAmount()) {
 console.log("messenger.getBufferedAmount() = " + messenger.getBufferedAmount());
                 delay = 100;
                 break; // Exit loop to check for exit condition and schedule to Event queue.
             }

             var index = sent % targets.length;
 //console.log("sent = " + sent + ", index = " + index);

             message.setAddress(targets[index]);
             message.setCorrelationID(sent);
             message.setCreationTime(new Date());
             messenger.put(message);
             sent += 1;

             if (opts.sendBatch && (messenger.outgoing() >= opts.sendBatch)) {
                 if (opts.verbose) {
                     console.log("Calling messenger.send()")
                 }
                 messenger.send();

                 if (opts.getReplies) {
                     received += processReplies();
                 }
                 break; // Exit loop to check for exit condition and yield to Event loop.
             }
         }

         // Check for exit condition.
         if (running && !(opts.messageCount === 0 || (sent < opts.messageCount))) {
             if (opts.getReplies && (received < sent)) {
                 received += processReplies();
                 if (opts.verbose) {
                     console.log("Messages sent = " + sent + ", received = " + received);
                 }
             } else if (messenger.outgoing()) {
                 if (opts.verbose) {
                     console.log("Flushing pending sends");
                 }
                 messenger.send();
             } else {
 //console.log("******* calling stop")
                 messenger.stop();
                 running = false;
                 stats.report(sent, received);
                 if (callback) {
                     callback(stats);
                 }
             }
         }

         if (messenger.isStopped()) {
 //console.log("-------------------- messenger.isStopped()");
             message.free();
             messenger.free();
         } else {
             // schedule next call on the JavaScript Event queue. If we don't do this
             // our messages won't get sent because none of the internal JavaScript
             // network code will get any CPU.

             // If available we call setImmediate rather than setTimeout when the delay
             // is zero. setImmediate is more efficient, in particular I noticed that
             // with Node.js v0.10.18 I could get max throughput and max out CPU using
             // setTimeout, but when I upgraded to v0.10.33 my throughput dropped and
             // my CPU was hovering around 55% but using setImmediate the performance
             // improved again. My guess is that v0.10.18 was checking for zero delay
             // and calling setImmediate internally whereas v0.10.33 wasn't, but I
             // can't say for sure. TODO it's possible that some browsers might do a
             // better job with setImmediate too (given what I'm seeing with Node.js),
             // Chrome might be one such case, but it's not universally supported.
             // It might be worth adding a proper polyfill to handle this.
             if (delay === 0 && typeof setImmediate === 'function') {
                 setImmediate(sendData);
             } else {
                 setTimeout(sendData, delay);
             }
         }
     };

     this.start = function() {
         message.body = Array(+opts.messageSize + 1).join('X');
         message.setReplyTo('~');

         messenger.on('error', function(error) {
             console.log(error);
             opts.messageCount = -1; // Force exit condition.
         });

         if (opts.outgoingWindow) {
             messenger.setOutgoingWindow(opts.outgoingWindow);
         }
         messenger.start();

         // Unpack targets that were specified using comma-separated list
         var addresses = opts.addresses.split(',');
         for (var i = 0; i < addresses.length; i++) {
             var address = addresses[i];
             targets.push(address);
         }

         stats.start();
         sendData();
     };
 };

 // If running in Node.js import the proton library and export MessengerSend.
 if (typeof module === 'object') {
     var proton = require("qpid-proton-messenger");
     module.exports.MessengerSend = MessengerSend;
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/

	/**
	* This file is essentially a "module" that is common to msgr-send.js and msgr-send.html.
	* It defines the Statistics and MessengerSend classes and if the environment is Node.js
	* it will import qpid-proton-messenger and export MessengerSend for use in msgr-send.js.
	* Because it's really a module/library trying to execute msgr-send-common.js won't
	* itself do anything terribly exciting.
	*/

	var Statistics = function() {
	this.startTime = 0;
	this.latencySamples = 0;
	this.latencyTotal = 0;
	this.latencyMin = 0;
	this.latencyMax = 0;
	};

	Statistics.prototype.start = function() {
	this.startTime = +new Date();
	};

	Statistics.prototype.messageReceived = function(msg) {
	var ts = +msg.getCreationTime(); // The + gets the value of the returned Data Object.
	if (ts) {
	var l = +new Date() - ts;
	if (l) {
	this.latencyTotal += l;
	this.latencySamples += 1;
	if (this.latencySamples === 1) {
	this.latencyMin = this.latencyMax = l;
	} else {
	if (this.latencyMin > l) {
	this.latencyMin = l;
	}
	if (this.latencyMax < l) {
	this.latencyMax = l;
	}
	}
	}
	}
	};

	Statistics.prototype.report = function(sent, received) {
	var seconds = (+new Date() - this.startTime)/1000;
	console.log("Messages sent: " + sent + " received: " + received);
	console.log("Total time: " + seconds + " seconds");
	if (seconds) {
	console.log("Throughput: " + (sent/seconds) + " msgs/sec sent");
	console.log("Throughput: " + (received/seconds) + " msgs/sec received");
	}

	if (this.latencySamples) {
	console.log("Latency (ms): " + this.latencyMin + " min " +
	this.latencyMax + " max " +
	(this.latencyTotal/this.latencySamples) + " avg");
	}
	};


	var MessengerSend = function(opts, callback) {
	//if (opts.verbose) {
	console.log("addresses = " + opts.addresses);
	console.log("messageCount = " + opts.messageCount);
	console.log("messageSize = " + opts.messageSize);
	console.log("recvCount = " + opts.recvCount);
	console.log("sendBatch = " + opts.sendBatch);
	console.log("outgoingWindow = " + opts.outgoingWindow);
	console.log("reportInterval = " + opts.reportInterval);
	console.log("getReplies = " + opts.getReplies);
	console.log("name = " + opts.name);
	console.log("verbose = " + opts.verbose);
	console.log();
	//}

	var stats = new Statistics();
	var targets = [];
	var running = true; // Used to avoid calling stop multiple times.
	var sent = 0;
	var received = 0;

	var message = new proton.Message();
	var replyMessage = new proton.Message();
	var messenger = new proton.Messenger(opts.name);

	// Retrieve replies and return the number of reply messages received.
	var processReplies = function() {
	var received = 0;
	if (opts.verbose) {
	console.log("Calling messenger.recv(" + opts.recvCount + ")");
	}
	messenger.recv(opts.recvCount);

	if (opts.verbose) {
	console.log("Messages on incoming queue: " + messenger.incoming());
	}
	while (messenger.incoming()) {
	messenger.get(replyMessage);
	received += 1;
	//console.log("Address: " + replyMessage.getAddress());
	//console.log("Content: " + replyMessage.body);
	stats.messageReceived(replyMessage);
	}
	return received;
	};

	// Send messages as fast as possible. This is analogous to the while loop in
	// the Python msgr-send.py but we wrap in a function in JavaScript so that
	// we can schedule on the JavaScript Event queue via setTimeout. This is needed
	// otherwise the JavaScript Event loop is blocked and no data gets sent.
	var sendData = function() {
	var delay = 0;
	while (opts.messageCount === 0 \|\| (sent < opts.messageCount)) {
	// Check the amount of data buffered on the socket, if it's non-zero
	// exit the loop and call senData again after a short delay. This
	// will throttle the send rate if necessary.
	if (messenger.getBufferedAmount()) {
	console.log("messenger.getBufferedAmount() = " + messenger.getBufferedAmount());
	delay = 100;
	break; // Exit loop to check for exit condition and schedule to Event queue.
	}

	var index = sent % targets.length;
	//console.log("sent = " + sent + ", index = " + index);

	message.setAddress(targets[index]);
	message.setCorrelationID(sent);
	message.setCreationTime(new Date());
	messenger.put(message);
	sent += 1;

	if (opts.sendBatch && (messenger.outgoing() >= opts.sendBatch)) {
	if (opts.verbose) {
	console.log("Calling messenger.send()")
	}
	messenger.send();

	if (opts.getReplies) {
	received += processReplies();
	}
	break; // Exit loop to check for exit condition and yield to Event loop.
	}
	}

	// Check for exit condition.
	if (running && !(opts.messageCount === 0 \|\| (sent < opts.messageCount))) {
	if (opts.getReplies && (received < sent)) {
	received += processReplies();
	if (opts.verbose) {
	console.log("Messages sent = " + sent + ", received = " + received);
	}
	} else if (messenger.outgoing()) {
	if (opts.verbose) {
	console.log("Flushing pending sends");
	}
	messenger.send();
	} else {
	//console.log("******* calling stop")
	messenger.stop();
	running = false;
	stats.report(sent, received);
	if (callback) {
	callback(stats);
	}
	}
	}

	if (messenger.isStopped()) {
	//console.log("-------------------- messenger.isStopped()");
	message.free();
	messenger.free();
	} else {
	// schedule next call on the JavaScript Event queue. If we don't do this
	// our messages won't get sent because none of the internal JavaScript
	// network code will get any CPU.

	// If available we call setImmediate rather than setTimeout when the delay
	// is zero. setImmediate is more efficient, in particular I noticed that
	// with Node.js v0.10.18 I could get max throughput and max out CPU using
	// setTimeout, but when I upgraded to v0.10.33 my throughput dropped and
	// my CPU was hovering around 55% but using setImmediate the performance
	// improved again. My guess is that v0.10.18 was checking for zero delay
	// and calling setImmediate internally whereas v0.10.33 wasn't, but I
	// can't say for sure. TODO it's possible that some browsers might do a
	// better job with setImmediate too (given what I'm seeing with Node.js),
	// Chrome might be one such case, but it's not universally supported.
	// It might be worth adding a proper polyfill to handle this.
	if (delay === 0 && typeof setImmediate === 'function') {
	setImmediate(sendData);
	} else {
	setTimeout(sendData, delay);
	}
	}
	};

	this.start = function() {
	message.body = Array(+opts.messageSize + 1).join('X');
	message.setReplyTo('~');

	messenger.on('error', function(error) {
	console.log(error);
	opts.messageCount = -1; // Force exit condition.
	});

	if (opts.outgoingWindow) {
	messenger.setOutgoingWindow(opts.outgoingWindow);
	}
	messenger.start();

	// Unpack targets that were specified using comma-separated list
	var addresses = opts.addresses.split(',');
	for (var i = 0; i < addresses.length; i++) {
	var address = addresses[i];
	targets.push(address);
	}

	stats.start();
	sendData();
	};
	};

	// If running in Node.js import the proton library and export MessengerSend.
	if (typeof module === 'object') {
	var proton = require("qpid-proton-messenger");
	module.exports.MessengerSend = MessengerSend;
	}