Implement Qpid/RabbitMQ/0-9-1 field table types.

3 files changed, 390 insertions, 73 deletions

diff --git a/librabbitmq/amqp.h b/librabbitmq/amqp.h
index b25d84c..643fc39 100644
--- a/librabbitmq/amqp.h
+++ b/librabbitmq/amqp.h
@@ -31,24 +31,135 @@ typedef struct amqp_table_t_ {
 
 #define AMQP_EMPTY_TABLE ((amqp_table_t) { .num_entries = 0, .entries = NULL })
 
-typedef struct amqp_table_entry_t_ {
-  amqp_bytes_t key;
+typedef struct amqp_array_t_ {
+  int num_entries;
+  struct amqp_field_value_t_ *entries;
+} amqp_array_t;
+
+#define AMQP_EMPTY_ARRAY ((amqp_array_t) { .num_entries = 0, .entries = NULL })
+
+/*
+  0-9   0-9-1   Qpid/Rabbit  Type               Remarks
+---------------------------------------------------------------------------
+        t       t            Boolean
+        b       b            Signed 8-bit
+        B                    Unsigned 8-bit
+        U       s            Signed 16-bit	(A1)
+        u                    Unsigned 16-bit
+  I     I       I	     Signed 32-bit
+        i		     Unsigned 32-bit
+        L       l	     Signed 64-bit	(B)
+        l		     Unsigned 64-bit
+        f       f	     32-bit float
+        d       d	     64-bit float
+  D     D       D	     Decimal
+        s		     Short string	(A2)
+  S     S       S	     Long string
+        A		     Nested Array
+  T     T       T	     Timestamp (u64)
+  F     F       F	     Nested Table
+  V     V       V	     Void
+                x	     Byte array
+
+Remarks:
+
+ A1, A2: Notice how the types **CONFLICT** here. In Qpid and Rabbit,
+         's' means a signed 16-bit integer; in 0-9-1, it means a
+	 short string.
+
+ B: Notice how the signednesses **CONFLICT** here. In Qpid and Rabbit,
+    'l' means a signed 64-bit integer; in 0-9-1, it means an unsigned
+    64-bit integer.
+
+I'm going with the Qpid/Rabbit types, where there's a conflict, and
+the 0-9-1 types otherwise. 0-8 is a subset of 0-9, which is a subset
+of the other two, so this will work for both 0-8 and 0-9-1 branches of
+the code.
+*/
+
+typedef struct amqp_field_value_t_ {
   char kind;
   union {
-    amqp_bytes_t bytes;
+    amqp_boolean_t boolean;
+    int8_t i8;
+    uint8_t u8;
+    int16_t i16;
+    uint16_t u16;
     int32_t i32;
-    amqp_decimal_t decimal;
+    uint32_t u32;
+    int64_t i64;
     uint64_t u64;
+    float f32;
+    double f64;
+    amqp_decimal_t decimal;
+    amqp_bytes_t bytes;
     amqp_table_t table;
+    amqp_array_t array;
   } value;
+} amqp_field_value_t;
+
+typedef struct amqp_table_entry_t_ {
+  amqp_bytes_t key;
+  amqp_field_value_t value;
 } amqp_table_entry_t;
 
-#define _AMQP_TE_INIT(ke,ki,v) { .key = (ke), .kind = (ki), .value = { v } }
-#define AMQP_TABLE_ENTRY_S(k,v) _AMQP_TE_INIT(amqp_cstring_bytes(k), 'S', .bytes = (v))
-#define AMQP_TABLE_ENTRY_I(k,v) _AMQP_TE_INIT(amqp_cstring_bytes(k), 'I', .i32 = (v))
-#define AMQP_TABLE_ENTRY_D(k,v) _AMQP_TE_INIT(amqp_cstring_bytes(k), 'D', .decimal = (v))
-#define AMQP_TABLE_ENTRY_T(k,v) _AMQP_TE_INIT(amqp_cstring_bytes(k), 'T', .u64 = (v))
-#define AMQP_TABLE_ENTRY_F(k,v) _AMQP_TE_INIT(amqp_cstring_bytes(k), 'F', .table = (v))
+typedef enum {
+  AMQP_FIELD_KIND_BOOLEAN = 't',
+  AMQP_FIELD_KIND_I8 = 'b',
+  AMQP_FIELD_KIND_U8 = 'B',
+  AMQP_FIELD_KIND_I16 = 's',
+  AMQP_FIELD_KIND_U16 = 'u',
+  AMQP_FIELD_KIND_I32 = 'I',
+  AMQP_FIELD_KIND_U32 = 'i',
+  AMQP_FIELD_KIND_I64 = 'l',
+  AMQP_FIELD_KIND_F32 = 'f',
+  AMQP_FIELD_KIND_F64 = 'd',
+  AMQP_FIELD_KIND_DECIMAL = 'D',
+  AMQP_FIELD_KIND_UTF8 = 'S',
+  AMQP_FIELD_KIND_ARRAY = 'A',
+  AMQP_FIELD_KIND_TIMESTAMP = 'T',
+  AMQP_FIELD_KIND_TABLE = 'F',
+  AMQP_FIELD_KIND_VOID = 'V',
+  AMQP_FIELD_KIND_BYTES = 'x',
+} amqp_field_value_kind_t;
+
+#define _AMQP_TEINIT(ke,ki,v) {.key = (ke), .value = {.kind = AMQP_FIELD_KIND_##ki, .value = {v}}}
+#define AMQP_TABLE_ENTRY_BOOLEAN(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), BOOLEAN, .boolean = (v))
+#define AMQP_TABLE_ENTRY_I8(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), I8, .i8 = (v))
+#define AMQP_TABLE_ENTRY_U8(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), U8, .u8 = (v))
+#define AMQP_TABLE_ENTRY_I16(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), I16, .i16 = (v))
+#define AMQP_TABLE_ENTRY_U16(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), U16, .u16 = (v))
+#define AMQP_TABLE_ENTRY_I32(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), I32, .i32 = (v))
+#define AMQP_TABLE_ENTRY_U32(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), U32, .u32 = (v))
+#define AMQP_TABLE_ENTRY_I64(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), I64, .i64 = (v))
+#define AMQP_TABLE_ENTRY_F32(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), F32, .f32 = (v))
+#define AMQP_TABLE_ENTRY_F64(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), F64, .f64 = (v))
+#define AMQP_TABLE_ENTRY_DECIMAL(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), DECIMAL, .decimal = (v))
+#define AMQP_TABLE_ENTRY_UTF8(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), UTF8, .bytes = (v))
+#define AMQP_TABLE_ENTRY_ARRAY(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), ARRAY, .array = (v))
+#define AMQP_TABLE_ENTRY_TIMESTAMP(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), TIMESTAMP, .u64 = (v))
+#define AMQP_TABLE_ENTRY_TABLE(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), TABLE, .table = (v))
+#define AMQP_TABLE_ENTRY_VOID(k)   _AMQP_TEINIT(amqp_cstring_bytes(k), VOID, .u8 = 0)
+#define AMQP_TABLE_ENTRY_BYTES(k,v) _AMQP_TEINIT(amqp_cstring_bytes(k), BYTES, .bytes = (v))
+
+#define _AMQP_FVINIT(ki,v) {.kind = AMQP_FIELD_KIND_##ki, .value = {v}}
+#define AMQP_FIELD_VALUE_BOOLEAN(v) _AMQP_FVINIT(BOOLEAN, .boolean = (v))
+#define AMQP_FIELD_VALUE_I8(v) _AMQP_FVINIT(I8, .i8 = (v))
+#define AMQP_FIELD_VALUE_U8(v) _AMQP_FVINIT(U8, .u8 = (v))
+#define AMQP_FIELD_VALUE_I16(v) _AMQP_FVINIT(I16, .i16 = (v))
+#define AMQP_FIELD_VALUE_U16(v) _AMQP_FVINIT(U16, .u16 = (v))
+#define AMQP_FIELD_VALUE_I32(v) _AMQP_FVINIT(I32, .i32 = (v))
+#define AMQP_FIELD_VALUE_U32(v) _AMQP_FVINIT(U32, .u32 = (v))
+#define AMQP_FIELD_VALUE_I64(v) _AMQP_FVINIT(I64, .i64 = (v))
+#define AMQP_FIELD_VALUE_F32(v) _AMQP_FVINIT(F32, .f32 = (v))
+#define AMQP_FIELD_VALUE_F64(v) _AMQP_FVINIT(F64, .f64 = (v))
+#define AMQP_FIELD_VALUE_DECIMAL(v) _AMQP_FVINIT(DECIMAL, .decimal = (v))
+#define AMQP_FIELD_VALUE_UTF8(v) _AMQP_FVINIT(UTF8, .bytes = (v))
+#define AMQP_FIELD_VALUE_ARRAY(v) _AMQP_FVINIT(ARRAY, .array = (v))
+#define AMQP_FIELD_VALUE_TIMESTAMP(v) _AMQP_FVINIT(TIMESTAMP, .u64 = (v))
+#define AMQP_FIELD_VALUE_TABLE(v) _AMQP_FVINIT(TABLE, .table = (v))
+#define AMQP_FIELD_VALUE_VOID(k)   _AMQP_FVINIT(VOID, .u8 = 0)
+#define AMQP_FIELD_VALUE_BYTES(v) _AMQP_FVINIT(BYTES, .bytes = (v))
 
 typedef struct amqp_pool_blocklist_t_ {
   int num_blocks;
diff --git a/librabbitmq/amqp_private.h b/librabbitmq/amqp_private.h
index 98a7a65..03a46fe 100644
--- a/librabbitmq/amqp_private.h
+++ b/librabbitmq/amqp_private.h
@@ -115,13 +115,15 @@ extern int amqp_encode_table(amqp_bytes_t encoded,
     }						\
   })
 
-#define AMQP_CHECK_RESULT(expr)			\
+#define AMQP_CHECK_RESULT_CLEANUP(expr, stmts)	\
   ({						\
     int _result = (expr);			\
-    if (_result < 0) return _result;		\
+    if (_result < 0) { stmts; return _result; }	\
     _result;					\
   })
 
+#define AMQP_CHECK_RESULT(expr) AMQP_CHECK_RESULT_CLEANUP(expr, )
+
 #define AMQP_CHECK_EOF_RESULT(expr)		\
   ({						\
     int _result = (expr);			\
diff --git a/librabbitmq/amqp_table.c b/librabbitmq/amqp_table.c
index 16d85ff..5ffadc4 100644
--- a/librabbitmq/amqp_table.c
+++ b/librabbitmq/amqp_table.c
@@ -9,8 +9,67 @@
 
 #include <assert.h>
 
+#define INITIAL_ARRAY_SIZE 16
 #define INITIAL_TABLE_SIZE 16
 
+static int amqp_decode_field_value(amqp_bytes_t encoded,
+				   amqp_pool_t *pool,
+				   amqp_field_value_t *entry,
+				   int *offsetptr); /* forward */
+
+static int amqp_encode_field_value(amqp_bytes_t encoded,
+				   amqp_field_value_t *entry,
+				   int *offsetptr); /* forward */
+
+/*---------------------------------------------------------------------------*/
+
+static int amqp_decode_array(amqp_bytes_t encoded,
+			     amqp_pool_t *pool,
+			     amqp_array_t *output,
+			     int *offsetptr)
+{
+  int offset = *offsetptr;
+  uint32_t arraysize = D_32(encoded, offset);
+  int num_entries = 0;
+  amqp_field_value_t *entries = malloc(INITIAL_ARRAY_SIZE * sizeof(amqp_field_value_t));
+  int allocated_entries = INITIAL_ARRAY_SIZE;
+  int limit;
+
+  if (entries == NULL) {
+    return -ENOMEM;
+  }
+
+  offset += 4;
+  limit = offset + arraysize;
+
+  while (offset < limit) {
+    if (num_entries >= allocated_entries) {
+      void *newentries;
+      allocated_entries = allocated_entries * 2;
+      newentries = realloc(entries, allocated_entries * sizeof(amqp_field_value_t));
+      if (newentries == NULL) {
+	free(entries);
+	return -ENOMEM;
+      }
+      entries = newentries;
+    }
+
+    AMQP_CHECK_RESULT_CLEANUP(amqp_decode_field_value(encoded,
+						      pool,
+						      &entries[num_entries],
+						      &offset),
+			      free(entries));
+    num_entries++;
+  }
+
+  output->num_entries = num_entries;
+  output->entries = amqp_pool_alloc(pool, num_entries * sizeof(amqp_field_value_t));
+  memcpy(output->entries, entries, num_entries * sizeof(amqp_field_value_t));
+
+  *offsetptr = offset;
+  return 0;
+}
+
 int amqp_decode_table(amqp_bytes_t encoded,
 		      amqp_pool_t *pool,
 		      amqp_table_t *output,
@@ -53,37 +112,11 @@ int amqp_decode_table(amqp_bytes_t encoded,
     entry->key.bytes = D_BYTES(encoded, offset, keylen);
     offset += keylen;
 
-    entry->kind = D_8(encoded, offset);
-    offset++;
-
-    switch (entry->kind) {
-      case 'S':
-	entry->value.bytes.len = D_32(encoded, offset);
-	offset += 4;
-	entry->value.bytes.bytes = D_BYTES(encoded, offset, entry->value.bytes.len);
-	offset += entry->value.bytes.len;
-	break;
-      case 'I':
-	entry->value.i32 = (int32_t) D_32(encoded, offset);
-	offset += 4;
-	break;
-      case 'D':
-	entry->value.decimal.decimals = D_8(encoded, offset);
-	offset++;
-	entry->value.decimal.value = D_32(encoded, offset);
-	offset += 4;
-	break;
-      case 'T':
-	entry->value.u64 = D_64(encoded, offset);
-	offset += 8;
-	break;
-      case 'F':
-	AMQP_CHECK_RESULT(amqp_decode_table(encoded, pool, &(entry->value.table), &offset));
-	break;
-      default:
-	return -EINVAL;
-    }
-
+    AMQP_CHECK_RESULT_CLEANUP(amqp_decode_field_value(encoded,
+						      pool,
+						      &entry->value,
+						      &offset),
+			      free(entries));
     num_entries++;
   }
 
@@ -95,6 +128,116 @@ int amqp_decode_table(amqp_bytes_t encoded,
   return 0;
 }
 
+static int amqp_decode_field_value(amqp_bytes_t encoded,
+				   amqp_pool_t *pool,
+				   amqp_field_value_t *entry,
+				   int *offsetptr)
+{
+  int offset = *offsetptr;
+
+  entry->kind = D_8(encoded, offset);
+  offset++;
+
+  switch (entry->kind) {
+    case AMQP_FIELD_KIND_BOOLEAN:
+      entry->value.boolean = D_8(encoded, offset) ? 1 : 0;
+      offset++;
+      break;
+    case AMQP_FIELD_KIND_I8:
+      entry->value.i8 = (int8_t) D_8(encoded, offset);
+      offset++;
+      break;
+    case AMQP_FIELD_KIND_U8:
+      entry->value.u8 = D_8(encoded, offset);
+      offset++;
+      break;
+    case AMQP_FIELD_KIND_I16:
+      entry->value.i16 = (int16_t) D_16(encoded, offset);
+      offset += 2;
+      break;
+    case AMQP_FIELD_KIND_U16:
+      entry->value.u16 = D_16(encoded, offset);
+      offset += 2;
+      break;
+    case AMQP_FIELD_KIND_I32:
+      entry->value.i32 = (int32_t) D_32(encoded, offset);
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_U32:
+      entry->value.u32 = D_32(encoded, offset);
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_I64:
+      entry->value.i64 = (int64_t) D_64(encoded, offset);
+      offset += 8;
+      break;
+    case AMQP_FIELD_KIND_F32:
+      entry->value.u32 = D_32(encoded, offset);
+      /* and by punning, f32 magically gets the right value...! */
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_F64:
+      entry->value.u64 = D_64(encoded, offset);
+      /* and by punning, f64 magically gets the right value...! */
+      offset += 8;
+      break;
+    case AMQP_FIELD_KIND_DECIMAL:
+      entry->value.decimal.decimals = D_8(encoded, offset);
+      offset++;
+      entry->value.decimal.value = D_32(encoded, offset);
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_UTF8:
+      /* AMQP_FIELD_KIND_UTF8 and AMQP_FIELD_KIND_BYTES have the
+	 same implementation, but different interpretations. */
+      /* fall through */
+    case AMQP_FIELD_KIND_BYTES:
+      entry->value.bytes.len = D_32(encoded, offset);
+      offset += 4;
+      entry->value.bytes.bytes = D_BYTES(encoded, offset, entry->value.bytes.len);
+      offset += entry->value.bytes.len;
+      break;
+    case AMQP_FIELD_KIND_ARRAY:
+      AMQP_CHECK_RESULT(amqp_decode_array(encoded, pool, &(entry->value.array), &offset));
+      break;
+    case AMQP_FIELD_KIND_TIMESTAMP:
+      entry->value.u64 = D_64(encoded, offset);
+      offset += 8;
+      break;
+    case AMQP_FIELD_KIND_TABLE:
+      AMQP_CHECK_RESULT(amqp_decode_table(encoded, pool, &(entry->value.table), &offset));
+      break;
+    case AMQP_FIELD_KIND_VOID:
+      break;
+    default:
+      return -EINVAL;
+  }
+
+  *offsetptr = offset;
+  return 0;
+}
+
+/*---------------------------------------------------------------------------*/
+
+static int amqp_encode_array(amqp_bytes_t encoded,
+			     amqp_array_t *input,
+			     int *offsetptr)
+{
+  int offset = *offsetptr;
+  int arraysize_offset = offset;
+  int i;
+
+  offset += 4; /* skip space for the size of the array to be filled in later */
+
+  for (i = 0; i < input->num_entries; i++) {
+    AMQP_CHECK_RESULT(amqp_encode_field_value(encoded, &(input->entries[i]), &offset));
+  }
+
+  E_32(encoded, arraysize_offset, (offset - *offsetptr - 4));
+  *offsetptr = offset;
+  return 0;
+}
+
 int amqp_encode_table(amqp_bytes_t encoded,
 		      amqp_table_t *input,
 		      int *offsetptr)
@@ -114,36 +257,7 @@ int amqp_encode_table(amqp_bytes_t encoded,
     E_BYTES(encoded, offset, entry->key.len, entry->key.bytes);
     offset += entry->key.len;
 
-    E_8(encoded, offset, entry->kind);
-    offset++;
-
-    switch (entry->kind) {
-      case 'S':
-	E_32(encoded, offset, entry->value.bytes.len);
-	offset += 4;
-	E_BYTES(encoded, offset, entry->value.bytes.len, entry->value.bytes.bytes);
-	offset += entry->value.bytes.len;
-	break;
-      case 'I':
-	E_32(encoded, offset, (uint32_t) entry->value.i32);
-	offset += 4;
-	break;
-      case 'D':
-	E_8(encoded, offset, entry->value.decimal.decimals);
-	offset++;
-	E_32(encoded, offset, entry->value.decimal.value);
-	offset += 4;
-	break;
-      case 'T':
-	E_64(encoded, offset, entry->value.u64);
-	offset += 8;
-	break;
-      case 'F':
-	AMQP_CHECK_RESULT(amqp_encode_table(encoded, &(entry->value.table), &offset));
-	break;
-      default:
-	return -EINVAL;
-    }
+    AMQP_CHECK_RESULT(amqp_encode_field_value(encoded, &(entry->value), &offset));
   }
 
   E_32(encoded, tablesize_offset, (offset - *offsetptr - 4));
@@ -151,6 +265,96 @@ int amqp_encode_table(amqp_bytes_t encoded,
   return 0;
 }
 
+static int amqp_encode_field_value(amqp_bytes_t encoded,
+				   amqp_field_value_t *entry,
+				   int *offsetptr)
+{
+  int offset = *offsetptr;
+
+  E_8(encoded, offset, entry->kind);
+  offset++;
+
+  switch (entry->kind) {
+    case AMQP_FIELD_KIND_BOOLEAN:
+      E_8(encoded, offset, entry->value.boolean ? 1 : 0);
+      offset++;
+      break;
+    case AMQP_FIELD_KIND_I8:
+      E_8(encoded, offset, (uint8_t) entry->value.i8);
+      offset++;
+      break;
+    case AMQP_FIELD_KIND_U8:
+      E_8(encoded, offset, entry->value.u8);
+      offset++;
+      break;
+    case AMQP_FIELD_KIND_I16:
+      E_16(encoded, offset, (uint16_t) entry->value.i16);
+      offset += 2;
+      break;
+    case AMQP_FIELD_KIND_U16:
+      E_16(encoded, offset, entry->value.u16);
+      offset += 2;
+      break;
+    case AMQP_FIELD_KIND_I32:
+      E_32(encoded, offset, (uint32_t) entry->value.i32);
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_U32:
+      E_32(encoded, offset, entry->value.u32);
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_I64:
+      E_64(encoded, offset, (uint64_t) entry->value.i64);
+      offset += 8;
+      break;
+    case AMQP_FIELD_KIND_F32:
+      /* by punning, u32 magically gets the right value...! */
+      E_32(encoded, offset, entry->value.u32);
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_F64:
+      /* by punning, u64 magically gets the right value...! */
+      E_64(encoded, offset, entry->value.u64);
+      offset += 8;
+      break;
+    case AMQP_FIELD_KIND_DECIMAL:
+      E_8(encoded, offset, entry->value.decimal.decimals);
+      offset++;
+      E_32(encoded, offset, entry->value.decimal.value);
+      offset += 4;
+      break;
+    case AMQP_FIELD_KIND_UTF8:
+      /* AMQP_FIELD_KIND_UTF8 and AMQP_FIELD_KIND_BYTES have the
+	 same implementation, but different interpretations. */
+      /* fall through */
+    case AMQP_FIELD_KIND_BYTES:
+      E_32(encoded, offset, entry->value.bytes.len);
+      offset += 4;
+      E_BYTES(encoded, offset, entry->value.bytes.len, entry->value.bytes.bytes);
+      offset += entry->value.bytes.len;
+      break;
+    case AMQP_FIELD_KIND_ARRAY:
+      AMQP_CHECK_RESULT(amqp_encode_array(encoded, &(entry->value.array), &offset));
+      break;
+    case AMQP_FIELD_KIND_TIMESTAMP:
+      E_64(encoded, offset, entry->value.u64);
+      offset += 8;
+      break;
+    case AMQP_FIELD_KIND_TABLE:
+      AMQP_CHECK_RESULT(amqp_encode_table(encoded, &(entry->value.table), &offset));
+      break;
+    case AMQP_FIELD_KIND_VOID:
+      break;
+    default:
+      return -EINVAL;
+  }
+
+  *offsetptr = offset;
+  return 0;
+}
+
+/*---------------------------------------------------------------------------*/
+
 int amqp_table_entry_cmp(void const *entry1, void const *entry2) {
   amqp_table_entry_t const *p1 = (amqp_table_entry_t const *) entry1;
   amqp_table_entry_t const *p2 = (amqp_table_entry_t const *) entry2;