--- parser3/src/classes/amqp.C 2026/01/08 15:45:25 1.8 +++ parser3/src/classes/amqp.C 2026/10/08 18:56:00 1.26 @@ -1,29 +1,33 @@ /** @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 */ + +#include "classes.h" #include "pa_vmethod_frame.h" +#include "pa_vamqp.h" #include "pa_request.h" -#include "pa_vstring.h" -#include "pa_vhash.h" #include "pa_varray.h" #include "pa_vbool.h" -#include "pa_vamqp.h" +#include "pa_os.h" #ifdef WITH_AMQP + #include #include -#include #include -#include -#include + +#ifdef WITH_AMQP_SSL +#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; +#endif + +volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.26 2026/10/08 18:56:00 moko Exp $" IDENT_PA_VAMQP_H; class MAmqp: public Methoded { public: // VStateless_class @@ -36,10 +40,12 @@ DECLARE_CLASS_VAR(amqp, new MAmqp); #ifdef WITH_AMQP -static void status_check(int ret, const char *detail=""){ +static void status_check(VAmqp& self, int ret, const char *detail=""){ if(ret == AMQP_STATUS_OK) return; + self.fstate = VAmqp::CONNECTION_DEAD; + const char* error_str = amqp_error_string2(ret); if(error_str) { throw Exception("amqp", 0, "%sfailed: %s", detail, error_str); @@ -48,10 +54,12 @@ static void status_check(int ret, const } } -static void check(amqp_rpc_reply_t rr, const char *detail=""){ +static void check(VAmqp& self, amqp_rpc_reply_t rr, const char *detail=""){ if(rr.reply_type == AMQP_RESPONSE_NORMAL) return; + self.fstate = (rr.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION && rr.reply.id == AMQP_CHANNEL_CLOSE_METHOD) ? VAmqp::CHANNEL_DEAD : VAmqp::CONNECTION_DEAD; + // Extract error message from reply const char* error_msg = 0; size_t error_len = 0; @@ -70,87 +78,89 @@ static void check(amqp_rpc_reply_t rr, c } } } - + if(error_msg) { throw Exception("amqp", 0, "%sfailed: %.*s", detail, (int)error_len, error_msg); } else if(rr.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION) { - status_check(rr.library_error, detail); + status_check(self, rr.library_error, detail); } throw Exception("amqp", 0, "%sfailed", detail); } -#endif // WITH_AMQP - - -static void _create(Request& r, MethodParams& params) { -VAmqp& self=GET_SELF(r, VAmqp); - -#ifdef WITH_AMQP +static void amqp_connect(VAmqp& self, Request& r, HashStringValue* options) { const char* host_c = "localhost"; int port = 5672; const char* user_c = "guest"; const char* pass_c = "guest"; const char* vhost_c = "/"; const char* locale_c = "en_US"; - int heartbeat = 30; // seconds + int heartbeat_sec = 30; + int reconnect_interval_sec = 0; + bool tls_specified = false; +#ifdef WITH_AMQP_SSL const char* tls_ca = 0; const char* tls_cert = 0; const char* tls_key = 0; - bool tls_specified = false; bool tls_verify = true; +#endif - 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=="host"){ - host_c=value->as_string().cstr(); - } else if(key=="port"){ - port=r.process(*value).as_int(); - } else if(key=="user"){ - user_c=value->as_string().cstr(); - } else if(key=="password"){ - pass_c=value->as_string().cstr(); - } else if(key=="vhost"){ - vhost_c=value->as_string().cstr(); - } else if(key=="locale"){ - locale_c=value->as_string().cstr(); - } else if(key=="heartbeat"){ - heartbeat=r.process(*value).as_int(); - } else if(key=="tls"){ - tls_specified = true; - if(HashStringValue* tls_options=value->get_hash()){ - for(HashStringValue::Iterator t(*tls_options); t; t.next()){ - String::Body tkey=t.key(); - Value* tval=t.value(); - if(tkey=="ca"){ - tls_ca=tval->as_string().cstr(); - } else if(tkey=="cert"){ - tls_cert=tval->as_string().cstr(); - } else if(tkey=="key"){ - tls_key=tval->as_string().cstr(); - } else if(tkey=="verify"){ - tls_verify=r.process(*tval).as_bool(); - } else - throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); - } + if(options){ + for(HashStringValue::Iterator i(*options); i; i.next()){ + String::Body key=i.key(); + Value* value=i.value(); + if(key=="host"){ + host_c=value->as_string().cstr(); + } else if(key=="port"){ + port=r.process(*value).as_int(); + } else if(key=="user"){ + user_c=value->as_string().cstr(); + } else if(key=="password"){ + pass_c=value->as_string().cstr(); + } else if(key=="vhost"){ + vhost_c=value->as_string().cstr(); + } else if(key=="locale"){ + locale_c=value->as_string().cstr(); + } else if(key=="heartbeat"){ + heartbeat_sec=r.process(*value).as_int(); + } else if(key=="auto_reconnect"){ + reconnect_interval_sec=r.process(*value).as_int(); + } else if(key=="tls"){ +#ifdef WITH_AMQP_SSL + tls_specified = true; + if(HashStringValue* tls_options=value->get_hash()){ + for(HashStringValue::Iterator t(*tls_options); t; t.next()){ + String::Body tkey=t.key(); + Value* tval=t.value(); + if(tkey=="ca"){ + tls_ca=tval->as_string().cstr(); + } else if(tkey=="cert"){ + tls_cert=tval->as_string().cstr(); + } else if(tkey=="key"){ + tls_key=tval->as_string().cstr(); + } else if(tkey=="verify"){ + tls_verify=r.process(*tval).as_bool(); + } else + throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } - } else - throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); - } + } +#else + throw Exception("amqp", 0, "AMQP SSL support is disabled"); +#endif + } else + throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } } amqp_connection_state_t conn = amqp_new_connection(); amqp_socket_t* socket = 0; - + +#ifdef WITH_AMQP_SSL if(tls_specified) { socket = amqp_ssl_socket_new(conn); if(!socket) throw Exception("amqp", 0, "failed to create SSL socket"); - + // Set CA certificate if provided if(tls_ca) if(amqp_ssl_socket_set_cacert(socket, tls_ca)) @@ -168,31 +178,84 @@ VAmqp& self=GET_SELF(r, VAmqp); amqp_ssl_socket_set_verify_peer(socket, tls_verify && tls_ca); // If verify=true, enable hostname verification amqp_ssl_socket_set_verify_hostname(socket, tls_verify); - } else { + } else +#endif + { socket = amqp_tcp_socket_new(conn); if(!socket) throw Exception("amqp", 0, "failed to create TCP socket"); } - - status_check(amqp_socket_open(socket, host_c, port), tls_specified ? "open SSL socket " : "open TCP socket "); - amqp_rpc_reply_t rlogin = amqp_login(conn, vhost_c, 0, 131072, heartbeat, AMQP_SASL_METHOD_PLAIN, user_c, pass_c); + status_check(self, amqp_socket_open(socket, host_c, port), tls_specified ? "open SSL socket " : "open TCP socket "); + + amqp_rpc_reply_t rlogin = amqp_login(conn, vhost_c, 0, 131072, heartbeat_sec, AMQP_SASL_METHOD_PLAIN, user_c, pass_c); if(rlogin.reply_type != AMQP_RESPONSE_NORMAL){ amqp_destroy_connection(conn); - check(rlogin, "login "); + check(self, rlogin, "login "); } - int channel = 1; + amqp_channel_t channel = 1; amqp_channel_open(conn, channel); amqp_rpc_reply_t ropen = amqp_get_rpc_reply(conn); if(ropen.reply_type != AMQP_RESPONSE_NORMAL){ amqp_connection_close(conn, AMQP_REPLY_SUCCESS); amqp_destroy_connection(conn); - check(ropen, "open channel "); + check(self, ropen, "open channel "); } + if(self.fconnection) + amqp_destroy_connection(self.fconnection); + self.fconnection = conn; self.fchannel = channel; + self.fstate = VAmqp::ALIVE; + self.freconnect_interval = reconnect_interval_sec; +} + +// repairs a channel-level error cheaply: reopen on the same connection. Only valid when the server has +// already sent (or is about to send) channel.close - we're completing that handshake with close-ok, not +// initiating our own. +static void amqp_repair_channel(VAmqp& self) { + amqp_channel_close_ok_t close_ok; + amqp_send_method(self.fconnection, self.fchannel, AMQP_CHANNEL_CLOSE_OK_METHOD, &close_ok); + amqp_channel_open(self.fconnection, self.fchannel); + check(self, amqp_get_rpc_reply(self.fconnection), "channel repair "); + self.fstate = VAmqp::ALIVE; +} + +// voluntarily closes and reopens an otherwise-healthy channel +static void amqp_reset_channel(VAmqp& self) { + amqp_channel_close(self.fconnection, self.fchannel, AMQP_REPLY_SUCCESS); + amqp_channel_open(self.fconnection, self.fchannel); + amqp_get_rpc_reply(self.fconnection); +} + +void VAmqp::ensure_connected() { + if(fstate == ALIVE) + return; + + if(fstate == CHANNEL_DEAD){ + amqp_repair_channel(*this); + return; + } + + // CONNECTION_DEAD + if(freconnect_interval <= 0) + throw Exception("amqp", 0, "connection is dead (auto_reconnect is not enabled)"); + + pa_sleep(freconnect_interval, 0); + + amqp_connect(*this, pa_thread_request(), fcreate_options); +} + +#endif // WITH_AMQP + +static void _create(Request& r, MethodParams& params) { + VAmqp& self=GET_SELF(r, VAmqp); + +#ifdef WITH_AMQP + self.fcreate_options = params.count()>0 ? params.as_hash(0) : 0; + amqp_connect(self, r, self.fcreate_options); #else (void)params; (void)self; throw Exception("amqp", 0, "compiled without amqp support"); @@ -204,6 +267,97 @@ VAmqp& self=GET_SELF(r, VAmqp); #define AMQP_STRING(s,l) new String(String::C(pa_strdup((const char*)(s), (l)), (l))) #define AMQP_VSTRING(s,l) new VString(*AMQP_STRING(s,l)) +// broker-defined arguments/headers (x-message-ttl, x-dead-letter-exchange, ...) +// Type is taken from how the value was assigned: [] -> string, () -> number, (true|false) -> bool +static amqp_table_t amqp_build_arguments_table(Request& r, HashStringValue* arguments) { + amqp_table_t table = amqp_empty_table; + size_t count = arguments ? arguments->count() : 0; + if(!count) + return table; + + amqp_table_entry_t* entries = (amqp_table_entry_t*)pa_malloc(count * sizeof(amqp_table_entry_t)); + size_t i = 0; + for(HashStringValue::Iterator it(*arguments); it; it.next()){ + entries[i].key = amqp_cstring_bytes(it.key().cstr()); + + Value& value = r.process(*it.value()); + if(value.is_string() || value.is_void()){ + entries[i].value.kind = AMQP_FIELD_KIND_UTF8; + entries[i].value.value.bytes = amqp_cstring_bytes(value.as_string().cstr()); + } else if(value.is_bool()){ + entries[i].value.kind = AMQP_FIELD_KIND_BOOLEAN; + entries[i].value.value.boolean = value.as_bool(); + } else { + double d = value.as_double(); + double rounded = round(d); + if(ulp_eq_double(d, rounded)){ + entries[i].value.kind = AMQP_FIELD_KIND_I64; + entries[i].value.value.i64 = (int64_t)rounded; + } else { + entries[i].value.kind = AMQP_FIELD_KIND_F64; + entries[i].value.value.f64 = d; + } + } + i++; + } + + table.num_entries = (int)count; + table.entries = entries; + return table; +} + +// reverse of amqp_build_arguments_table - decodes a broker-provided field table (e.g. message headers) back into a hash. +static VHash* amqp_table_to_hash(amqp_table_t& table); + +static Value* amqp_field_to_value(amqp_field_value_t& f) { + switch(f.kind){ + case AMQP_FIELD_KIND_UTF8: + case AMQP_FIELD_KIND_BYTES: + return AMQP_VSTRING(f.value.bytes.bytes, f.value.bytes.len); + case AMQP_FIELD_KIND_BOOLEAN: + return &VBool::get(f.value.boolean); + case AMQP_FIELD_KIND_I8: return new VInt((pa_wint)f.value.i8); + case AMQP_FIELD_KIND_U8: return new VInt((pa_wint)f.value.u8); + case AMQP_FIELD_KIND_I16: return new VInt((pa_wint)f.value.i16); + case AMQP_FIELD_KIND_U16: return new VInt((pa_wint)f.value.u16); + case AMQP_FIELD_KIND_I32: return new VInt((pa_wint)f.value.i32); + case AMQP_FIELD_KIND_U32: return new VInt((pa_wint)f.value.u32); + case AMQP_FIELD_KIND_I64: return new VInt((pa_wint)f.value.i64); + case AMQP_FIELD_KIND_U64: return new VInt((pa_wint)f.value.u64); + case AMQP_FIELD_KIND_TIMESTAMP: return new VInt((pa_wint)f.value.u64); + case AMQP_FIELD_KIND_F32: return new VDouble((double)f.value.f32); + case AMQP_FIELD_KIND_F64: return new VDouble(f.value.f64); + case AMQP_FIELD_KIND_DECIMAL: { + double divisor=1; + for(uint8_t i=0; iarray(); + for(int i=0; iget_hash(); + for(int i=0; iput(*AMQP_STRING(e.key.bytes, e.key.len), amqp_field_to_value(e.value)); + } + return result; +} + static void _publish(Request& r, MethodParams& params) { VAmqp& self=GET_SELF(r, VAmqp); const String &msg=params.as_string(0, "msg must be string"); @@ -275,24 +429,14 @@ static void _publish(Request& r, MethodP const char* v=value->as_string().cstr(); props.app_id=amqp_cstring_bytes(v); props._flags|=AMQP_BASIC_APP_ID_FLAG; + } else if(key=="cluster_id"){ + const char* v=value->as_string().cstr(); + props.cluster_id=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_CLUSTER_ID_FLAG; } else if(key=="headers"){ -/* if(HashStringValue* hh=pval->get_hash()){ - size_t count=hh->count(); - amqp_table_entry_t* entries=count ? new amqp_table_entry_t[count] : 0; - size_t idx=0; - for(HashStringValue::Iterator hi(*hh); hi; hi.next()){ - String::Body hkey=hi.key(); - const char* hv=hi.value()->as_string().cstr(); - entries[idx].key=amqp_cstring_bytes(hkey.cstr()); - entries[idx].value.kind=AMQP_FIELD_KIND_UTF8; - entries[idx].value.value.bytes=amqp_cstring_bytes(hv); - idx++; - } - props.headers.num_entries=(int)count; - props.headers.entries=entries; - props._flags|=AMQP_BASIC_HEADERS_FLAG; - } -*/ } else + props.headers=amqp_build_arguments_table(r, value->get_hash()); + props._flags|=AMQP_BASIC_HEADERS_FLAG; + } else throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } } @@ -307,13 +451,7 @@ static void _publish(Request& r, MethodP int ret = amqp_basic_publish(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), mandatory, 0, &props, body); - if(ret!=AMQP_STATUS_OK) - throw Exception("amqp", 0, "publish failed"); - - // free temporary headers entries if allocated - if(props._flags & AMQP_BASIC_HEADERS_FLAG){ -// delete [] props.headers.entries; - } + status_check(self, ret, "publish "); } static void _release(Request& r, MethodParams&) { @@ -325,20 +463,25 @@ static void _release(Request& r, MethodP amqp_destroy_connection(conn); self.fconnection=0; self.fchannel=0; + self.fstate=VAmqp::CONNECTION_DEAD; } } +// 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); - if(ret!=AMQP_STATUS_OK) - throw Exception("amqp", 0, "ack failed"); + int ret = amqp_basic_ack(self.connection(), self.channel(), as_delivery_tag(params), 0); + status_check(self, ret, "ack "); } 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,9 +493,8 @@ static void _nack(Request& r, MethodPara } } } - int ret = amqp_basic_nack(self.connection(), self.channel(), pa_atoul(tag_s.cstr()), 0, requeue); - if(ret!=AMQP_STATUS_OK) - throw Exception("amqp", 0, "nack failed"); + int ret = amqp_basic_nack(self.connection(), self.channel(), tag, 0, requeue); + status_check(self, ret, "nack "); } static void _qos(Request& r, MethodParams& params) { @@ -369,14 +511,13 @@ static void _qos(Request& r, MethodParam } } } - amqp_basic_qos_ok_t *ret = amqp_basic_qos(self.connection(), self.channel(), 0, prefetch_count, 0); - if(!ret) - throw Exception("amqp", 0, "qos failed"); + amqp_basic_qos(self.connection(), self.channel(), 0, prefetch_count, 0); + check(self, amqp_get_rpc_reply(self.connection()), "qos "); } 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,9 +529,8 @@ static void _reject(Request& r, MethodPa } } } - int ret = amqp_basic_reject(self.connection(), self.channel(), pa_atoul(tag_s.cstr()), requeue); - if(ret!=AMQP_STATUS_OK) - throw Exception("amqp", 0, "reject failed"); + int ret = amqp_basic_reject(self.connection(), self.channel(), tag, requeue); + status_check(self, ret, "reject "); } static void _declare(Request& r, MethodParams& params) { @@ -398,7 +538,8 @@ 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; + amqp_table_t arguments=amqp_empty_table; if(HashStringValue* options=params.as_hash(0)){ for(HashStringValue::Iterator i(*options); i; i.next()){ String::Body key=i.key(); @@ -415,8 +556,12 @@ 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 if(key=="arguments"){ + arguments=amqp_build_arguments_table(r, value->get_hash()); } else throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } @@ -425,13 +570,13 @@ 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); - check(amqp_get_rpc_reply(self.connection())); + amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, internal, arguments); + check(self, 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); - check(amqp_get_rpc_reply(self.connection())); + 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, arguments); + check(self, amqp_get_rpc_reply(self.connection())); if(!*queue_c && ok){ r.write(*AMQP_STRING(ok->queue.bytes, ok->queue.len)); } @@ -442,7 +587,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 +600,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); } @@ -466,12 +609,12 @@ static void _delete(Request& r, MethodPa if(exchange_c){ amqp_exchange_delete(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), if_unused); - check(amqp_get_rpc_reply(self.connection())); + check(self, amqp_get_rpc_reply(self.connection())); } 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())); + check(self, amqp_get_rpc_reply(self.connection())); } } @@ -496,7 +639,7 @@ static void _bind(Request& r, MethodPara } if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required"); amqp_queue_bind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table); - check(amqp_get_rpc_reply(self.connection())); + check(self, amqp_get_rpc_reply(self.connection())); } static void _unbind(Request& r, MethodParams& params) { @@ -520,7 +663,7 @@ static void _unbind(Request& r, MethodPa } if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required"); amqp_queue_unbind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table); - check(amqp_get_rpc_reply(self.connection())); + check(self, amqp_get_rpc_reply(self.connection())); } static void _purge(Request& r, MethodParams& params) { @@ -539,8 +682,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)); - check(amqp_get_rpc_reply(self.connection())); + amqp_queue_purge_ok_t *ok = amqp_queue_purge(self.connection(), self.channel(), amqp_cstring_bytes(queue_c)); + check(self, amqp_get_rpc_reply(self.connection())); + r.write(*new VInt(ok ? ok->message_count : 0)); } static void _info(Request& r, MethodParams& params) { @@ -563,12 +707,14 @@ static void _info(Request& r, MethodPara 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())); + check(self, 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); } @@ -579,15 +725,117 @@ static VHash *amqp_message_hash(amqp_env 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)); + h->put("routing_key", AMQP_VSTRING(envelope.routing_key.bytes, envelope.routing_key.len)); + h->put("redelivered", &VBool::get(envelope.redelivered)); + + // properties are only put when the publisher actually set them (per the wire's own _flags bitmask) - + // same handful of allocations as the fields above, just conditional on what's actually present + amqp_basic_properties_t& props=envelope.message.properties; + if(props._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) + h->put("content_type", AMQP_VSTRING(props.content_type.bytes, props.content_type.len)); + if(props._flags & AMQP_BASIC_CONTENT_ENCODING_FLAG) + h->put("content_encoding", AMQP_VSTRING(props.content_encoding.bytes, props.content_encoding.len)); + if(props._flags & AMQP_BASIC_HEADERS_FLAG) + h->put("headers", amqp_table_to_hash(props.headers)); + if(props._flags & AMQP_BASIC_DELIVERY_MODE_FLAG) + h->put("delivery_mode", new VInt(props.delivery_mode)); + if(props._flags & AMQP_BASIC_PRIORITY_FLAG) + h->put("priority", new VInt(props.priority)); + if(props._flags & AMQP_BASIC_CORRELATION_ID_FLAG) + h->put("correlation_id", AMQP_VSTRING(props.correlation_id.bytes, props.correlation_id.len)); + if(props._flags & AMQP_BASIC_REPLY_TO_FLAG) + h->put("reply_to", AMQP_VSTRING(props.reply_to.bytes, props.reply_to.len)); + if(props._flags & AMQP_BASIC_EXPIRATION_FLAG) + h->put("expiration", AMQP_VSTRING(props.expiration.bytes, props.expiration.len)); + if(props._flags & AMQP_BASIC_MESSAGE_ID_FLAG) + h->put("message_id", AMQP_VSTRING(props.message_id.bytes, props.message_id.len)); + if(props._flags & AMQP_BASIC_TIMESTAMP_FLAG) + h->put("timestamp", new VInt((pa_wint)props.timestamp)); + if(props._flags & AMQP_BASIC_TYPE_FLAG) + h->put("type", AMQP_VSTRING(props.type.bytes, props.type.len)); + if(props._flags & AMQP_BASIC_USER_ID_FLAG) + h->put("user_id", AMQP_VSTRING(props.user_id.bytes, props.user_id.len)); + if(props._flags & AMQP_BASIC_APP_ID_FLAG) + h->put("app_id", AMQP_VSTRING(props.app_id.bytes, props.app_id.len)); + if(props._flags & AMQP_BASIC_CLUSTER_ID_FLAG) + h->put("cluster_id", AMQP_VSTRING(props.cluster_id.bytes, props.cluster_id.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; +} + +// amqp_consume_message() returns AMQP_STATUS_UNEXPECTED_STATE when it read a frame other than a deliver while polling. +// Per its own documented contract, read that pending frame to find out what it actually was: +// - a basic.return for an earlier mandatory publish that found no route: benign, doesn't close anything; +// drain the returned message (header+body, still pending right behind it) and retry - the real +// message (if any) is still coming +// - anything else: a genuine channel/connection close (typically after a protocol violation) - build a +// reply so check() can classify it correctly (cheap channel repair vs full reconnect) +static amqp_rpc_reply_t amqp_consume_message_checked(amqp_connection_state_t conn, amqp_channel_t channel, amqp_envelope_t* envelope, struct timeval* tv) { + for(;;) { + amqp_rpc_reply_t res = amqp_consume_message(conn, envelope, tv, 0); + if(res.reply_type != AMQP_RESPONSE_LIBRARY_EXCEPTION || res.library_error != AMQP_STATUS_UNEXPECTED_STATE) + return res; + + amqp_frame_t frame; + if(amqp_simple_wait_frame(conn, &frame) != AMQP_STATUS_OK || frame.frame_type != AMQP_FRAME_METHOD) + return res; // give up, let the original UNEXPECTED_STATE be treated as fatal + + if(frame.payload.method.id == AMQP_BASIC_RETURN_METHOD) { + amqp_message_t message; + if(amqp_read_message(conn, channel, &message, 0).reply_type == AMQP_RESPONSE_NORMAL) + amqp_destroy_message(&message); + continue; + } + + res.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION; + res.reply.id = frame.payload.method.id; + res.reply.decoded = frame.payload.method.decoded; + return res; + } +} + +// Auto-object used for cancelling the consumer when the enclosing scope exits. A cancel failure here must +// never mask whatever error is already propagating, so its result is discarded rather than check()ed. +class Temp_amqp_consumer { +public: + Temp_amqp_consumer(VAmqp& aself, const char* aconsumer_tag, bool adrain): self(aself), consumer_tag(aconsumer_tag), drain(adrain) {} + ~Temp_amqp_consumer() { + if(!consumer_tag || self.fstate != VAmqp::ALIVE) return; + amqp_basic_cancel(self.fconnection, self.fchannel, amqp_cstring_bytes(consumer_tag)); + amqp_get_rpc_reply(self.fconnection); + if(drain){ + // The broker may have already sent a deliver frame for this (now cancelled) consumer before it even saw our cancel, + // so it can still be sitting in the library's own buffer. Its delivery tag belongs to the pre-reset channel and + // won't exist anymore afterward, so drain anything already in flight - otherwise it would surface later, misattributed. + amqp_envelope_t envelope; + struct timeval no_wait = {0,0}; + while(amqp_consume_message(self.fconnection, &envelope, &no_wait, 0).reply_type == AMQP_RESPONSE_NORMAL) + amqp_destroy_envelope(&envelope); + amqp_reset_channel(self); + } + } +private: + VAmqp& self; + const char* consumer_tag; + bool drain; +}; + 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 ack=true, 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)){ @@ -600,31 +848,50 @@ static void _consume(Request& r, MethodP queue_c=value->as_string().cstr(); } else if(key=="consumer_tag"){ 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=="ack"){ + ack=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); - check(amqp_get_rpc_reply(self.connection())); + 0 /*no_local*/, !ack, exclusive, amqp_empty_table); + check(self, 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; + Temp_amqp_consumer cancel_consumer(self, consumer_tag_copy, ack && callback); + + struct timeval tv; if(callback){ self.fstop=false; - while(!self.fstop){ + for(int received=0; !self.fstop && (count<0 || received