--- parser3/src/classes/amqp.C 2026/09/18 17:10:15 1.20 +++ parser3/src/classes/amqp.C 2026/10/09 22:05:41 1.29 @@ -5,38 +5,21 @@ Authors: Konstantin Morshnev */ + +#include "classes.h" #include "pa_vmethod_frame.h" +#include "pa_vamqp.h" #include "pa_request.h" -#include "pa_int.h" -#include "pa_vstring.h" -#include "pa_vhash.h" #include "pa_varray.h" #include "pa_vbool.h" -#include "pa_vvoid.h" -#include "pa_vdouble.h" -#include "pa_vamqp.h" #include "pa_os.h" -#include "pa_globals.h" - -#ifdef WITH_AMQP -#include -#include -#include -#include -#include -#include -#endif -#ifdef _MSC_VER -#pragma comment(lib, "rabbitmq.4.lib") -#endif - -volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.20 2026/09/18 17:10:15 moko Exp $" IDENT_PA_VAMQP_H; +volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.29 2026/10/09 22:05:41 moko Exp $" IDENT_PA_VAMQP_H; class MAmqp: public Methoded { public: // VStateless_class - Value* create_new_value(Pool&) { return new VAmqp(); } + Value* create_new_value(Pool& apool) { return new VAmqp(apool); } public: MAmqp(); }; @@ -93,6 +76,15 @@ static void check(VAmqp& self, amqp_rpc_ throw Exception("amqp", 0, "%sfailed", detail); } +// destroys the new connection if connecting fails +class Temp_amqp_connection { + amqp_connection_state_t fconnection; +public: + Temp_amqp_connection(amqp_connection_state_t aconnection): fconnection(aconnection) {} + ~Temp_amqp_connection() { if(fconnection) amqp_destroy_connection(fconnection); } + amqp_connection_state_t release() { amqp_connection_state_t result=fconnection; fconnection=0; return result; } +}; + static void amqp_connect(VAmqp& self, Request& r, HashStringValue* options) { const char* host_c = "localhost"; int port = 5672; @@ -100,13 +92,15 @@ static void amqp_connect(VAmqp& self, Re 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; - int reconnect_interval_sec = 0; +#endif if(options){ for(HashStringValue::Iterator i(*options); i; i.next()){ @@ -125,10 +119,11 @@ static void amqp_connect(VAmqp& self, Re } else if(key=="locale"){ locale_c=value->as_string().cstr(); } else if(key=="heartbeat"){ - heartbeat=r.process(*value).as_int(); + 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()){ @@ -146,14 +141,19 @@ static void amqp_connect(VAmqp& self, Re 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(); + Temp_amqp_connection temp_connection(conn); amqp_socket_t* socket = 0; +#ifdef WITH_AMQP_SSL if(tls_specified) { socket = amqp_ssl_socket_new(conn); if(!socket) @@ -176,7 +176,9 @@ static void amqp_connect(VAmqp& self, Re 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"); @@ -184,25 +186,22 @@ static void amqp_connect(VAmqp& self, Re 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); + 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) 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(self, ropen, "open channel "); } if(self.fconnection) amqp_destroy_connection(self.fconnection); - self.fconnection = conn; + self.fconnection = temp_connection.release(); self.fchannel = channel; self.fstate = VAmqp::ALIVE; self.freconnect_interval = reconnect_interval_sec; @@ -451,16 +450,7 @@ static void _publish(Request& r, MethodP } static void _release(Request& r, MethodParams&) { - VAmqp& self=GET_SELF(r, VAmqp); - if(self.fconnection){ - amqp_connection_state_t conn=self.fconnection; - amqp_channel_close(conn, self.fchannel, AMQP_REPLY_SUCCESS); - amqp_connection_close(conn, AMQP_REPLY_SUCCESS); - amqp_destroy_connection(conn); - self.fconnection=0; - self.fchannel=0; - self.fstate=VAmqp::CONNECTION_DEAD; - } + GET_SELF(r, VAmqp).release(); } // to allow both [] and () call syntax @@ -775,7 +765,7 @@ static struct timeval* amqp_wait_tv(stru // 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, int channel, amqp_envelope_t* envelope, struct timeval* tv) { +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)