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

 #!/usr/bin/env node
 /*
  * 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.
  *
  */

 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 MessengerReceive = function(opts, callback) {
     //if (opts.verbose) {
         console.log("subscriptions = " + opts.subscriptions);
         console.log("messageCount = " + opts.messageCount);
         console.log("recvCount = " + opts.recvCount);
         console.log("incomingWindow = " + opts.incomingWindow);
         console.log("reportInterval = " + opts.reportInterval);
         console.log("reply = " + opts.reply);
         console.log("forwardingTargets = " + opts.forwardingTargets);
         console.log("name = " + opts.name);
         console.log("readyText = " + opts.readyText);
         console.log("verbose = " + opts.verbose);
         console.log();
     //}

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

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

     var pumpData = function() {
         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()) {
             // start the timer only after receiving the first msg
             if (received === 0) {
                 stats.start();
             }

             messenger.get(message);
             received += 1;
             //console.log("Address: " + message.getAddress());
             //console.log("CorrelationID: " + message.getCorrelationID());
             //console.log("Content: " + message.body);
             stats.messageReceived(message);

             if (opts.reply) {
                 var replyTo = message.getReplyTo();
                 if (replyTo) {
                     if (opts.verbose) {
                         console.log("Replying to: " + replyTo);
                     }
                     message.setAddress(replyTo);
                     message.setCreationTime(new Date());
                     messenger.put(message);
                     sent += 1;
                 }
             }
         }

         // Check for exit condition.
         if (running && !(opts.messageCount === 0 || received < opts.messageCount)) {
             // Wait for outgoing to be zero before calling stop so pending sends
             // get flushed properly.
             if (messenger.outgoing()) {
                 if (opts.verbose) {
                     console.log("Flushing pending sends");
                 }
             } else {
 //console.log("******* messenger.stop()");
                 messenger.stop();
                 running = false;
                 stats.report(sent, received);
                 if (callback) {
                     callback(stats);
                 }
             }
         }

         if (messenger.isStopped()) {
 //console.log("-------------------- messenger.isStopped()");
             message.free();
             messenger.free();
         }
     };

     this.start = function() {
         messenger.on('error', function(error) {console.log("** error **"); console.log(error);});
         messenger.on('work', pumpData);
         messenger.on('subscription', function(subscription) {
             // Hack to let test scripts know when the receivers are ready (so that the
             // senders may be started).
 console.log("****** subscription " + subscription.getAddress() + " succeeded")
             if (opts.readyText) {
                 console.log(opts.readyText);
             }
         });

         if (opts.incomingWindow) {
             messenger.setIncomingWindow(opts.incomingWindow);
         }
         messenger.start();

         // Unpack addresses that were specified using comma-separated list
         var subscriptions = opts.subscriptions.split(',');
         for (var i = 0; i < subscriptions.length; i++) {
             var subscription = subscriptions[i];
             if (opts.verbose) {
                 console.log("Subscribing to " + subscription);
             }
             messenger.subscribe(subscription);
         }
     };
 };


 // Check if the environment is Node.js and if not log an error and exit.
 if (typeof process === 'object' && typeof require === 'function') {
     var usage =
     'Usage: msgr-recv [OPTIONS]\n' +
     ' -a <addr>[,<addr>]* \tAddresses to listen on [amqp://~0.0.0.0]\n' +
     ' -c # \tNumber of messages to receive before exiting [0=forever]\n' +
     ' -b # \tArgument to Messenger::recv(n) [2048]\n' +
     ' -w # \tSize for incoming window [0]\n' +
     ' -e # \t# seconds to report statistics, 0 = end of test [0] *TBD*\n' +
     ' -R \tSend reply if \'reply-to\' present\n' +
     ' -F <addr>[,<addr>]* \tAddresses used for forwarding received messages\n' +
     ' -N <name> \tSet the container name to <name>\n' +
     ' -X <text> \tPrint \'<text>\\n\' to stdout after all subscriptions are created\n' +
     ' -V \tEnable debug logging\n';

     // Increase the virtual heap available to the emscripten compiled C runtime.
     // This allows us to test a really big string.
     PROTON_TOTAL_MEMORY = 140000000;
     PROTON_TOTAL_STACK = 25000000; // Needs to be bigger than the biggest string.
     var proton = require("qpid-proton-messenger");

     var opts = {};
     opts.subscriptions = 'amqp://~0.0.0.0';
     opts.messageCount = 0;
     opts.recvCount = -1;
     opts.incomingWindow;
     opts.reportInterval = 0;
     opts.reply = false;
     opts.forwardingTargets;
     opts.name;
     opts.readyText;
     opts.verbose = false;

     var args = process.argv.slice(2);
     if (args.length > 0) {
         if (args[0] === '-h' || args[0] === '--help') {
             console.log(usage);
             process.exit(0);
         }

         for (var i = 0; i < args.length; i++) {
             var arg = args[i];
             if (arg.charAt(0) === '-') {
                 if (arg === '-V') {
                     opts.verbose = true;
                 } else if (arg === '-R') {
                     opts.reply = true;
                 } else {
                     i++;
                     var val = args[i];
                     if (arg === '-a') {
                         opts.subscriptions = val;
                     } else if (arg === '-c') {
                         opts.messageCount = val;
                     } else if (arg === '-b') {
                         opts.recvCount = val;
                     } else if (arg === '-w') {
                         opts.incomingWindow = val;
                     } else if (arg === '-e') {
                         opts.reportInterval = val;
                     } else if (arg === '-F') {
                         opts.forwardingTargets = val;
                     } else if (arg === '-N') {
                         opts.name = val;
                     } else if (arg === '-X') {
                         opts.readyText = val;
                     }
                 }
             }
         }
     }

     var receiver = new MessengerReceive(opts);
     receiver.start();
 } else {
     console.error("msgr-recv.js should be run in Node.js");
 }
	#!/usr/bin/env node
	/*
	* 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.
	*
	*/

	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 MessengerReceive = function(opts, callback) {
	//if (opts.verbose) {
	console.log("subscriptions = " + opts.subscriptions);
	console.log("messageCount = " + opts.messageCount);
	console.log("recvCount = " + opts.recvCount);
	console.log("incomingWindow = " + opts.incomingWindow);
	console.log("reportInterval = " + opts.reportInterval);
	console.log("reply = " + opts.reply);
	console.log("forwardingTargets = " + opts.forwardingTargets);
	console.log("name = " + opts.name);
	console.log("readyText = " + opts.readyText);
	console.log("verbose = " + opts.verbose);
	console.log();
	//}

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

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

	var pumpData = function() {
	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()) {
	// start the timer only after receiving the first msg
	if (received === 0) {
	stats.start();
	}

	messenger.get(message);
	received += 1;
	//console.log("Address: " + message.getAddress());
	//console.log("CorrelationID: " + message.getCorrelationID());
	//console.log("Content: " + message.body);
	stats.messageReceived(message);

	if (opts.reply) {
	var replyTo = message.getReplyTo();
	if (replyTo) {
	if (opts.verbose) {
	console.log("Replying to: " + replyTo);
	}
	message.setAddress(replyTo);
	message.setCreationTime(new Date());
	messenger.put(message);
	sent += 1;
	}
	}
	}

	// Check for exit condition.
	if (running && !(opts.messageCount === 0 \|\| received < opts.messageCount)) {
	// Wait for outgoing to be zero before calling stop so pending sends
	// get flushed properly.
	if (messenger.outgoing()) {
	if (opts.verbose) {
	console.log("Flushing pending sends");
	}
	} else {
	//console.log("******* messenger.stop()");
	messenger.stop();
	running = false;
	stats.report(sent, received);
	if (callback) {
	callback(stats);
	}
	}
	}

	if (messenger.isStopped()) {
	//console.log("-------------------- messenger.isStopped()");
	message.free();
	messenger.free();
	}
	};

	this.start = function() {
	messenger.on('error', function(error) {console.log(" error "); console.log(error);});
	messenger.on('work', pumpData);
	messenger.on('subscription', function(subscription) {
	// Hack to let test scripts know when the receivers are ready (so that the
	// senders may be started).
	console.log("****** subscription " + subscription.getAddress() + " succeeded")
	if (opts.readyText) {
	console.log(opts.readyText);
	}
	});

	if (opts.incomingWindow) {
	messenger.setIncomingWindow(opts.incomingWindow);
	}
	messenger.start();

	// Unpack addresses that were specified using comma-separated list
	var subscriptions = opts.subscriptions.split(',');
	for (var i = 0; i < subscriptions.length; i++) {
	var subscription = subscriptions[i];
	if (opts.verbose) {
	console.log("Subscribing to " + subscription);
	}
	messenger.subscribe(subscription);
	}
	};
	};


	// Check if the environment is Node.js and if not log an error and exit.
	if (typeof process === 'object' && typeof require === 'function') {
	var usage =
	'Usage: msgr-recv [OPTIONS]\n' +
	' -a <addr>[,<addr>]* \tAddresses to listen on [amqp://~0.0.0.0]\n' +
	' -c # \tNumber of messages to receive before exiting [0=forever]\n' +
	' -b # \tArgument to Messenger::recv(n) [2048]\n' +
	' -w # \tSize for incoming window [0]\n' +
	' -e # \t# seconds to report statistics, 0 = end of test [0] TBD\n' +
	' -R \tSend reply if \'reply-to\' present\n' +
	' -F <addr>[,<addr>]* \tAddresses used for forwarding received messages\n' +
	' -N <name> \tSet the container name to <name>\n' +
	' -X <text> \tPrint \'<text>\\n\' to stdout after all subscriptions are created\n' +
	' -V \tEnable debug logging\n';

	// Increase the virtual heap available to the emscripten compiled C runtime.
	// This allows us to test a really big string.
	PROTON_TOTAL_MEMORY = 140000000;
	PROTON_TOTAL_STACK = 25000000; // Needs to be bigger than the biggest string.
	var proton = require("qpid-proton-messenger");

	var opts = {};
	opts.subscriptions = 'amqp://~0.0.0.0';
	opts.messageCount = 0;
	opts.recvCount = -1;
	opts.incomingWindow;
	opts.reportInterval = 0;
	opts.reply = false;
	opts.forwardingTargets;
	opts.name;
	opts.readyText;
	opts.verbose = false;

	var args = process.argv.slice(2);
	if (args.length > 0) {
	if (args[0] === '-h' \|\| args[0] === '--help') {
	console.log(usage);
	process.exit(0);
	}

	for (var i = 0; i < args.length; i++) {
	var arg = args[i];
	if (arg.charAt(0) === '-') {
	if (arg === '-V') {
	opts.verbose = true;
	} else if (arg === '-R') {
	opts.reply = true;
	} else {
	i++;
	var val = args[i];
	if (arg === '-a') {
	opts.subscriptions = val;
	} else if (arg === '-c') {
	opts.messageCount = val;
	} else if (arg === '-b') {
	opts.recvCount = val;
	} else if (arg === '-w') {
	opts.incomingWindow = val;
	} else if (arg === '-e') {
	opts.reportInterval = val;
	} else if (arg === '-F') {
	opts.forwardingTargets = val;
	} else if (arg === '-N') {
	opts.name = val;
	} else if (arg === '-X') {
	opts.readyText = val;
	}
	}
	}
	}
	}

	var receiver = new MessengerReceive(opts);
	receiver.start();
	} else {
	console.error("msgr-recv.js should be run in Node.js");
	}