event_rabbitmq: restore event functionality (#3738)

* event_rabbitmq: restore event functionality

* event_rabbitmq: fix server reconnect

Author: ovidiusas

modules/event_rabbitmq/event_rabbitmq.c

@@ -398,6 +398,7 @@ static evi_reply_sock* rmq_parse(str socket)
 				st = ST_HOST;
 				if (dupl_string(&tmp, begin, socket.s + i)) goto err;
 				param->conn.uri.user = tmp.s;
+				//param->conn.uri.user[tmp.len] = '\0';
 				begin = socket.s + i + 1;
 				param->conn.flags |= RMQ_PARAM_USER;
 				break;
@@ -566,6 +567,7 @@ static evi_reply_sock* rmq_parse(str socket)
 		goto err;
 	}
 
+	param->conn.state = RMQS_PREINIT;
 	param->conn.heartbeat = heartbeat;
 	sock->params = param;
 	sock->flags |= EVI_PARAMS | RMQ_FLAG | EVI_ASYNC_STATUS;

modules/event_rabbitmq/event_rabbitmq.h

@@ -79,10 +79,10 @@ extern struct timeval conn_timeout_tv;
 extern struct timeval rpc_timeout_tv;
 
 int rmq_error(char const *context, amqp_rpc_reply_t x);
-void rmq_destroy_connection(rmq_connection_t *conn);
-int rmq_reconnect(rmq_connection_t *conn, int max_frames, str cid);
+void rmq_destroy_connection(rmq_connection_t *conn, int temporarely);
+int rmq_server_reconnect(rmq_connection_t *conn, int max_frames, str cid);
 int amqp_check_status(rmq_connection_t *conn, int r, int* retry, str cid);
-int rmq_basic_publish(rmq_connection_t *conn, int max_frames,
+int rmq_basic_server_publish(rmq_connection_t *conn, int max_frames,
 							 str *cid, amqp_bytes_t akey, amqp_bytes_t abody,
 							 amqp_basic_properties_t *props, int retries);
 

modules/event_rabbitmq/rabbitmq_send.c

@@ -151,7 +151,7 @@ void rmq_free_param(rmq_params_t *rmqp)
 		shm_free(rmqp->routing_key.s);
 }
 
-void rmq_destroy_connection(rmq_connection_t *conn)
+void rmq_destroy_connection(rmq_connection_t *conn, int temporarely)
 {
 	switch (conn->state)
 	{
@@ -165,12 +165,16 @@ void rmq_destroy_connection(rmq_connection_t *conn)
 		if (amqp_destroy_connection(conn->conn) < 0)
 			LM_ERR("cannot destroy connection\n");
 	case RMQS_OFF:
+	case RMQS_PREINIT:
 		break;
 	default:
 		LM_WARN("Unknown rmq server state %d\n", conn->state);
 	}
 
-	conn->state = RMQS_OFF;
+	if (temporarely)
+		conn->state = RMQS_PREINIT;
+	else
+		conn->state = RMQS_OFF;
 
 	if (conn->tls_dom) {
 		tls_api.release_domain(conn->tls_dom);
@@ -187,26 +191,274 @@ void rmq_destroy(evi_reply_sock *sock)
 	if ((sock->flags & EVI_PARAMS) && sock->params) {
 		rmq_free_param((rmq_params_t *)sock->params);
 		rmq_params_t *rmqp = (rmq_params_t *)sock->params;
-		rmq_destroy_connection(&rmqp->conn);
+		rmq_destroy_connection(&rmqp->conn, 0);
 	}
 	shm_free(sock);
 }
 
+static int rmq_reconnect(evi_reply_sock *sock)
+{
+	rmq_params_t * rmqp = (rmq_params_t *)sock->params;
+	rmq_connection_t *conn;
+#if defined AMQP_VERSION_v04
+	amqp_socket_t *amqp_sock;
+#endif
+	int socket;
+
+	if (!rmqp) {
+		LM_ERR("not enough socket info\n");
+		return -1;
+	}
+
+	conn = &rmqp->conn;
+
+	switch (conn->state) {
+	case RMQS_OFF:
+	case RMQS_PREINIT:
+		if (!(conn->conn = amqp_new_connection())) {
+			LM_ERR("cannot create amqp connection!\n");
+			return -1;
+		}
+#if defined AMQP_VERSION_v04
+		if (use_tls && (conn->uri.ssl || (conn->flags&RMQ_PARAM_TLS))) {
+			if (!conn->tls_dom) {
+				conn->tls_dom = tls_api.find_client_domain_name(&conn->tls_dom_name);
+				if (!conn->tls_dom) {
+					LM_ERR("TLS domain: %.*s not found\n",
+						conn->tls_dom_name.len, conn->tls_dom_name.s);
+					return -1;
+				}
+			}
+
+			amqp_sock = amqp_ssl_socket_new(conn->conn);
+			if (!amqp_sock) {
+				LM_ERR("cannot create AMQP TLS socket\n");
+				return -1;
+			}
+
+			#if AMQP_VERSION < AMQP_VERSION_CODE(0, 10, 0, 0)
+			/* if amqp_ssl_socket_get_context() is not available, serialize the CA,
+			 * cert and key loading in order to prevent openssl multiprocess issues */
+			lock_get(ssl_lock);
+			if (amqp_ssl_socket_set_cacert(amqp_sock, conn->tls_dom->ca.s) !=
+				AMQP_STATUS_OK) {
+				LM_ERR("Failed to set CA certificate\n");
+				lock_release(ssl_lock);
+				return -1;
+			}
+
+			if (amqp_ssl_socket_set_key(amqp_sock, conn->tls_dom->cert.s,
+				conn->tls_dom->pkey.s) != AMQP_STATUS_OK) {
+				LM_ERR("Failed to set certificate and private key\n");
+				lock_release(ssl_lock);
+				return -1;
+			}
+			lock_release(ssl_lock);
+			#else
+			/* point the CA, cert and key from librabbitmq's SSL_CTX to
+			 * the info loaded through the tls_mgm's SSL_CTX, in order to
+			 * prevent openssl multiprocess issues */
+			void *ssl_ctx;
+			ssl_ctx = amqp_ssl_socket_get_context(amqp_sock);
+
+			/* set CA in AMQP's SSL_CTX  */
+			openssl_api.ctx_set_cert_store(ssl_ctx,
+				((void**)conn->tls_dom->ctx)[process_no]);
+
+			/* set certificate in AMQP's SSL_CTX */
+			if (openssl_api.ctx_set_cert_chain(ssl_ctx,
+				((void**)conn->tls_dom->ctx)[process_no]) < 0) {
+				LM_ERR("Failed to set certificate\n");
+				return -1;
+			}
+
+			/* set private key in AMQP's SSL_CTX */
+			if (openssl_api.ctx_set_pkey_file(ssl_ctx, conn->tls_dom->pkey.s) < 0) {
+				LM_ERR("Failed to set private key\n");
+				return -1;
+			}
+			#endif
+
+			#if AMQP_VERSION >= AMQP_VERSION_CODE(0, 8, 0, 0)
+			amqp_ssl_socket_set_verify_peer(amqp_sock, conn->tls_dom->verify_cert);
+			amqp_ssl_socket_set_verify_hostname(amqp_sock, 0);
+			#else
+			amqp_ssl_socket_set_verify(amqp_sock, conn->tls_dom->verify_cert);
+			#endif
+
+			#if AMQP_VERSION >= AMQP_VERSION_CODE(0, 8, 0, 0)
+			amqp_tls_version_t method_min, method_max;
+
+			if (conn->tls_dom->method != TLS_METHOD_UNSPEC) {
+				switch (conn->tls_dom->method) {
+				case TLS_USE_TLSv1:
+					method_min = AMQP_TLSv1;
+					break;
+				case TLS_USE_TLSv1_2:
+					method_min = AMQP_TLSv1_2;
+					break;
+				default:
+					LM_NOTICE("Unsupported TLS minimum method for AMQP, using TLSv1\n");
+					method_min = AMQP_TLSv1;
+				}
+			} else {
+				LM_INFO("Minimum TLS method unspecified, using TLSv1\n");
+				method_min = AMQP_TLSv1;
+			}
+
+			if (conn->tls_dom->method_max != TLS_METHOD_UNSPEC) {
+				switch (conn->tls_dom->method_max) {
+				case TLS_USE_TLSv1:
+					method_max = AMQP_TLSv1;
+					break;
+				case TLS_USE_TLSv1_2:
+					method_max = AMQP_TLSv1_2;
+					break;
+				default:
+					LM_NOTICE("Unsupported TLS maximum method for AMQP, using latest"
+						" supported by librabbitmq\n");
+					method_max = AMQP_TLSvLATEST;
+				}
+			} else {
+				method_max = AMQP_TLSvLATEST;
+				LM_INFO("Maximum TLS method unspecified, using latest supported by"
+					" librabbitmq\n");
+			}
+
+			if (amqp_ssl_socket_set_ssl_versions(amqp_sock, method_min, method_max) !=
+				AMQP_STATUS_OK) {
+				LM_ERR("Failed to set TLS method range\n");
+				return -1;
+			}
+			#endif
+		} else {
+			amqp_sock = amqp_tcp_socket_new(conn->conn);
+			if (!amqp_sock) {
+				LM_ERR("cannot create AMQP socket\n");
+				return -1;
+			}
+		}
+
+		socket = amqp_socket_open_noblock(amqp_sock, sock->address.s,
+			sock->port, &conn_timeout_tv);
+		if (socket < 0) {
+			amqp_connection_close(conn->conn, AMQP_REPLY_SUCCESS);
+			LM_ERR("cannot open AMQP socket\n");
+			return -1;
+		}
+#if defined AMQP_VERSION && AMQP_VERSION >= 0x00090000
+		if (rpc_timeout_tv.tv_sec > 0 &&
+				amqp_set_rpc_timeout(conn->conn, &rpc_timeout_tv) < 0)
+			LM_ERR("setting RPC timeout - going blocking\n");
+#endif
+
+#else
+		socket = amqp_open_socket_noblock(sock->address.s, sock->port,
+				&conn_timeout_tv);
+		if (socket < 0) {
+			LM_ERR("cannot open AMQP socket\n");
+			return -1;
+		}
+		amqp_set_sockfd(conn->conn, socket);
+#endif
+		conn->state = RMQS_INIT;
+		/* fall through */
+	case RMQS_INIT:
+		if (rmq_error("Logging in", amqp_login(
+				conn->conn,
+				RMQ_DEFAULT_VHOST,
+				0,
+				RMQ_DEFAULT_MAX,
+				conn->heartbeat,
+				AMQP_SASL_METHOD_PLAIN,
+				conn->uri.user ? conn->uri.user : RMQ_DEFAULT_UP,
+				conn->uri.password ? conn->uri.password : RMQ_DEFAULT_UP)))
+			return -2;
+		/* all good - return success */
+		conn->state = RMQS_CONN;
+		/* fall through */
+	case RMQS_CONN:
+		/* don't use more than 1 channel */
+		amqp_channel_open(conn->conn, RMQ_DEFAULT_CHANNEL);
+		if (rmq_error("Opening channel", amqp_get_rpc_reply(conn->conn)))
+			return -2;
+		LM_INFO("[] successfully connected!\n");
+		conn->state = RMQS_ON;
+		/* fall through */
+	case RMQS_ON:
+		return 0;
+	default:
+		LM_WARN("Unknown rmq server state %d\n", conn->state);
+		return -1;
+	}
+
+}
+
+static int rmq_basic_publish(rmq_connection_t *conn, int max_frames,
+						str *cid, amqp_bytes_t akey, amqp_bytes_t abody,
+						amqp_basic_properties_t *props, int retries, rmq_send_t *rmqs)
+{
+	int ret;
+	evi_reply_sock *sock;
+
+	if (conn->flags & RMQF_NOPER) {
+		props->delivery_mode = 2;
+		props->_flags |= AMQP_BASIC_DELIVERY_MODE_FLAG;
+	}
+
+	do {
+		sock = rmqs->sock;
+		LM_INFO("rmq_reconnect()\n");
+		ret = rmq_reconnect(sock);
+
+		if (ret == -1) {
+			if (amqp_destroy_connection(conn->conn) < 0)
+				LM_ERR("cannot destroy connection\n");
+			if (conn->tls_dom) {
+				tls_api.release_domain(conn->tls_dom);
+				conn->tls_dom = NULL;
+			}
+			LM_ERR("cannot connect to RabbitMQ server %s:%u\n",
+				conn->uri.host, conn->uri.port);
+			return ret;
+		}
+		if (ret == -2) {
+			rmq_destroy_connection(conn, 1);
+			LM_ERR("cannot connect to RabbitMQ server %s:%u\n",
+				conn->uri.host, conn->uri.port);
+				return ret;
+		}
+
+		ret = amqp_basic_publish(conn->conn, RMQ_DEFAULT_CHANNEL, conn->exchange, akey, \
+				(conn->flags & RMQF_MAND), (conn->flags & RMQF_IMM),
+				props, abody);
+		ret = amqp_check_status(conn, ret, &retries, *cid);
+	} while (ret > 0);
+
+	return ret;
+}
+
 /* sends the buffer */
 static int rmq_sendmsg(rmq_send_t *rmqs)
 {
 	rmq_params_t * rmqp = (rmq_params_t *)rmqs->sock->params;
 	int ret;
-	int re_publish = 0;
+	int re_publish = 2;
 	amqp_basic_properties_t props;
 
 	if (!rmqp || !(rmqp->conn.flags & RMQF_MAND)) {
 		LM_ERR("not enough socket info\n");
 		return -1;;
 	}
 
-	if (rmqp->conn.state == RMQS_OFF)
+	/* FIXME:
+	 * We need a new state for un-initialised connections
+	 * Unlike server connections, this ones are not initialised at startup
+	 */
+	if (rmqp->conn.state == RMQS_OFF) {
+		LM_INFO("server disconnected\n");
 		return 0;
+	}
 
 	rmqp->conn.uri.host = rmqs->sock->address.s;
 
@@ -218,7 +470,8 @@ static int rmq_sendmsg(rmq_send_t *rmqs)
 			amqp_cstring_bytes(rmqp->routing_key.s),
 			amqp_cstring_bytes(rmqs->msg),
 			((rmqp->conn.flags & RMQF_NOPER)?&props:0),
-			re_publish);
+			re_publish,
+			rmqs);
 
 	return ret;
 }