--- parser3/src/classes/amqp.C 2025/11/07 01:26:13 1.2 +++ parser3/src/classes/amqp.C 2026/09/15 17:53:05 1.16 @@ -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 */ @@ -10,18 +10,22 @@ #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" +#include "pa_globals.h" #ifdef WITH_AMQP #include #include +#include #include #include #include #endif -volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.2 2025/11/07 01:26:13 moko Exp $" IDENT_PA_VAMQP_H; +volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.16 2026/09/15 17:53:05 moko Exp $" IDENT_PA_VAMQP_H; class MAmqp: public Methoded { public: // VStateless_class @@ -32,10 +36,57 @@ public: DECLARE_CLASS_VAR(amqp, new MAmqp); -static void _create(Request& r, MethodParams& params) { -VAmqp& self=GET_SELF(r, VAmqp); - #ifdef WITH_AMQP + +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); + } else { + throw Exception("amqp", 0, "%sfailed: error %d", detail, ret); + } +} + +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; + if(rr.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION) { + if(rr.reply.id == AMQP_CHANNEL_CLOSE_METHOD) { + amqp_channel_close_t *m = (amqp_channel_close_t *)rr.reply.decoded; + if(m->reply_text.len > 0 && m->reply_text.bytes) { + error_msg = (const char*)m->reply_text.bytes; + error_len = m->reply_text.len; + } + } else if(rr.reply.id == AMQP_CONNECTION_CLOSE_METHOD) { + amqp_connection_close_t *m = (amqp_connection_close_t *)rr.reply.decoded; + if(m->reply_text.len > 0 && m->reply_text.bytes) { + error_msg = (const char*)m->reply_text.bytes; + error_len = m->reply_text.len; + } + } + } + + 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(self, rr.library_error, detail); + } + + throw Exception("amqp", 0, "%sfailed", detail); +} + +static void amqp_connect(VAmqp& self, Request& r, HashStringValue* options) { const char* host_c = "localhost"; int port = 5672; const char* user_c = "guest"; @@ -43,43 +94,93 @@ VAmqp& self=GET_SELF(r, VAmqp); const char* vhost_c = "/"; const char* locale_c = "en_US"; int heartbeat = 30; // seconds + const char* tls_ca = 0; + const char* tls_cert = 0; + const char* tls_key = 0; + bool tls_specified = false; + bool tls_verify = true; + int reconnect_interval_sec = 0; - 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 - 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=r.process(*value).as_int(); + } else if(key=="auto_reconnect"){ + reconnect_interval_sec=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); + } + } + } else + throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } } amqp_connection_state_t conn = amqp_new_connection(); - amqp_socket_t* socket = amqp_tcp_socket_new(conn); - if(!socket) - throw Exception("amqp", 0, "failed to create TCP socket"); - if(amqp_socket_open(socket, host_c, port)) - throw Exception("amqp", 0, "failed to open TCP socket"); + amqp_socket_t* socket = 0; + + 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)) + throw Exception("amqp", 0, "failed to set CA certificate"); + + // Set client certificate and key if provided + if(tls_cert && tls_key) { + if(amqp_ssl_socket_set_key(socket, tls_cert, tls_key)) + throw Exception("amqp", 0, "failed to set client certificate/key"); + } else if(tls_cert || tls_key) { + throw Exception("amqp", 0, "both cert and key must be specified for TLS"); + } + + // If CA is provided, peer verification will use it + 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 { + socket = amqp_tcp_socket_new(conn); + if(!socket) + throw Exception("amqp", 0, "failed to create TCP socket"); + } + + 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, AMQP_SASL_METHOD_PLAIN, user_c, pass_c); if(rlogin.reply_type != AMQP_RESPONSE_NORMAL){ amqp_destroy_connection(conn); - throw Exception("amqp", 0, "login failed"); + check(self, rlogin, "login "); } int channel = 1; @@ -88,24 +189,61 @@ VAmqp& self=GET_SELF(r, VAmqp); if(ropen.reply_type != AMQP_RESPONSE_NORMAL){ amqp_connection_close(conn, AMQP_REPLY_SUCCESS); amqp_destroy_connection(conn); - throw Exception("amqp", 0, "channel open failed"); + 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 +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; +} + +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"); -#endif +#endif // WITH_AMQP } #ifdef WITH_AMQP -static void check(const char* action, amqp_rpc_reply_t rr){ - if(rr.reply_type != AMQP_RESPONSE_NORMAL) - throw Exception("amqp", 0, "%s failed", action); -} - #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)) @@ -114,7 +252,6 @@ static void _publish(Request& r, MethodP const String &msg=params.as_string(0, "msg must be string"); const char* exchange_c = ""; // default exchange const char* routing_key_c = 0; - bool have_queue_sugar=false; bool mandatory=false; amqp_basic_properties_t props; @@ -130,85 +267,75 @@ static void _publish(Request& r, MethodP } else if(key=="routing_key"){ routing_key_c=value->as_string().cstr(); } else if(key=="queue"){ - routing_key_c=value->as_string().cstr(); have_queue_sugar=true; + routing_key_c=value->as_string().cstr(); } else if(key=="mandatory"){ mandatory=r.process(*value).as_bool(); - } else if(key=="properties"){ - // parse message properties - if(HashStringValue* ph=value->get_hash()){ - for(HashStringValue::Iterator p(*ph); p; p.next()){ - String::Body pkey=p.key(); - Value* pval=p.value(); - if(pkey=="content_type"){ - const char* v=pval->as_string().cstr(); - props.content_type=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_CONTENT_TYPE_FLAG; - } else if(pkey=="content_encoding"){ - const char* v=pval->as_string().cstr(); - props.content_encoding=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_CONTENT_ENCODING_FLAG; - } else if(pkey=="delivery_mode"){ - uint8_t dm=(uint8_t)pval->as_int(); - props.delivery_mode=dm; - props._flags|=AMQP_BASIC_DELIVERY_MODE_FLAG; - } else if(pkey=="priority"){ - uint8_t pr=(uint8_t)pval->as_int(); - props.priority=pr; - props._flags|=AMQP_BASIC_PRIORITY_FLAG; - } else if(pkey=="correlation_id"){ - const char* v=pval->as_string().cstr(); - props.correlation_id=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_CORRELATION_ID_FLAG; - } else if(pkey=="reply_to"){ - const char* v=pval->as_string().cstr(); - props.reply_to=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_REPLY_TO_FLAG; - } else if(pkey=="expiration"){ - const char* v=pval->as_string().cstr(); - props.expiration=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_EXPIRATION_FLAG; - } else if(pkey=="message_id"){ - const char* v=pval->as_string().cstr(); - props.message_id=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_MESSAGE_ID_FLAG; - } else if(pkey=="timestamp"){ - uint64_t ts=(uint64_t)pval->as_double(); - props.timestamp=ts; - props._flags|=AMQP_BASIC_TIMESTAMP_FLAG; - } else if(pkey=="type"){ - const char* v=pval->as_string().cstr(); - props.type=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_TYPE_FLAG; - } else if(pkey=="user_id"){ - const char* v=pval->as_string().cstr(); - props.user_id=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_USER_ID_FLAG; - } else if(pkey=="app_id"){ - const char* v=pval->as_string().cstr(); - props.app_id=amqp_cstring_bytes(v); - props._flags|=AMQP_BASIC_APP_ID_FLAG; - } else if(pkey=="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 - throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); + } else if(key=="content_type"){ + const char* v=value->as_string().cstr(); + props.content_type=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_CONTENT_TYPE_FLAG; + } else if(key=="content_encoding"){ + const char* v=value->as_string().cstr(); + props.content_encoding=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_CONTENT_ENCODING_FLAG; + } else if(key=="delivery_mode"){ + uint8_t dm=(uint8_t)value->as_int(); + props.delivery_mode=dm; + props._flags|=AMQP_BASIC_DELIVERY_MODE_FLAG; + } else if(key=="priority"){ + uint8_t pr=(uint8_t)value->as_int(); + props.priority=pr; + props._flags|=AMQP_BASIC_PRIORITY_FLAG; + } else if(key=="correlation_id"){ + const char* v=value->as_string().cstr(); + props.correlation_id=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_CORRELATION_ID_FLAG; + } else if(key=="reply_to"){ + const char* v=value->as_string().cstr(); + props.reply_to=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_REPLY_TO_FLAG; + } else if(key=="expiration"){ + const char* v=value->as_string().cstr(); + props.expiration=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_EXPIRATION_FLAG; + } else if(key=="message_id"){ + const char* v=value->as_string().cstr(); + props.message_id=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_MESSAGE_ID_FLAG; + } else if(key=="timestamp"){ + uint64_t ts=(uint64_t)value->as_double(); + props.timestamp=ts; + props._flags|=AMQP_BASIC_TIMESTAMP_FLAG; + } else if(key=="type"){ + const char* v=value->as_string().cstr(); + props.type=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_TYPE_FLAG; + } else if(key=="user_id"){ + const char* v=value->as_string().cstr(); + props.user_id=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_USER_ID_FLAG; + } else if(key=="app_id"){ + const char* v=value->as_string().cstr(); + props.app_id=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_APP_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 +*/ } else throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } } @@ -221,16 +348,14 @@ static void _publish(Request& r, MethodP body.len = msg.length(); body.bytes=(void*)msg.cstr(); - int ret = amqp_basic_publish(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), mandatory , 0, &props, body); + 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"); + status_check(self, ret, "publish "); // free temporary headers entries if allocated if(props._flags & AMQP_BASIC_HEADERS_FLAG){ // delete [] props.headers.entries; } - (void)have_queue_sugar; } static void _release(Request& r, MethodParams&) { @@ -242,20 +367,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); - 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); - 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)){ @@ -267,9 +397,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) { @@ -286,14 +415,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)){ @@ -305,238 +433,341 @@ 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_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("declare exchange", 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(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, exclusive, auto_delete, amqp_empty_table); + check(self, 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("delete exchange", 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("declare queue", 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(self, 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(self, 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("delete queue", 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"); 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("bind queue", amqp_get_rpc_reply(self.connection())); + check(self, 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"); 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("unbind queue", amqp_get_rpc_reply(self.connection())); + check(self, 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(self, 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(self, 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; +} + +// amqp_consume_message() returns AMQP_STATUS_UNEXPECTED_STATE when it read a frame other than a deliver +// while polling - typically the broker notifying a channel/connection close after a protocol violation +// Per its own documented contract, read that pending frame to find out which, so check() can classify +// it correctly (cheap channel repair vs full reconnect) instead of always assuming the worse. +static amqp_rpc_reply_t amqp_resolve_unexpected_state(amqp_connection_state_t conn) { + amqp_rpc_reply_t result; + memset(&result, 0, sizeof(result)); + + amqp_frame_t frame; + if(amqp_simple_wait_frame(conn, &frame) != AMQP_STATUS_OK || frame.frame_type != AMQP_FRAME_METHOD) { + result.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION; + result.library_error = AMQP_STATUS_UNEXPECTED_STATE; + return result; + } + + result.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION; + result.reply.id = frame.payload.method.id; + result.reply.decoded = frame.payload.method.decoded; + return result; +} + +static void amqp_check_consume_result(VAmqp& self, amqp_rpc_reply_t& res) { + if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_UNEXPECTED_STATE) + check(self, amqp_resolve_unexpected_state(self.connection()), "consume "); // read the real frame, classify properly + else + check(self, res, "consume "); // real error - throw } 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)){ for(HashStringValue::Iterator i(*options); i; i.next()){ String::Body key=i.key(); Value* value=i.value(); - if(key=="queue"){ + if(key=="callback"){ + callback=value->get_junction(); + } else if(key=="queue"){ 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=="callback"){ - callback=value->get_junction(); + } 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); - amqp_rpc_reply_t rr = amqp_get_rpc_reply(self.connection()); - if(rr.reply_type != AMQP_RESPONSE_NORMAL) - throw Exception("amqp", 0, "consume failed"); - - 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; - } + 0 /*no_local*/, no_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; + + struct timeval tv; + + if(callback){ + self.fstop=false; + for(int received=0; !self.fstop && (count<0 || receivedmethod, 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 { + amqp_check_consume_result(self, res); + } + } + } else { + VArray& result=*new VArray(); + ArrayValue& result_array=result.array(); + + for(int received=0; count<0 || received