Initial connection.blocked/unblocked support

Add encode/decode support for connection.blocked, connection.unblocked
RabbitMQ AMQP extension

3 files changed, 39 insertions, 0 deletions

diff --git a/codegen b/codegen
-Subproject 9fc9146d057679bc63f152d6c78c3e4adcce3a1
+Subproject 0a95a69f86cdc2b538f4c58241840563fe4ccf4
diff --git a/librabbitmq/amqp_framing.c b/librabbitmq/amqp_framing.c
index e065e48..b2f0a83 100644
--- a/librabbitmq/amqp_framing.c
+++ b/librabbitmq/amqp_framing.c
@@ -105,6 +105,8 @@ char const *amqp_method_name(amqp_method_number_t methodNumber) {
     case AMQP_CONNECTION_OPEN_OK_METHOD: return "AMQP_CONNECTION_OPEN_OK_METHOD";
     case AMQP_CONNECTION_CLOSE_METHOD: return "AMQP_CONNECTION_CLOSE_METHOD";
     case AMQP_CONNECTION_CLOSE_OK_METHOD: return "AMQP_CONNECTION_CLOSE_OK_METHOD";
+    case AMQP_CONNECTION_BLOCKED_METHOD: return "AMQP_CONNECTION_BLOCKED_METHOD";
+    case AMQP_CONNECTION_UNBLOCKED_METHOD: return "AMQP_CONNECTION_UNBLOCKED_METHOD";
     case AMQP_CHANNEL_OPEN_METHOD: return "AMQP_CHANNEL_OPEN_METHOD";
     case AMQP_CHANNEL_OPEN_OK_METHOD: return "AMQP_CHANNEL_OPEN_OK_METHOD";
     case AMQP_CHANNEL_FLOW_METHOD: return "AMQP_CHANNEL_FLOW_METHOD";
@@ -326,6 +328,23 @@ int amqp_decode_method(amqp_method_number_t methodNumber,
       *decoded = m;
       return 0;
     }
+    case AMQP_CONNECTION_BLOCKED_METHOD: {
+      amqp_connection_blocked_t *m = (amqp_connection_blocked_t *) amqp_pool_alloc(pool, sizeof(amqp_connection_blocked_t));
+      if (m == NULL) { return AMQP_STATUS_NO_MEMORY; }
+      {
+        uint8_t len;
+        if (!amqp_decode_8(encoded, &offset, &len)
+            || !amqp_decode_bytes(encoded, &offset, &m->reason, len))
+          return AMQP_STATUS_BAD_AMQP_DATA;
+      }
+      *decoded = m;
+      return 0;
+    }
+    case AMQP_CONNECTION_UNBLOCKED_METHOD: {
+      amqp_connection_unblocked_t *m = NULL; /* no fields */
+      *decoded = m;
+      return 0;
+    }
     case AMQP_CHANNEL_OPEN_METHOD: {
       amqp_channel_open_t *m = (amqp_channel_open_t *) amqp_pool_alloc(pool, sizeof(amqp_channel_open_t));
       if (m == NULL) { return AMQP_STATUS_NO_MEMORY; }
@@ -1267,6 +1286,16 @@ int amqp_encode_method(amqp_method_number_t methodNumber,
     case AMQP_CONNECTION_CLOSE_OK_METHOD: {
       return offset;
     }
+    case AMQP_CONNECTION_BLOCKED_METHOD: {
+      amqp_connection_blocked_t *m = (amqp_connection_blocked_t *) decoded;
+      if (!amqp_encode_8(encoded, &offset, m->reason.len)
+          || !amqp_encode_bytes(encoded, &offset, m->reason))
+        return AMQP_STATUS_BAD_AMQP_DATA;
+      return offset;
+    }
+    case AMQP_CONNECTION_UNBLOCKED_METHOD: {
+      return offset;
+    }
     case AMQP_CHANNEL_OPEN_METHOD: {
       amqp_channel_open_t *m = (amqp_channel_open_t *) decoded;
       if (!amqp_encode_8(encoded, &offset, m->out_of_band.len)
diff --git a/librabbitmq/amqp_framing.h b/librabbitmq/amqp_framing.h
index f6dafe4..0c04cb8 100644
--- a/librabbitmq/amqp_framing.h
+++ b/librabbitmq/amqp_framing.h
@@ -183,6 +183,16 @@ typedef struct amqp_connection_close_ok_t_ {
   char dummy; /* Dummy field to avoid empty struct */
 } amqp_connection_close_ok_t;
 
+#define AMQP_CONNECTION_BLOCKED_METHOD ((amqp_method_number_t) 0x000A003C) /* 10, 60; 655420 */
+typedef struct amqp_connection_blocked_t_ {
+  amqp_bytes_t reason;
+} amqp_connection_blocked_t;
+
+#define AMQP_CONNECTION_UNBLOCKED_METHOD ((amqp_method_number_t) 0x000A003D) /* 10, 61; 655421 */
+typedef struct amqp_connection_unblocked_t_ {
+  char dummy; /* Dummy field to avoid empty struct */
+} amqp_connection_unblocked_t;
+
 #define AMQP_CHANNEL_OPEN_METHOD ((amqp_method_number_t) 0x0014000A) /* 20, 10; 1310730 */
 typedef struct amqp_channel_open_t_ {
   amqp_bytes_t out_of_band;