amqp - npm Package Compare versions

Comparing version 0.2.5 to 0.2.6

lib/connection.js

@@ -83,8 +83,8 @@ 'use strict';
   var sslo = (options && options.ssl && options.ssl.enabled) ? defaultSslOptions : {};
-  this.options = _.extend({}, defaultOptions, sslo, urlo, options || {});
-  this.options.clientProperties =  _.extend({}, defaultClientProperties, (options && options.clientProperties) || {});
+  this.options = _.assignIn({}, defaultOptions, sslo, urlo, options || {});
+  this.options.clientProperties =  _.assignIn({}, defaultClientProperties, (options && options.clientProperties) || {});
 };
 
 Connection.prototype.setImplOptions = function (options) {
-  this.implOptions = _.extend({}, defaultImplOptions, options || {});
+  this.implOptions = _.assignIn({}, defaultImplOptions, options || {});
 };
@@ -265,3 +265,3 @@
       // Reconnect any channels which were open.
-      _.each(self.channels, function(channel, index) {
+      _.forEach(self.channels, function(channel, index) {
         if (index !== '0') channel.reconnect();
@@ -402,4 +402,5 @@ });
     this._inboundHeartbeatTimer = setTimeout(function () {
-      if(self.socket.readable)
+      if(self.socket.readable || self.options.heartbeatForceReconnect){
         self.emit('error', new Error('no heartbeat or data in last ' + gracePeriod + ' seconds'));
+      }
     }, gracePeriod * 1000);
@@ -487,2 +488,4 @@ }
           maxFrameBuffer = args.frameMax;
+          this._sendBuffer = new Buffer(maxFrameBuffer);
+          this.parser.setMaxFrameBuffer(maxFrameBuffer);
       }
@@ -631,3 +634,3 @@ if (args.channelMax) {
     debug && debug('making ssl connection');
-    options = _.extend(options, this._getSSLOptions());
+    options = _.assignIn(options, this._getSSLOptions());
     this.socket = tls.connect(options, resetConnectionTimeout);
@@ -655,3 +658,3 @@ } else {
   var events = ['close', 'connect', 'data', 'drain', 'error', 'end', 'secureConnect', 'timeout'];
-  _.each(events, function(event){
+  _.forEach(events, function(event){
     self.socket.on(event, self.emit.bind(self, event));
@@ -662,3 +665,3 @@ });
   var methods = ['destroy', 'write', 'pause', 'resume', 'setEncoding', 'ref', 'unref', 'address'];
-  _.each(methods, function(method){
+  _.forEach(methods, function(method){
     self[method] = function(){
@@ -734,3 +737,3 @@ self.socket[method].apply(self.socket, arguments);
 
-  properties = _.extend(props || {}, properties);
+  properties = _.assignIn(props || {}, properties);
 
@@ -737,0 +740,0 @@ this._sendHeader(channel, buffer.length, properties);

lib/exchange.js

@@ -102,3 +102,3 @@ 'use strict';
             , durable:    !!this.options.durable
-            , auto_delete: !!this.options.autoDelete
+            , autoDelete: !!this.options.autoDelete
             , internal:   !!this.options.internal
@@ -246,3 +246,3 @@ , noWait:     false
 
-  options = _.extend({}, options || {});
+  options = _.assignIn({}, options || {});
   options.routingKey = routingKey;
@@ -429,2 +429,1 @@ options.exchange   = self.name;
 };
-

lib/parser.js

@@ -19,3 +19,3 @@ 'use strict';
 
-var maxFrameBuffer = 131072; // 128k, same as rabbitmq (which was
+var MAX_FRAME_BUFFER_DEFAULT = 131072; // 128k, same as rabbitmq (which was
                              // copying qpid)
@@ -40,2 +40,3 @@
   this.state = this.isClient ? 'frameHeader' : 'protocolHeader';
+  this.maxFrameBuffer = MAX_FRAME_BUFFER_DEFAULT;
 
@@ -61,3 +62,3 @@ if (version != '0-9-1') this.throwError("Unsupported protocol version");
       fh.used = 0; // for reuse
-      if (frameSize > maxFrameBuffer) {
+      if (frameSize > self.maxFrameBuffer) {
         self.throwError("Oversized frame " + frameSize);
@@ -143,2 +144,12 @@ }
 
+/**
+ * Set the maximum frame buffer size in bytes. The connection needs to change this
+ * if the server responds with a connection tune event where the maxFrameBuffer
+ * was changed in the server config.
+ *
+ * @param maxFrameBuffer the maximum frame buffer size in bytes
+ */
+AMQPParser.prototype.setMaxFrameBuffer = function(maxFrameBuffer) {
+  this.maxFrameBuffer = maxFrameBuffer;
+};
 
@@ -145,0 +156,0 @@ // parse Network Byte Order integers. size can be 1,2,4,8

lib/queue.js

 'use strict';
 var util = require('util');
 var fs = require('fs');
+var StringDecoder = require('string_decoder').StringDecoder;
 var _ = require('lodash');
@@ -30,3 +31,3 @@ var Channel = require('./channel');
   this.options = { autoDelete: true, closeChannelOnUnsubscribe: false };
-  _.extend(this.options, options || {});
+  _.assignIn(this.options, options || {});
 
@@ -121,2 +122,3 @@ this._openCallback = callback;
     var contentType = m.contentType;
+    var decoder = new StringDecoder('utf8');
 
@@ -143,3 +145,3 @@ if (contentType == null && m.headers && m.headers.properties) {
       if (isJSON) {
-        buffer += d.toString();
+        buffer += decoder.write(d);
       } else {
@@ -155,2 +157,3 @@ d.copy(buffer, buffer.used);
       if (isJSON) {
+        decoder.end();
         try {
@@ -427,4 +430,4 @@ json = JSON.parse(buffer);
       // Run test-purge to make sure you got this right
-      _.each(this._bindings, function(exchange, exchangeName){
-        _.each(exchange, function(count, routingKey){
+      _.forEach(this._bindings, function(exchange, exchangeName){
+        _.forEach(exchange, function(count, routingKey){
           self.bind(exchangeName, routingKey);
@@ -519,3 +522,3 @@ });
 Queue.prototype._onContentHeader = function (channel, classInfo, weight, properties, size) {
-  _.extend(this.currentMessage, properties);
+  _.assignIn(this.currentMessage, properties);
   this.currentMessage.read = 0;
@@ -522,0 +525,0 @@ this.currentMessage.size = size;

package.json

 { "name" : "amqp"
 , "description" : "AMQP driver for node"
 , "keywords" : [ "amqp" ]
-, "version" : "0.2.5"
+, "version" : "0.2.6"
 , "author" : { "name" : "Ryan Dahl" }
@@ -38,3 +38,3 @@ , "contributors" :
   {
-    "lodash": "~1.3.1"
+    "lodash": "^4.0.0"
   }
@@ -41,0 +41,0 @@ , "devDependencies" :

README.md

@@ -51,2 +51,7 @@ [![build status](https://secure.travis-ci.org/postwait/node-amqp.png)](http://travis-ci.org/postwait/node-amqp)
 
+// add this for better debuging
+connection.on('error', function(e) {
+  console.log("Error from amqp: ", e);
+});
+
 // Wait for connection to become established.
@@ -390,3 +395,3 @@ connection.on('ready', function () {
 
-Ths method will emit `'queueUnbindOk'` when complete.
+This method will emit `'queueUnbindOk'` when complete.
 
@@ -475,2 +480,3 @@
     will be called.
+- `arguments`: a map of additional arguments to pass in when creating an exchange.
 
@@ -505,3 +511,3 @@ An exchange will emit the `'open'` event when it is finally declared.
 - `priority`: The message priority, 0 to 9.
-- `correlationId`: default null. Application correlation identifier
+- `correlationId`: string, default null. Application correlation identifier
 - `replyTo`: Usually used to name a reply queue for a request message.
@@ -508,0 +514,0 @@ - `expiration`: default null. Message expiration specification

test/harness.js

@@ -33,3 +33,3 @@ require('longjohn');
 global.amqp = require('../amqp');
-global.options = _.extend(global.options || {}, argv);
+global.options = _.assignIn(global.options || {}, argv);
 
@@ -36,0 +36,0 @@

test/test-json.js

@@ -14,9 +14,9 @@ require('./harness').run();
         origMessage2 = {foo:'bar', hello: 'world'},
-        origMessage3 = {coffee:'caf\u00E9', tea: 'th\u00E9'};
+        origMessage3 = {coffee:'caf\u00E9', tea: 'th\u00E9', hearts: (new Array(50000)).join('❤')};
 
     q.bind(exchange, "*");
-  
+
     q.subscribe(function (json, headers, deliveryInfo) {
       recvCount++;
-  
+
       assert.equal("node-json-fanout", deliveryInfo.exchange);
@@ -30,11 +30,11 @@ assert.equal("node-json-queue", deliveryInfo.queue);
           break;
-  
+
         case 'message.json2':
           assert.deepEqual(origMessage2, json);
           break;
-  
+
         case 'message.json3':
           assert.deepEqual(origMessage3, json);
           break;
-  
+
         default:
@@ -49,3 +49,3 @@ throw new Error('unexpected routing key: ' + deliveryInfo.routingKey);
       exchange.publish('message.json3', origMessage3, {contentType: 'application/json'});
-  
+
       setTimeout(function () {
@@ -52,0 +52,0 @@ // wait one second to receive the message, then quit

test/test-parser.js

@@ -7,3 +7,3 @@ // Make sure we get the correct results when splitting frames across
 var EventEmitter = require('events').EventEmitter;
-_.extend(Connection.prototype, EventEmitter.prototype);
+_.assignIn(Connection.prototype, EventEmitter.prototype);
 
@@ -10,0 +10,0 @@ var errorThrown = false;

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

501300

increased by0.2%

Lines of code

6137

increased by0.28%

Number of lines in readme file

557

increased by1.09%

Dependency changes

• + Addedlodash@4.17.21(transitive)

• - Removedlodash@1.3.1(transitive)

• Updatedlodash@^4.0.0