| #!/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. |
| * |
| */ |
| |
| // Check if the environment is Node.js and if not log an error and exit. |
| if (typeof process === 'object' && typeof require === 'function') { |
| var proton = require("qpid-proton-messenger"); |
| |
| var addr = 'guest:guest@localhost:5673'; |
| //var addr = 'localhost:5673'; |
| var address = 'amqp://' + addr; |
| console.log(address); |
| |
| var subscriptionQueue = ''; |
| var count = 0; |
| var start = 0; // Start Time. |
| |
| var message = new proton.Message(); |
| var messenger = new proton.Messenger(); |
| |
| var pumpData = function() { |
| while (messenger.incoming()) { |
| // The second parameter forces Binary payloads to be decoded as strings |
| // this is useful because the broker QMF Agent encodes strings as AMQP |
| // binary, which is a right pain from an interoperability perspective. |
| var t = messenger.get(message, true); |
| //console.log("Address: " + message.getAddress()); |
| //console.log("Content: " + message.body); |
| messenger.accept(t); |
| |
| if (count % 1000 === 0) { |
| var time = +new Date(); |
| console.log("count = " + count + ", duration = " + (time - start) + ", rate = " + ((count*1000)/(time - start))); |
| } |
| |
| sendMessage(); |
| } |
| |
| if (messenger.isStopped()) { |
| message.free(); |
| messenger.free(); |
| } |
| }; |
| |
| var sendMessage = function() { |
| var msgtext = "Message Number " + count; |
| count++; |
| |
| message.setAddress(address + '/' + subscriptionQueue); |
| message.body = msgtext; |
| messenger.put(message); |
| //messenger.settle(); |
| }; |
| |
| messenger.on('error', function(error) {console.log(error);}); |
| messenger.on('work', pumpData); |
| messenger.on('subscription', function(subscription) { |
| var subscriptionAddress = subscription.getAddress(); |
| var splitAddress = subscriptionAddress.split('/'); |
| subscriptionQueue = splitAddress[splitAddress.length - 1]; |
| |
| console.log("Subscription Queue: " + subscriptionQueue); |
| start = +new Date(); |
| sendMessage(); |
| }); |
| |
| //messenger.setOutgoingWindow(1024); |
| messenger.setIncomingWindow(1024); // The Java Broker seems to need this. |
| messenger.recv(); // Receive as many messages as messenger can buffer. |
| messenger.start(); |
| |
| messenger.subscribe('amqp://' + addr + '/#'); |
| } else { |
| console.error("soak.js should be run in Node.js"); |
| } |
| |