--- parser3/src/classes/amqp.C 2026/01/08 15:45:25 1.8 +++ 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 */ @@ -23,7 +23,7 @@ #include #endif -volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.8 2026/01/08 15:45:25 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 @@ -328,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); - double tag=params.as_double(0, "delivery tag must be number", r); - int ret = amqp_basic_ack(self.connection(), self.channel(), (uint64_t)tag, 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)){ @@ -350,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"); } @@ -376,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)){ @@ -388,7 +393,7 @@ 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"); } @@ -398,7 +403,7 @@ static void _declare(Request& r, MethodP 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; + 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(); @@ -415,8 +420,10 @@ static void _declare(Request& r, MethodP 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 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); } @@ -425,12 +432,12 @@ static void _declare(Request& r, MethodP throw Exception("amqp", 0, "exchange or queue must be specified"); if(exchange_c){ - amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, nowait, amqp_empty_table); + 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, auto_delete, nowait, amqp_empty_table); + 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)); @@ -442,7 +449,7 @@ static void _delete(Request& r, MethodPa VAmqp& self=GET_SELF(r, VAmqp); const char* exchange_c = 0; const char* queue_c = 0; - bool if_unused=false, if_empty=false, nowait=false; + 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(); @@ -455,8 +462,6 @@ static void _delete(Request& r, MethodPa 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); } @@ -539,8 +544,9 @@ static void _purge(Request& r, MethodPar if(!queue_c) throw Exception("amqp", 0, "queue must be specified"); - amqp_queue_purge(self.connection(), self.channel(), amqp_cstring_bytes(queue_c)); + 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) { @@ -566,9 +572,11 @@ static void _info(Request& r, MethodPara check(amqp_get_rpc_reply(self.connection())); Value& result=*new VHash; - 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)); + 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); } @@ -582,12 +590,22 @@ static VHash *amqp_message_hash(amqp_env 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; - int count=1; + 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)){ @@ -602,29 +620,47 @@ 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 must be specified"); + if(count==0) throw Exception("amqp", 0, "count must not be zero"); + + // 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(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), + 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())); + // 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; - while(!self.fstop){ + for(int received=0; !self.fstop && (count<0 || received