--- parser3/src/classes/amqp.C	2026/10/08 18:56:00	1.26
+++ parser3/src/classes/amqp.C	2026/10/16 18:59:50	1.31
@@ -15,23 +15,11 @@
 #include "pa_vbool.h"
 #include "pa_os.h"
 
-#ifdef WITH_AMQP
-
-#include <amqp.h>
-#include <amqp_tcp_socket.h>
-#include <amqp_framing.h>
-
-#ifdef WITH_AMQP_SSL
-#include <amqp_ssl_socket.h>
-#endif
-
-#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;
+volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.31 2026/10/16 18:59:50 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();
 };
@@ -88,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;
@@ -153,6 +150,7 @@ static void amqp_connect(VAmqp& self, Re
 	}
 
 	amqp_connection_state_t conn = amqp_new_connection();
+	Temp_amqp_connection temp_connection(conn);
 	amqp_socket_t* socket = 0;
 
 #ifdef WITH_AMQP_SSL
@@ -189,37 +187,34 @@ 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_sec, AMQP_SASL_METHOD_PLAIN, user_c, pass_c);
-	if(rlogin.reply_type != AMQP_RESPONSE_NORMAL){
-		amqp_destroy_connection(conn);
+	if(rlogin.reply_type != AMQP_RESPONSE_NORMAL)
 		check(self, rlogin, "login ");
-	}
 
 	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);
+	if(ropen.reply_type != AMQP_RESPONSE_NORMAL) // no amqp_connection_close: its reply could overwrite the error text in ropen
 		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;
 }
 
-// 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.
+// 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.
+// The new channel gets a new number: the deliveries the broker has sent on the dead one may still be in the library's buffers.
 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);
+	amqp_channel_t channel = self.fchannel < amqp_get_channel_max(self.fconnection) ? self.fchannel+1 : 1;
+	amqp_channel_open(self.fconnection, channel);
 	check(self, amqp_get_rpc_reply(self.fconnection), "channel repair ");
+	self.fchannel = channel;
 	self.fstate = VAmqp::ALIVE;
 }
 
@@ -455,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
@@ -782,6 +768,12 @@ static struct timeval* amqp_wait_tv(stru
 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_NORMAL && envelope->channel != channel) {
+			// Only the deliveries from the given channel are returned: the ones that arrived
+			// on a channel closed by the broker are skipped, the broker will deliver them again.
+			amqp_destroy_envelope(envelope);
+			continue;
+		}
 		if(res.reply_type != AMQP_RESPONSE_LIBRARY_EXCEPTION || res.library_error != AMQP_STATUS_UNEXPECTED_STATE)
 			return res;