--- parser3/src/classes/amqp.C 2025/11/07 22:36:35 1.3 +++ parser3/src/classes/amqp.C 2025/11/22 15:36:40 1.7 @@ -16,12 +16,13 @@ #ifdef WITH_AMQP #include #include +#include #include #include #include #endif -volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.3 2025/11/07 22:36:35 moko Exp $" IDENT_PA_VAMQP_H; +volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.7 2025/11/22 15:36:40 moko Exp $" IDENT_PA_VAMQP_H; class MAmqp: public Methoded { public: // VStateless_class @@ -32,6 +33,55 @@ public: DECLARE_CLASS_VAR(amqp, new MAmqp); +#ifdef WITH_AMQP + +static void status_check(int ret, const char *detail=""){ + if(ret == AMQP_STATUS_OK) + return; + + 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(amqp_rpc_reply_t rr, const char *detail=""){ + if(rr.reply_type == AMQP_RESPONSE_NORMAL) + return; + + // 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(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); @@ -43,6 +93,11 @@ 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; if(params.count()>0){ if(HashStringValue* options=params.as_hash(0)){ @@ -63,6 +118,24 @@ VAmqp& self=GET_SELF(r, VAmqp); 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); + } + } } else throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } @@ -70,16 +143,42 @@ VAmqp& self=GET_SELF(r, VAmqp); } 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(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(rlogin, "login "); } int channel = 1; @@ -88,7 +187,7 @@ 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(ropen, "open channel "); } self.fconnection = conn; @@ -96,16 +195,11 @@ VAmqp& self=GET_SELF(r, VAmqp); #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)) @@ -235,8 +329,8 @@ static void _release(Request& r, MethodP 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); + 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"); } @@ -328,7 +422,7 @@ static void _declare_exchange(Request& r 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())); + check(amqp_get_rpc_reply(self.connection())); } static void _delete_exchange(Request& r, MethodParams& params) { @@ -353,7 +447,7 @@ static void _delete_exchange(Request& r, } 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())); + check(amqp_get_rpc_reply(self.connection())); } static void _declare_queue(Request& r, MethodParams& params) { @@ -380,7 +474,7 @@ static void _declare_queue(Request& r, M } } 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())); + check(amqp_get_rpc_reply(self.connection())); if(!queue_c && ok){ r.write(*AMQP_STRING(ok->queue.bytes, ok->queue.len)); } @@ -409,7 +503,7 @@ static void _delete_queue(Request& r, Me } 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())); + check(amqp_get_rpc_reply(self.connection())); } static void _bind_queue(Request& r, MethodParams& params) { @@ -435,7 +529,7 @@ static void _bind_queue(Request& r, Meth } 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(amqp_get_rpc_reply(self.connection())); } static void _unbind_queue(Request& r, MethodParams& params) { @@ -461,7 +555,7 @@ static void _unbind_queue(Request& r, Me } 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(amqp_get_rpc_reply(self.connection())); } static void _consume(Request& r, MethodParams& params) { @@ -496,9 +590,7 @@ static void _consume(Request& r, MethodP 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"); + check(amqp_get_rpc_reply(self.connection())); self.fstop=false; while(!self.fstop){ @@ -509,7 +601,7 @@ static void _consume(Request& r, MethodP 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("delivery_tag", new VDouble((double)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)); @@ -533,7 +625,7 @@ static void _stop_consume(Request& r, Me self.fstop=true; } -#endif +#endif // WITH_AMQP // constructor MAmqp::MAmqp(): Methoded("amqp") { @@ -553,5 +645,5 @@ MAmqp::MAmqp(): Methoded("amqp") { add_native_method("unbind_queue", Method::CT_DYNAMIC, _unbind_queue, 0, 1); add_native_method("consume", Method::CT_DYNAMIC, _consume, 1, 1); add_native_method("stop_consume", Method::CT_DYNAMIC, _stop_consume, 0, 0); -#endif +#endif // WITH_AMQP }