Updated amqp_simple_rpc to take a 0 terminated array of amqp_method_number_t replies

Updated AMQP_SIMPLE_RPC macro for the new format
Added AMQP_MULTIPLE_RESPONSE_RPC macro to take a 0 terminated array of amqp_method_number_t replies
Added amqp_basic_get function
Added amqp_queue_purge function
Added amqp_get_rpc_reply function to expose amqp_rpc_reply when we are not statically linking
Added amqp_data_in_buffer function to check to see if amqp_simple_wait_frames will hit a blocking read

1 files changed, 32 insertions, 3 deletions

diff --git a/librabbitmq/amqp.h b/librabbitmq/amqp.h
index 4c135b1..71d77de 100644
--- a/librabbitmq/amqp.h
+++ b/librabbitmq/amqp.h
@@ -191,18 +191,31 @@ extern int amqp_send_method(amqp_connection_state_t state,
 extern amqp_rpc_reply_t amqp_simple_rpc(amqp_connection_state_t state,
 					amqp_channel_t channel,
 					amqp_method_number_t request_id,
-					amqp_method_number_t expected_reply_id,
+					amqp_method_number_t *expected_reply_ids,
 					void *decoded_request_method);
 
+#define AMQP_EXPAND_METHOD(classname, methodname) (AMQP_ ## classname ## _ ## methodname ## _METHOD)
+
 #define AMQP_SIMPLE_RPC(state, channel, classname, requestname, replyname, structname, ...) \
   ({									\
     structname _simple_rpc_request___ = (structname) { __VA_ARGS__ };	\
+    amqp_method_number_t _replies__[2] = { AMQP_EXPAND_METHOD(classname, replyname), 0}; \
+    amqp_simple_rpc(state, channel,					\
+		    AMQP_EXPAND_METHOD(classname, requestname),	\
+		    (amqp_method_number_t *)&_replies__,	\
+		    &_simple_rpc_request___);				\
+  })
+
+#define AMQP_MULTIPLE_RESPONSE_RPC(state, channel, classname, requestname, replynames, structname, ...) \
+  ({									\
+    structname _simple_rpc_request___ = (structname) { __VA_ARGS__ };	\
     amqp_simple_rpc(state, channel,					\
-		    AMQP_ ## classname ## _ ## requestname ## _METHOD,	\
-		    AMQP_ ## classname ## _ ## replyname ## _METHOD,	\
+		    AMQP_EXPAND_METHOD(classname, requestname),	\
+		    replynames,	\
 		    &_simple_rpc_request___);				\
   })
 
+
 extern amqp_rpc_reply_t amqp_login(amqp_connection_state_t state,
 				   char const *vhost,
 				   int channel_max,
@@ -276,6 +289,22 @@ extern int amqp_basic_ack(amqp_connection_state_t state,
 			  uint64_t delivery_tag,
 			  amqp_boolean_t multiple);
 
+extern amqp_rpc_reply_t amqp_basic_get(amqp_connection_state_t state,
+          amqp_channel_t channel,
+          amqp_bytes_t queue,
+          amqp_boolean_t no_ack);
+
+extern struct amqp_queue_purge_ok_t_ *amqp_queue_purge(amqp_connection_state_t state,
+            amqp_channel_t channel,
+            amqp_bytes_t queue,
+            amqp_boolean_t no_wait);
+
+// Can be used to see if there is data still in the buffer, if so
+// calling amqp_simple_wait_frame will not enter the blocking read
+// possibly amqp_frames_enqueued should be used for this?
+extern amqp_boolean_t amqp_data_in_buffer(amqp_connection_state_t state);
+// Expose amqp_rpc_reply to libraries
+extern amqp_rpc_reply_t amqp_get_rpc_reply();
 #ifdef __cplusplus
 }
 #endif