Code snippets for AMQP codec post

base_types.js

var Octet = rangeInt('octet', 0, 255);

var ShortStr = label('shortstr',
                     transform(function(s) {
                       return s.substr(0, 255);
                     }, arb.Str));

var LongStr = label('longstr',
                    transform(
                      function(bytes) { return new Buffer(bytes); },
                      repeat(Octet)));

// ...

decode_start.js

module.exports.ConnectionStart = 655370;
function decodeConnectionStart(buffer) {
  var fields = {}, offset = 0, val, len;
  val = buffer[offset]; offset++;
  fields['versionMajor'] = val;
  val = buffer[offset]; offset++;
  fields['versionMinor'] = val;
  len = buffer.readUInt32BE(offset); offset += 4;
  val = decodeFields(buffer.slice(offset, offset + len));
  offset += len;
  fields['serverProperties'] = val;
  len = buffer.readUInt32BE(offset); offset += 4;
  val = buffer.slice(offset, offset + len);
  offset += len;
  fields['mechanisms'] = val;
  len = buffer.readUInt32BE(offset); offset += 4;
  val = buffer.slice(offset, offset + len);
  offset += len;
  fields['locales'] = val;
  return fields;
}

frame_eg

<Buffer 01                         // frame type 1 = method
        00 00                      // channel 0
        00 00 00 1c                // payload size = 28
        00 0a 00 0a                // method id = ConnectionStart
        00                         // major version (int)
        09                         // minor version (int)
        00 00 00 00                // server properties (empty table)
        00 00 00 05 50 4c 41 49 4e // mechanisms ("PLAIN")
        00 00 00 05 65 6e 5f 55 53 // locales ("en_US")
  ce>                              // frame delimiter
  

frame_parse.js

var frameHeaderPattern = Bits.matcher('type:8', 'channel:16',
                                      'size:32', 'rest/binary');

function parseFrame(bin, max) {
  var fh = frameHeaderPattern(bin);
  if (fh) {
    var size = fh.size, rest = fh.rest;
    if (size > max) {
      throw new Error('Frame size exceeds frame max');
    }
    else if (rest.length > size) {
      if (rest[size] !== FRAME_END)
        throw new Error('Invalid frame');

      return {
        type: fh.type,
        channel: fh.channel,
        size: size,
        payload: rest.slice(0, size),
        rest: rest.slice(size + 1)
      };
    }
  }
  return false;
}

product_types.js

// These are the domains used in method arguments
ARG_TYPES = {
  'octet': Octet,
  'shortstr': ShortStr,
  'longstr': LongStr,
  'short': UShort,
  'long': ULong,
  'longlong': ULongLong,
  'bit': Bit,
  'table': FieldTable,
  'timestamp': ArgTimestamp
};

function argtype(thing) {
  return ARG_TYPES[thing.type];
}

function zipObject(vals, names) {
  var obj = {};
  vals.forEach(function(v, i) { obj[names[i]] = v; });
  return obj;
}

function name(arg) { return arg.name; }

var defs = require('../lib/defs');

function method(info) {
  var domain = sequence.apply(null, info.args.map(argtype));
  var names = info.args.map(name);
  return label(info.name, transform(function(fieldVals) {
    return {id: info.id,
            fields: zipObject(fieldVals, names)};
  }, domain));
}

sum_types.js

var FieldArray = label('field-array', recursive(function() {
  return arb.Array(
    LongStr, ShortStr, Octet,
    UShort, ULong, ULongLong,
    Short, Long, LongLong,
    Bit, Float, Double, FieldTable, FieldArray)
}));

var FieldTable = label('field-table', recursive(function() {
  return sized(function() { return 5; },
               arb.Object(
                 LongStr, ShortStr, Octet,
                 UShort, ULong, ULongLong,
                 Short, Long, LongLong,
                 Bit, Float, Double, FieldArray, FieldTable))
}));

test_trace.js

var Trace = label('frame trace',
                  repeat(choice.apply(choice, amqp.methods)));

suite("Parsing", function() {

  function testPartitioning(partition) {
    return forAll(Trace).satisfy(function(t) {
      var bufs = [];
      var input = inputs();
      var frames = new Frames(input);
      var i = 0;
      frames.accept = function(f) {
        assert.deepEqual(f, t[i]);
        i++;
      };
      
      t.forEach(function(f) {
        f.channel = 0;
        bufs.push(defs.encodeMethod(f.id, 0, f.fields));
      });

      partition(bufs).forEach(input.write.bind(input));
      frames.acceptLoop();
      return i === t.length;
    }).asTest({times: 20})
  };

  test("Parse trace of methods",
       testPartitioning(function(bufs) { return bufs; }));

  test("Parse concat'd methods",
       testPartitioning(function(bufs) {
         return [Buffer.concat(bufs)];
       }));

  test("Parse partitioned methods",
       testPartitioning(function(bufs) {
         var full = Buffer.concat(bufs);
         var onethird = Math.floor(full.length / 3);
         var twothirds = 2 * onethird;
         return [
           full.slice(0, onethird),
           full.slice(onethird, twothirds),
           full.slice(twothirds)
         ];
       }));
});