Options to block startup unless RabbitMQ Transport / Event Handler are connected

conf/janus.eventhandler.rabbitmqevh.jcfg.sample

@@ -21,6 +21,7 @@ general: {
 	route_key = "janus-events"		# Routing key to use when publishing messages
 	#exchange_type = "fanout" 		# Rabbitmq exchange_type can be one of the available types: direct, topic, headers and fanout (fanout by defualt).
 	#heartbeat = 60 				# Defines the seconds without communication that should pass before considering the TCP connection unreachable.
+	#block_startup_until_connected = true # Whether to block the server from starting until a connection to the RabbitMQ server is established.
 	#declare_outgoing_queue = true # By default (for backwards compatibility), we declare an outgoing queue. Set this to false to disable that behavior
 
 	#ssl_enable = false				# Whether ssl support must be enabled

conf/janus.transport.rabbitmq.jcfg.sample

@@ -35,6 +35,7 @@ general: {
 	#queue_autodelete = false			# Whether or not incoming queue should autodelete after janus disconnects from RabbitMQ
 	#queue_exclusive = false			# Whether or not incoming queue should only allow one subscriber
 	#heartbeat = 60 				# Defines the seconds without communication that should pass before considering the TCP connection unreachable.
+	#block_startup_until_connected = true # Whether to block the server from starting until a connection to the RabbitMQ server is established.
 
 	#ssl_enabled = false				# Whether ssl support must be enabled
 	#ssl_verify_peer = true				# Whether peer verification must be enabled

src/events/janus_rabbitmqevh.c

@@ -87,6 +87,7 @@ static GThread *handler_thread;
 static GThread *in_thread;
 static void *jns_rmqevh_hdlr(void *data);
 static void *jns_rmqevh_hrtbt(void *data);
+int janus_rabbitmqevh_tryconnect(void);
 int janus_rabbitmqevh_connect(void);
 
 /* Queue of events to handle */
@@ -121,6 +122,7 @@ static gboolean ssl_verify_peer = FALSE;
 static gboolean ssl_verify_hostname = FALSE;
 static char *route_key = NULL, *exchange = NULL, *exchange_type = NULL ;
 static uint16_t heartbeat = 0;
+static gboolean block_startup_until_connected = FALSE;
 static uint16_t rmqport = AMQP_PROTOCOL_PORT;
 static gboolean declare_outgoing_queue = TRUE;
 
@@ -236,6 +238,10 @@ int janus_rabbitmqevh_init(const char *config_path) {
 		heartbeat = 0;
 	}
 
+	item = janus_config_get(config, config_general, janus_config_type_item, "block_startup_until_connected");
+	if(item && item->value && janus_is_true(item->value))
+		block_startup_until_connected = TRUE;
+
 	/* SSL config*/
 	item = janus_config_get(config, config_general, janus_config_type_item, "ssl_enable");
 	if(!item || !item->value || !janus_is_true(item->value)) {
@@ -346,6 +352,26 @@ int janus_rabbitmqevh_init(const char *config_path) {
 }
 
 int janus_rabbitmqevh_connect(void) {
+	uint8_t retry_interval = 5; // Retry interval in seconds
+
+	while (!g_atomic_int_get(&stopping)) {
+		if (janus_rabbitmqevh_tryconnect() == 0) {
+			return 0;
+		}
+
+		if (!block_startup_until_connected) {
+			break;
+		}
+
+		JANUS_LOG(LOG_ERR, "RabbitMQEventHandler: Connection failed, retrying in %d seconds\n", retry_interval);
+		g_usleep(retry_interval * 1000000);
+	}
+
+	JANUS_LOG(LOG_FATAL, "Failed to connect to RabbitMQ\n");
+	return -1;
+}
+
+int janus_rabbitmqevh_tryconnect(void){
 	rmq_conn = amqp_new_connection();
 	amqp_socket_t *socket = NULL;
 	int status = AMQP_STATUS_OK;
@@ -665,7 +691,7 @@ static void *jns_rmqevh_hrtbt(void *data) {
 			}
 			if(!g_atomic_int_get(&stopping)) {
 				JANUS_LOG(LOG_VERB, "RabbitMQEventHandler: Trying to reconnect\n");
-				int result = janus_rabbitmqevh_connect();
+				int result = janus_rabbitmqevh_tryconnect();
 				if(result < 0) {
 					g_usleep(5000000);
 				} else {

src/transports/janus_rabbitmq.c

@@ -190,6 +190,7 @@ static gboolean declare_outgoing_queue = FALSE, declare_outgoing_queue_admin = F
 amqp_boolean_t queue_durable = 0, queue_exclusive = 0, queue_autodelete = 0,
 	queue_durable_admin = 0, queue_exclusive_admin = 0, queue_autodelete_admin = 0;
 static uint16_t heartbeat = 0;
+static gboolean block_startup_until_connected = FALSE;
 
 /* Transport implementation */
 int janus_rabbitmq_init(janus_transport_callbacks *callback, const char *config_path) {
@@ -312,6 +313,11 @@ int janus_rabbitmq_init(janus_transport_callbacks *callback, const char *config_
 		heartbeat = 0;
 	}
 
+	/* Check if we need to block startup until RabbitMQ is connected */
+	item = janus_config_get(config, config_general, janus_config_type_item, "block_startup_until_connected");
+	if(item && item->value && janus_is_true(item->value))
+		block_startup_until_connected = TRUE;
+
 	/* Now check if the Janus API must be supported */
 	item = janus_config_get(config, config_general, janus_config_type_item, "enabled");
 	if(item == NULL) {
@@ -455,7 +461,22 @@ int janus_rabbitmq_init(janus_transport_callbacks *callback, const char *config_
 		rmq_client = g_malloc0(sizeof(janus_rabbitmq_client));
 
 		/* Connect */
-		int result = janus_rabbitmq_connect();
+		uint8_t retry_interval = 5; // Retry interval in seconds
+		int result = -1;
+		while(!rmq_client->destroy && !g_atomic_int_get(&stopping)) {
+			result = janus_rabbitmq_connect();
+			if (result == 0) {
+				break;
+			}
+
+			if (!block_startup_until_connected) {
+				break;
+			}
+
+			JANUS_LOG(LOG_ERR, "RabbitMQ: Connection failed, retrying in %d seconds\n", retry_interval);
+			g_usleep(retry_interval * 1000000);
+		}
+
 		if(result < 0) {
 			goto error;
 		}