--- parser3/src/classes/amqp.C	2025/11/08 00:27:08	1.6
+++ parser3/src/classes/amqp.C	2026/09/15 13:01:57	1.15
@@ -1,7 +1,7 @@
 /** @file
 	Parser: @b amqp parser class.
 
-	Copyright (c) 2001-2025 Art. Lebedev Studio (http://www.artlebedev.com)
+	Copyright (c) 2001-2026 Art. Lebedev Studio (https://www.artlebedev.com)
 	Authors: Konstantin Morshnev <moko@design.ru>
 */
 
@@ -10,6 +10,7 @@
 #include "pa_request.h"
 #include "pa_vstring.h"
 #include "pa_vhash.h"
+#include "pa_varray.h"
 #include "pa_vbool.h"
 #include "pa_vamqp.h"
 
@@ -22,7 +23,7 @@
 #include <string.h>
 #endif
 
-volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.6 2025/11/08 00:27:08 moko Exp $" IDENT_PA_VAMQP_H;
+volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.15 2026/09/15 13:01:57 moko Exp $" IDENT_PA_VAMQP_H;
 
 class MAmqp: public Methoded {
 public: // VStateless_class
@@ -327,17 +328,22 @@ static void _release(Request& r, MethodP
 	}
 }
 
+// to allow both [] and () call syntax
+static uint64_t as_delivery_tag(MethodParams& params, int index=0) {
+	const String &tag_s=params.as_string(index, "delivery tag must not be code");
+	return pa_atoul(tag_s.cstr());
+}
+
 static void _ack(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
-	const String &tag_s=params.as_string(0, "delivery tag must not be code");
-	int ret = amqp_basic_ack(self.connection(), self.channel(), pa_atoul(tag_s.cstr()), 0);
+	int ret = amqp_basic_ack(self.connection(), self.channel(), as_delivery_tag(params), 0);
 	if(ret!=AMQP_STATUS_OK)
 		throw Exception("amqp", 0, "ack failed");
 }
 
 static void _nack(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
-	const String &tag_s=params.as_string(0, "delivery tag must not be code");
+	uint64_t tag=as_delivery_tag(params);
 	bool requeue=false;
 	if(params.count()>1){
 		if(HashStringValue* options=params.as_hash(1)){
@@ -349,7 +355,7 @@ static void _nack(Request& r, MethodPara
 			}
 		}
 	}
-	int ret = amqp_basic_nack(self.connection(), self.channel(), pa_atoul(tag_s.cstr()), 0, requeue);
+	int ret = amqp_basic_nack(self.connection(), self.channel(), tag, 0, requeue);
 	if(ret!=AMQP_STATUS_OK)
 		throw Exception("amqp", 0, "nack failed");
 }
@@ -375,7 +381,7 @@ static void _qos(Request& r, MethodParam
 
 static void _reject(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
-	const String &tag_s=params.as_string(0, "delivery tag must not be code");
+	uint64_t tag=as_delivery_tag(params);
 	bool requeue=true; // by default return to queue
 	if(params.count()>1){
 		if(HashStringValue* options = params.as_hash(1)){
@@ -387,144 +393,110 @@ static void _reject(Request& r, MethodPa
 			}
 		}
 	}
-	int ret = amqp_basic_reject(self.connection(), self.channel(), pa_atoul(tag_s.cstr()), requeue);
+	int ret = amqp_basic_reject(self.connection(), self.channel(), tag, requeue);
 	if(ret!=AMQP_STATUS_OK)
 		throw Exception("amqp", 0, "reject failed");
 }
 
-static void _declare_exchange(Request& r, MethodParams& params) {
+static void _declare(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
-	const char* name_c = 0;
+	const char* exchange_c = 0;
+	const char* queue_c = 0;
 	const char* type_c = "direct";
-	bool passive=false, durable=false, auto_delete=true, nowait=false;
-	if(params.count()>0){
-		if(HashStringValue* options=params.as_hash(0)){
-			for(HashStringValue::Iterator i(*options); i; i.next()){
-				String::Body key=i.key();
-				Value* value=i.value();
-				if(key=="name"){
-					name_c=value->as_string().cstr();
-				} else if(key=="type"){
-					type_c=value->as_string().cstr();
-				} else if(key=="passive"){
-					passive=r.process(*value).as_bool();
-				} else if(key=="durable"){
-					durable=r.process(*value).as_bool();
-				} else if(key=="auto_delete"){
-					auto_delete=r.process(*value).as_bool();
-				} else if(key=="nowait"){
-					nowait=r.process(*value).as_bool();
-				} else
-					throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
-			}
+	bool passive=false, durable=false, auto_delete=false, internal=false, exclusive=false;
+	if(HashStringValue* options=params.as_hash(0)){
+		for(HashStringValue::Iterator i(*options); i; i.next()){
+			String::Body key=i.key();
+			Value* value=i.value();
+			if(key=="exchange"){
+				exchange_c=value->as_string().cstr();
+			} else if(key=="queue"){
+				queue_c=value->as_string().cstr();
+			} else if(key=="type"){
+				type_c=value->as_string().cstr();
+			} else if(key=="passive"){
+				passive=r.process(*value).as_bool();
+			} else if(key=="durable"){
+				durable=r.process(*value).as_bool();
+			} else if(key=="auto_delete"){
+				auto_delete=r.process(*value).as_bool();
+			} else if(key=="internal"){
+				internal=r.process(*value).as_bool();
+			} else if(key=="exclusive"){
+				exclusive=r.process(*value).as_bool();
+			} else
+				throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
 		}
 	}
-	if(!name_c)
-		throw Exception("amqp", 0, "name is required");
-	amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(name_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, nowait, amqp_empty_table);
-	check(amqp_get_rpc_reply(self.connection()));
-}
+	if(!exchange_c && !queue_c)
+		throw Exception("amqp", 0, "exchange or queue must be specified");
 
-static void _delete_exchange(Request& r, MethodParams& params) {
-	VAmqp& self=GET_SELF(r, VAmqp);
-	const char* name_c = 0;
-	bool if_unused=false, nowait=false;
-	if(params.count()>0){
-		if(HashStringValue* options=params.as_hash(0)){
-			for(HashStringValue::Iterator i(*options); i; i.next()){
-				String::Body key=i.key();
-				Value* value=i.value();
-				if(key=="exchange"){
-					name_c=value->as_string().cstr();
-				} else if(key=="if_unused"){
-					if_unused=r.process(*value).as_bool();
-				} else if(key=="nowait"){
-					nowait=r.process(*value).as_bool();
-				} else
-					throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
-			}
+	if(exchange_c){
+		amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, internal, amqp_empty_table);
+		check(amqp_get_rpc_reply(self.connection()));
+	}
+
+	if(queue_c){
+		amqp_queue_declare_ok_t *ok = amqp_queue_declare(self.connection(), self.channel(), *queue_c ? amqp_cstring_bytes(queue_c) : amqp_empty_bytes, passive, durable, exclusive, auto_delete, amqp_empty_table);
+		check(amqp_get_rpc_reply(self.connection()));
+		if(!*queue_c && ok){
+			r.write(*AMQP_STRING(ok->queue.bytes, ok->queue.len));
 		}
 	}
-	if(!name_c) throw Exception("amqp", 0, "exchange is required");
-	amqp_exchange_delete(self.connection(), self.channel(), amqp_cstring_bytes(name_c), if_unused);
-	check(amqp_get_rpc_reply(self.connection()));
 }
 
-static void _declare_queue(Request& r, MethodParams& params) {
+static void _delete(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
-	const char* queue_c = 0; bool passive=false, durable=false, auto_delete=true, nowait=false;
-	if(params.count()>0){
-		if(HashStringValue* options=params.as_hash(0)){
-			for(HashStringValue::Iterator i(*options); i; i.next()){
-				String::Body key=i.key();
-				Value* value=i.value();
-				if(key=="queue"){
-					queue_c=value->as_string().cstr();
-				} else if(key=="passive"){
-					passive=r.process(*value).as_bool();
-				} else if(key=="durable"){
-					durable=r.process(*value).as_bool();
-				} else if(key=="auto_delete"){
-					auto_delete=r.process(*value).as_bool();
-				} else if(key=="nowait"){
-					nowait=r.process(*value).as_bool();
-				} else
-					throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
-			}
+	const char* exchange_c = 0;
+	const char* queue_c = 0;
+	bool if_unused=false, if_empty=false;
+	if(HashStringValue* options=params.as_hash(0)){
+		for(HashStringValue::Iterator i(*options); i; i.next()){
+			String::Body key=i.key();
+			Value* value=i.value();
+			if(key=="exchange"){
+				exchange_c=value->as_string().cstr();
+			} else if(key=="queue"){
+				queue_c=value->as_string().cstr();
+			} else if(key=="if_unused"){
+				if_unused=r.process(*value).as_bool();
+			} else if(key=="if_empty"){
+				if_empty=r.process(*value).as_bool();
+			} else
+				throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
 		}
 	}
-	amqp_queue_declare_ok_t *ok = amqp_queue_declare(self.connection(), self.channel(), queue_c ? amqp_cstring_bytes(queue_c) : amqp_empty_bytes, passive, durable, auto_delete, nowait, amqp_empty_table);
-	check(amqp_get_rpc_reply(self.connection()));
-	if(!queue_c && ok){
-		r.write(*AMQP_STRING(ok->queue.bytes, ok->queue.len));
+	if(!exchange_c && !queue_c)
+		throw Exception("amqp", 0, "exchange or queue must be specified");
+
+	if(exchange_c){
+		amqp_exchange_delete(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), if_unused);
+		check(amqp_get_rpc_reply(self.connection()));
 	}
-}
 
-static void _delete_queue(Request& r, MethodParams& params) {
-	VAmqp& self=GET_SELF(r, VAmqp);
-	const char* queue_c = 0; bool if_unused=false, if_empty=false, nowait=false;
-	if(params.count()>0){
-		if(HashStringValue* options=params.as_hash(0)){
-			for(HashStringValue::Iterator i(*options); i; i.next()){
-				String::Body key=i.key();
-				Value* value=i.value();
-				if(key=="queue"){
-					queue_c=value->as_string().cstr();
-				} else if(key=="if_unused"){
-					if_unused=r.process(*value).as_bool();
-				} else if(key=="if_empty"){
-					if_empty=r.process(*value).as_bool();
-				} else if(key=="nowait"){
-					nowait=r.process(*value).as_bool();
-				} else
-					throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
-			}
-		}
+	if(queue_c){
+		amqp_queue_delete(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), if_unused, if_empty);
+		check(amqp_get_rpc_reply(self.connection()));
 	}
-	if(!queue_c) throw Exception("amqp", 0, "queue is required");
-	amqp_queue_delete(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), if_unused, if_empty);
-	check(amqp_get_rpc_reply(self.connection()));
 }
 
-static void _bind_queue(Request& r, MethodParams& params) {
+static void _bind(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
 	const char* exchange_c=0;
 	const char* queue_c=0;
 	const char* routing_key_c="";
-	if(params.count()>0){
-		if(HashStringValue* options=params.as_hash(0)){
-			for(HashStringValue::Iterator i(*options); i; i.next()){
-				String::Body key=i.key();
-				Value* value=i.value();
-				if(key=="exchange"){
-					exchange_c=value->as_string().cstr();
-				} else if(key=="queue"){
-					queue_c=value->as_string().cstr();
-				} else if(key=="routing_key"){
-					routing_key_c=value->as_string().cstr();
-				} else
-					throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
-			}
+	if(HashStringValue* options=params.as_hash(0)){
+		for(HashStringValue::Iterator i(*options); i; i.next()){
+			String::Body key=i.key();
+			Value* value=i.value();
+			if(key=="exchange"){
+				exchange_c=value->as_string().cstr();
+			} else if(key=="queue"){
+				queue_c=value->as_string().cstr();
+			} else if(key=="routing_key"){
+				routing_key_c=value->as_string().cstr();
+			} else
+				throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
 		}
 	}
 	if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");
@@ -532,25 +504,23 @@ static void _bind_queue(Request& r, Meth
 	check(amqp_get_rpc_reply(self.connection()));
 }
 
-static void _unbind_queue(Request& r, MethodParams& params) {
+static void _unbind(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
 	const char* exchange_c=0;
 	const char* queue_c=0;
 	const char* routing_key_c="";
-	if(params.count()>0){
-		if(HashStringValue* options=params.as_hash(0)){
-			for(HashStringValue::Iterator i(*options); i; i.next()){
-				String::Body key=i.key();
-				Value* value=i.value();
-				if(key=="exchange"){
-					exchange_c=value->as_string().cstr();
-				} else if(key=="queue"){
-					queue_c=value->as_string().cstr();
-				} else if(key=="routing_key"){
-					routing_key_c=value->as_string().cstr();
-				} else
-					throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
-			}
+	if(HashStringValue* options=params.as_hash(0)){
+		for(HashStringValue::Iterator i(*options); i; i.next()){
+			String::Body key=i.key();
+			Value* value=i.value();
+			if(key=="exchange"){
+				exchange_c=value->as_string().cstr();
+			} else if(key=="queue"){
+				queue_c=value->as_string().cstr();
+			} else if(key=="routing_key"){
+				routing_key_c=value->as_string().cstr();
+			} else
+				throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
 		}
 	}
 	if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");
@@ -558,11 +528,84 @@ static void _unbind_queue(Request& r, Me
 	check(amqp_get_rpc_reply(self.connection()));
 }
 
+static void _purge(Request& r, MethodParams& params) {
+	VAmqp& self=GET_SELF(r, VAmqp);
+	const char* queue_c = 0;
+	if(HashStringValue* options=params.as_hash(0)){
+		for(HashStringValue::Iterator i(*options); i; i.next()){
+			String::Body key=i.key();
+			Value* value=i.value();
+			if(key=="queue"){
+				queue_c=value->as_string().cstr();
+			} else
+				throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
+		}
+	}
+	if(!queue_c)
+		throw Exception("amqp", 0, "queue must be specified");
+
+	amqp_queue_purge_ok_t *ok = amqp_queue_purge(self.connection(), self.channel(), amqp_cstring_bytes(queue_c));
+	check(amqp_get_rpc_reply(self.connection()));
+	r.write(*new VInt(ok ? ok->message_count : 0));
+}
+
+static void _info(Request& r, MethodParams& params) {
+	VAmqp& self = GET_SELF(r, VAmqp);
+	const char* queue_c = 0;
+
+	if (HashStringValue* options=params.as_hash(0)) {
+		for (HashStringValue::Iterator i(*options); i; i.next()) {
+			String::Body key=i.key();
+			Value* value=i.value();
+			if(key=="queue"){
+				queue_c=value->as_string().cstr();
+			} else {
+				throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
+			}
+		}
+	}
+
+	if (!queue_c)
+		throw Exception("amqp", 0, "queue must be specified");
+
+	amqp_queue_declare_ok_t* ok = amqp_queue_declare(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), /*passive*/ 1, 0, 0, 0, amqp_empty_table);
+	check(amqp_get_rpc_reply(self.connection()));
+
+	Value& result=*new VHash;
+	if(ok){
+		result.put_element(*new String("queue"), AMQP_VSTRING(ok->queue.bytes, ok->queue.len));
+		result.put_element(*new String("messages"), new VInt(ok->message_count));
+		result.put_element(*new String("consumers"), new VInt(ok->consumer_count));
+	}
+	r.write(result);
+}
+
+static VHash *amqp_message_hash(amqp_envelope_t &envelope) {
+	VHash *result=new VHash;
+	HashStringValue* h=result->get_hash();
+	h->put("msg", AMQP_VSTRING(envelope.message.body.bytes, envelope.message.body.len));
+	h->put("delivery_tag", new VInt(envelope.delivery_tag));
+	h->put("consumer_tag", AMQP_VSTRING(envelope.consumer_tag.bytes, envelope.consumer_tag.len));
+	h->put("exchange", AMQP_VSTRING(envelope.exchange.bytes, envelope.exchange.len));
+	return result;
+}
+
+// timeout<0 - wait forever (NULL), timeout==0 - don't wait (poll), timeout>0 - wait up to timeout seconds
+static struct timeval* amqp_wait_tv(struct timeval& tv, int timeout){
+	if(timeout<0)
+		return NULL;
+	tv.tv_sec=timeout;
+	tv.tv_usec=0;
+	return &tv;
+}
+
 static void _consume(Request& r, MethodParams& params) {
 	VAmqp& self=GET_SELF(r, VAmqp);
 	const char* queue_c=0;
 	const char* consumer_tag_c=0;
-	bool no_ack=true, nowait=false;
+	bool no_ack=false, exclusive=false;
+	int count=1; bool count_specified=false;
+	int timeout=0; bool timeout_specified=false;
 	Junction* callback=0;
 
 	if(HashStringValue* options=params.as_hash(0)){
@@ -577,46 +620,87 @@ static void _consume(Request& r, MethodP
 				consumer_tag_c=value->as_string().cstr();
 			} else if(key=="no_ack"){
 				no_ack=r.process(*value).as_bool();
-			} else if(key=="nowait"){
-				nowait=r.process(*value).as_bool();
+			} else if(key=="exclusive"){
+				exclusive=r.process(*value).as_bool();
+			} else if(key=="count"){
+				count=r.process(*value).as_int();
+				count_specified=true;
+			} else if(key=="timeout"){
+				timeout=r.process(*value).as_int();
+				timeout_specified=true;
 			} else
 				throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
 		}
 	}
 
-	if(!queue_c) throw Exception("amqp", 0, "queue is required");
-	if(!callback) throw Exception("amqp", 0, "callback is required");
+	if(!queue_c) throw Exception("amqp", 0, "queue must be specified");
+	if(count==0) throw Exception("amqp", 0, "count must not be zero");
 
-	amqp_basic_consume(self.connection(), self.channel(), amqp_cstring_bytes(queue_c),
+	// listen mode (callback): count/timeout unlimited by default - runs until stop_consume/timeout/error
+	// pull mode (no callback): count(1)/timeout(0) by default - returns whatever is ready right now
+	if(!count_specified)
+		count = callback ? -1 : 1;
+	if(!timeout_specified)
+		timeout = callback ? -1 : 0;
+
+	amqp_basic_consume_ok_t *ok = amqp_basic_consume(self.connection(), self.channel(), amqp_cstring_bytes(queue_c),
 		consumer_tag_c ? amqp_cstring_bytes(consumer_tag_c) : amqp_empty_bytes,
-		0 /*no_local*/, no_ack, nowait, amqp_empty_table);
+		0 /*no_local*/, no_ack, exclusive, amqp_empty_table);
 	check(amqp_get_rpc_reply(self.connection()));
 
-	self.fstop=false;
-	while(!self.fstop){
-		amqp_envelope_t envelope;
-		memset(&envelope, 0, sizeof(envelope));
-		amqp_maybe_release_buffers(self.connection());
-		amqp_rpc_reply_t res = amqp_consume_message(self.connection(), &envelope, NULL, 0);
-		if(res.reply_type == AMQP_RESPONSE_NORMAL){
-			VHash &vh=*new VHash; HashStringValue* h=vh.get_hash();
-			h->put("msg", AMQP_VSTRING(envelope.message.body.bytes, envelope.message.body.len));
-			h->put("delivery_tag", new VString(String::Body::uitoa((unsigned long long)envelope.delivery_tag)));
-			h->put("consumer_tag", AMQP_VSTRING(envelope.consumer_tag.bytes, envelope.consumer_tag.len));
-			h->put("exchange", AMQP_VSTRING(envelope.exchange.bytes, envelope.exchange.len));
-
-			Value *params_cb[]={&vh};
-			METHOD_FRAME_ACTION(*callback->method, r.method_frame, callback->self, {
-				frame.store_params(params_cb, 1);
-				r.call(frame);
-			});
-
-			amqp_destroy_envelope(&envelope);
-		} else if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) {
-			continue;
-		} else {
-			break;
+	// amqp_maybe_release_buffers() recycles ok->consumer_tag
+	const char* consumer_tag_copy = ok ? pa_strdup((const char*)ok->consumer_tag.bytes, ok->consumer_tag.len) : 0;
+
+	struct timeval tv;
+
+	if(callback){
+		self.fstop=false;
+		for(int received=0; !self.fstop && (count<0 || received<count); received++){
+			amqp_envelope_t envelope;
+			memset(&envelope, 0, sizeof(envelope));
+			amqp_maybe_release_buffers(self.connection());
+			// timeout is the max idle gap between messages, re-armed on every wait
+			amqp_rpc_reply_t res = amqp_consume_message(self.connection(), &envelope, amqp_wait_tv(tv, timeout), 0);
+			if(res.reply_type == AMQP_RESPONSE_NORMAL){
+				VHash *vh=amqp_message_hash(envelope);
+				Value *params_cb[]={vh};
+				METHOD_FRAME_ACTION(*callback->method, r.method_frame, callback->self, {
+					frame.store_params(params_cb, 1);
+					r.call(frame);
+				});
+				amqp_destroy_envelope(&envelope);
+			} else if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) {
+				break; // idle too long - stop cleanly
+			} else {
+				check(res); // real error - throw
+			}
 		}
+	} else {
+		VArray& result=*new VArray();
+		ArrayValue& result_array=result.array();
+
+		for(int received=0; count<0 || received<count; received++){
+			amqp_envelope_t envelope;
+			memset(&envelope, 0, sizeof(envelope));
+			amqp_maybe_release_buffers(self.connection());
+			// timeout applies only to the first message of the batch;
+			// the rest is drained with a non-blocking poll until nothing more is ready
+			amqp_rpc_reply_t res = amqp_consume_message(self.connection(), &envelope, amqp_wait_tv(tv, received==0 ? timeout : 0), 0);
+			if(res.reply_type == AMQP_RESPONSE_NORMAL){
+				result_array+=amqp_message_hash(envelope);
+				amqp_destroy_envelope(&envelope);
+			} else if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) {
+				break; // nothing (more) ready - return what we have
+			} else {
+				check(res); // real error - throw
+			}
+		}
+		r.write(result);
+	}
+
+	if(ok){
+		amqp_basic_cancel(self.connection(), self.channel(), amqp_cstring_bytes(consumer_tag_copy));
+		check(amqp_get_rpc_reply(self.connection()));
 	}
 }
 
@@ -637,12 +721,12 @@ MAmqp::MAmqp(): Methoded("amqp") {
 	add_native_method("nack", Method::CT_DYNAMIC, _nack, 1, 2);
 	add_native_method("reject", Method::CT_DYNAMIC, _reject, 1, 2);
 	add_native_method("qos", Method::CT_DYNAMIC, _qos, 0, 1);
-	add_native_method("declare_exchange", Method::CT_DYNAMIC, _declare_exchange, 0, 1);
-	add_native_method("delete_exchange", Method::CT_DYNAMIC, _delete_exchange, 0, 1);
-	add_native_method("declare_queue", Method::CT_DYNAMIC, _declare_queue, 0, 1);
-	add_native_method("delete_queue", Method::CT_DYNAMIC, _delete_queue, 0, 1);
-	add_native_method("bind_queue", Method::CT_DYNAMIC, _bind_queue, 0, 1);
-	add_native_method("unbind_queue", Method::CT_DYNAMIC, _unbind_queue, 0, 1);
+	add_native_method("declare", Method::CT_DYNAMIC, _declare, 1, 1);
+	add_native_method("delete", Method::CT_DYNAMIC, _delete, 1, 1);
+	add_native_method("bind", Method::CT_DYNAMIC, _bind, 1, 1);
+	add_native_method("unbind", Method::CT_DYNAMIC, _unbind, 1, 1);
+	add_native_method("purge", Method::CT_DYNAMIC, _purge, 1, 1);
+	add_native_method("info", Method::CT_DYNAMIC, _info, 1, 1);
 	add_native_method("consume", Method::CT_DYNAMIC, _consume, 1, 1);
 	add_native_method("stop_consume", Method::CT_DYNAMIC, _stop_consume, 0, 0);
 #endif // WITH_AMQP