add support for data and sequence sections

Author: grs

lib/message.js

@@ -69,6 +69,15 @@ function define_map_section(def) {
     define_section(descriptor, unwrap, wrap);
 }
 
+function Section(typecode, content) {
+    this.typecode = typecode;
+    this.content = content;
+}
+
+Section.prototype.described = function () {
+    return types.described(types.wrap_ulong(this.typecode), types.wrap(this.content));
+};
+
 define_composite_section({name:'header',
                           code:0x70,
                           fields:[
@@ -101,12 +110,31 @@ define_composite_section({name:'properties',
                          });
 define_map_section({name:'application_properties', code:0x74});
 
-define_section({numeric:0x77, symbolic:'amqp:value:*'},
-               function(msg, section) { msg.body = types.unwrap(section); },
-               function(sections, msg) { sections.push(types.described(types.wrap_ulong(0x77), types.wrap(msg.body))); });
+define_section({numeric:0x75, symbolic:'amqp:data:binary'}, function (msg, section) { msg.body = new Section(0x75, types.unwrap(section)); });
+define_section({numeric:0x76, symbolic:'amqp:amqp-sequence:list'}, function (msg, section) { msg.body = new Section(0x76, types.unwrap(section)); });
+define_section({numeric:0x77, symbolic:'amqp:value:*'}, function (msg, section) { msg.body = types.unwrap(section); });
 
 define_map_section({name:'footer', code:0x78});
 
+
+function wrap_body (sections, msg) {
+    if (msg.body && msg.body.constructor === Section) {
+        sections.push(msg.body.described());
+    } else {
+        sections.push(types.described(types.wrap_ulong(0x77), types.wrap(msg.body)));
+    }
+}
+
+wrappers.push(wrap_body);
+
+message.data_section = function (data) {
+    return new Section(0x75, data);
+};
+
+message.sequence_section = function (list) {
+    return new Section(0x76, list);
+};
+
 function reverse_lookup(map) {
     var reversed = {};
     for (var key in map) {

lib/types.js

@@ -384,6 +384,8 @@ types.wrap = function(o) {
         return types.wrap_timestamp(o.getTime());
     } else if (o instanceof Typed) {
         return o;
+    } else if (o instanceof Buffer) {
+        return types.wrap_binary(o);
     } else if (t === 'undefined' || o === null) {
         return new types.Null();
     } else if (Array.isArray(o)) {

test/messages.js

@@ -18,6 +18,7 @@
 var assert = require('assert');
 var rhea = require('../lib/container.js');
 var amqp_types = require('../lib/types.js');
+var amqp_message = require('../lib/message.js');
 
 describe('message content', function() {
     var container, sender, listener;
@@ -52,6 +53,16 @@ describe('message content', function() {
     it('sends and receives binary body', transfer_test({body:amqp_types.wrap_binary(new Buffer('hello world!'))}, function(message) {
         assert.equal(message.body.toString(), 'hello world!');
     }));
+    it('sends and receives body as data section', transfer_test({body:amqp_message.data_section(new Buffer('hello world!'))}, function(message) {
+        assert.equal(message.body.typecode, 0x75);
+        assert.equal(message.body.content.toString(), 'hello world!');
+    }));
+    it('sends and receives body as sequence section', transfer_test({body:amqp_message.sequence_section(['hello', 1, 'world!'])}, function(message) {
+        assert.equal(message.body.typecode, 0x76);
+        assert.equal(message.body.content[0], 'hello');
+        assert.equal(message.body.content[1], 1);
+        assert.equal(message.body.content[2], 'world!');
+    }));
     it('sends and receives subject', transfer_test({properties:{subject:'my-subject'}}, function(message) {
         assert.equal(message.properties.subject, 'my-subject');
     }));