initial changes to implement version 0.9.1 of the AMQP specification

Author: shimaore

README.md

@@ -3,7 +3,7 @@
 IMPORTANT: This module only works with node v0.1.90 and later.
 
 This is a client for RabbitMQ (and maybe other servers?). It partially
-implements the 0.8 version of the AMQP protocol.
+implements the 0.9.1 version of the AMQP protocol.
 
 
 ## Synopsis
@@ -35,7 +35,7 @@ An example of connecting to a server and listening on a queue.
 ## Connection
 
 `new amqp.Connection()` Instantiates a new connection. Use
-`connection.connect()` to connect to a server. 
+`connection.connect()` to connect to a server.
 
 `amqp.createConnection()` returns an instance of `amqp.Connection`, which is
 a subclass of `net.Stream`. All the event and methods which work on
@@ -145,11 +145,21 @@ bound it will not receive any messages.
 
 If the `exchange` argument is left out `'amq.topic'` will be used.
 
+### queue.bind_headers([exchange,] routing)
+
+This method binds a queue to an exchange.  Until a queue is
+bound it will not receive any messages.
+
+This method is to be used on an "headers"-type exchange. The routing
+argument must contain the routing keys and the `x-match` value (`all` or `any`).
+
+If the `exchange` argument is left out `'amq.headers'` will be used.
+
 
 ### queue.destroy(options)
 
 Delete the queue. Without options, the queue will be deleted even if it has
-pending messages or attached consumers. If +options.ifUnused+ is true, then 
+pending messages or attached consumers. If +options.ifUnused+ is true, then
 the queue will only be deleted if there are no consumers. If
 +options.ifEmpty+ is true, the queue will only be deleted if it has no
 messages.

amqp-0-9-1.xml

@@ -1,7 +1,7 @@
 var events = require('events'),
     sys = require('sys'),
     net = require('net'),
-    protocol = require('./amqp-definitions-0-8'),
+    protocol,
     Buffer = require('buffer').Buffer,
     Promise = require('./promise').Promise;
 
@@ -80,32 +80,6 @@ var methods = {};
 // classes keyed on their index
 var classes = {};
 
-(function () { // anon scope for init
-  //debug("initializing amqp methods...");
-  for (var i = 0; i < protocol.classes.length; i++) {
-    var classInfo = protocol.classes[i];
-    classes[classInfo.index] = classInfo;
-    for (var j = 0; j < classInfo.methods.length; j++) {
-      var methodInfo = classInfo.methods[j];
-
-      var name = classInfo.name
-               + methodInfo.name[0].toUpperCase()
-               + methodInfo.name.slice(1);
-      //debug(name);
-
-      var method = { name: name
-                   , fields: methodInfo.fields
-                   , methodIndex: methodInfo.index
-                   , classIndex: classInfo.index
-                   };
-
-      if (!methodTable[classInfo.index]) methodTable[classInfo.index] = {};
-      methodTable[classInfo.index][methodInfo.index] = method;
-      methods[name] = method;
-    }
-  }
-})(); // end anon scope
-
 
 
 // parser
@@ -117,7 +91,7 @@ var maxFrameBuffer = 131072; // same as rabbitmq
 // An interruptible AMQP parser.
 //
 // type is either 'server' or 'client'
-// version is '0-8' or '0-9-1'. Currently only supporting '0-8'.
+// version is '0-9-1'.
 //
 // Instances of this class have several callbacks
 // - onMethod(channel, method, args);
@@ -132,7 +106,35 @@ function AMQPParser (version, type) {
   this.isClient = (type == 'client');
   this.state = this.isClient ? 'frameHeader' : 'protocolHeader';
 
-  if (version != '0-8') throw new Error("Unsupported protocol version");
+  if (version != '0-9-1') throw new Error("Unsupported protocol version");
+
+  protocol = require('./amqp-definitions-'+version);
+
+(function () { // anon scope for init
+  //debug("initializing amqp methods...");
+  for (var i = 0; i < protocol.classes.length; i++) {
+    var classInfo = protocol.classes[i];
+    classes[classInfo.index] = classInfo;
+    for (var j = 0; j < classInfo.methods.length; j++) {
+      var methodInfo = classInfo.methods[j];
+
+      var name = classInfo.name
+               + methodInfo.name[0].toUpperCase()
+               + methodInfo.name.slice(1);
+      //debug(name);
+
+      var method = { name: name
+                   , fields: methodInfo.fields
+                   , methodIndex: methodInfo.index
+                   , classIndex: classInfo.index
+                   };
+
+      if (!methodTable[classInfo.index]) methodTable[classInfo.index] = {};
+      methodTable[classInfo.index][methodInfo.index] = method;
+      methods[name] = method;
+    }
+  }
+})(); // end anon scope
 
   this.frameHeader = new Buffer(7);
   this.frameHeader.used = 0;
@@ -619,7 +621,7 @@ function serializeFields (buffer, fields, args, strict) {
           buffer[buffer.used++] = bitField;
           bitField = 0;
           bitIndex = 0;
-        } 
+        }
         break;
 
       case 'octet':
@@ -692,7 +694,7 @@ function Connection (options) {
     self.queues = {};
     self.exchanges = {};
 
-    parser = new AMQPParser('0-8', 'client');
+    parser = new AMQPParser('0-9-1', 'client');
 
     parser.onMethod = function (channel, method, args) {
       self._onMethod(channel, method, args);
@@ -723,7 +725,7 @@ function Connection (options) {
     //debug("connected...");
     // Time to start the AMQP 7-way connection initialization handshake!
     // 1. The client sends the server a version string
-    self.write("AMQP" + String.fromCharCode(1,1,8,0));
+    self.write("AMQP" + String.fromCharCode(0,0,9,1));
     state = 'handshake';
   });
 
@@ -791,8 +793,8 @@ Connection.prototype._onMethod = function (channel, method, args) {
     // 2. The server responds, after the version string, with the
     // 'connectionStart' method (contains various useless information)
     case methods.connectionStart:
-      // We check that they're serving us AMQP 0-8
-      if (args.versionMajor != 8 && args.versionMinor != 0) {
+      // We check that they're serving us AMQP 0-9
+      if (args.versionMajor != 0 && args.versionMinor != 9) {
         this.end();
         this.emit('error', new Error("Bad server version"));
         return;
@@ -825,8 +827,10 @@ Connection.prototype._onMethod = function (channel, method, args) {
       // 6. Then we have to send a connectionOpen request
       this._sendMethod(0, methods.connectionOpen,
           { virtualHost: this.options.vhost
-          , capabilities: ''
-          , insist: true
+          // , capabilities: ''
+          // , insist: true
+          , reserved1: ''
+          , reserved2: true
           });
       break;
 
@@ -1140,7 +1144,7 @@ function Channel (connection, channel) {
   this.connection = connection;
   this._tasks = [];
 
-  this.connection._sendMethod(channel, methods.channelOpen, {outOfBand: ""});
+  this.connection._sendMethod(channel, methods.channelOpen, {reserved1: ""});
 }
 sys.inherits(Channel, events.EventEmitter);
 
@@ -1212,13 +1216,13 @@ Queue.prototype.subscribeRaw = function (/* options, messageListener */) {
 
   return this._taskPush(methods.basicConsumeOk, function () {
     self.connection._sendMethod(self.channel, methods.basicConsume,
-        { ticket: 0
+        { reserved1: 0
         , queue: self.name
         , consumerTag: "."
         , noLocal: options.noLocal ? true : false
         , noAck: options.noAck ? true : false
         , exclusive: options.exclusive ? true : false
-        , nowait: false
+        , noWait: false
         , "arguments": {}
         });
   });
@@ -1306,31 +1310,62 @@ Queue.prototype.bind = function (/* [exchange,] routingKey */) {
   // The first argument, exchange is optional.
   // If not supplied the connection will use the default 'amq.topic'
   // exchange.
- 
+
   var exchange, routingKey;
 
   if (arguments.length == 2) {
     exchange = arguments[0];
     routingKey = arguments[1];
   } else {
-    exchange = 'amq.topic';   
+    exchange = 'amq.topic';
     routingKey = arguments[0];
   }
 
 
   return this._taskPush(methods.queueBindOk, function () {
     var exchangeName = exchange instanceof Exchange ? exchange.name : exchange;
     self.connection._sendMethod(self.channel, methods.queueBind,
-        { ticket: 0
+        { reserved1: 0
         , queue: self.name
         , exchange: exchangeName
         , routingKey: routingKey
-        , nowait: false
+        , noWait: false
         , "arguments": {}
         });
   });
 };
 
+Queue.prototype.bind_headers = function (/* [exchange,] matchingPairs */) {
+  var self = this;
+
+  // The first argument, exchange is optional.
+  // If not supplied the connection will use the default 'amq.headers'
+  // exchange.
+
+  var exchange, matchingPairs;
+
+  if (arguments.length == 2) {
+    exchange = arguments[0];
+    matchingPairs = arguments[1];
+  } else {
+    exchange = 'amq.headers';
+    matchingPairs = arguments[0];
+  }
+
+
+  return this._taskPush(methods.queueBindOk, function () {
+    var exchangeName = exchange instanceof Exchange ? exchange.name : exchange;
+    self.connection._sendMethod(self.channel, methods.queueBind,
+        { reserved1: 0
+        , queue: self.name
+        , exchange: exchangeName
+        , routingKey: ''
+        , noWait: false
+        , "arguments": matchingPairs
+        });
+  });
+};
+
 
 Queue.prototype.destroy = function (options) {
   var self = this;
@@ -1354,13 +1389,13 @@ Queue.prototype._onMethod = function (channel, method, args) {
   switch (method) {
     case methods.channelOpenOk:
       this.connection._sendMethod(channel, methods.queueDeclare,
-          { ticket: 0
+          { reserved1: 0
           , queue: this.name
           , passive: this.options.passive ? true : false
           , durable: this.options.durable ? true : false
           , exclusive: this.options.exclusive ? true : false
           , autoDelete: this.options.autoDelete ? true : false
-          , nowait: false
+          , noWait: false
           , "arguments": {}
           });
       this.state = "declare queue";
@@ -1516,7 +1551,7 @@ Exchange.prototype.publish = function (routingKey, data, options) {
   var self = this;
   return this._taskPush(null, function () {
     self.connection._sendMethod(self.channel, methods.basicPublish,
-        { ticket: 0
+        { reserved1: 0
         , exchange:   self.name
         , routingKey: routingKey
         , mandatory:  options.mandatory ? true : false

amqp-0.8.xml

@@ -1,6 +1,6 @@
 { "name" : "amqp"
-, "version" : "0.0.2"
+, "version" : "0.0.3"
 , "description" : "amqp bindings for RabbitMQ"
-, "author": "Ryan Dahl"
+, "author": "Ryan Dahl, Stephane Alnet"
 , "main": "./amqp"
 }