--- parser3/src/classes/amqp.C	2026/09/15 21:22:19	1.19
+++ parser3/src/classes/amqp.C	2026/10/09 21:10:38	1.28
@@ -5,34 +5,21 @@
 	Authors: Konstantin Morshnev <moko@design.ru>
 */
 
+
+#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 <amqp.h>
-#include <amqp_tcp_socket.h>
-#include <amqp_ssl_socket.h>
-#include <amqp_framing.h>
-#include <stdlib.h>
-#include <string.h>
-#endif
 
-volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.19 2026/09/15 21:22:19 moko Exp $" IDENT_PA_VAMQP_H;
+volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.28 2026/10/09 21:10:38 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();
 };
@@ -96,13 +83,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()){
@@ -121,10 +110,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()){
@@ -142,6 +132,9 @@ 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);
 		}
@@ -150,6 +143,7 @@ static void amqp_connect(VAmqp& self, Re
 	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)
@@ -172,7 +166,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");
@@ -180,13 +176,13 @@ 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);
+	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(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){
@@ -447,16 +443,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
@@ -771,7 +758,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)