GUC: DMQ Heartbeat timeout

Author: danolivo

contrib/pg_exchange/dmq.c

@@ -129,6 +129,7 @@ static shmem_startup_hook_type PreviousShmemStartupHook;
 dmq_receiver_hook_type dmq_receiver_start_hook;
 dmq_receiver_hook_type dmq_receiver_stop_hook;
 
+int dmq_heartbeat_timeout;
 void dmq_sender_main(Datum main_arg);
 
 PG_FUNCTION_INFO_V1(dmq_receiver_loop);
@@ -1138,9 +1139,9 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 		// XXX: is it enough?
 		CHECK_FOR_INTERRUPTS();
 
-		if (dmq_now() - last_message_at > 2000)
+		if (dmq_now() - last_message_at > dmq_heartbeat_timeout)
 		{
-			mtm_log(ERROR, "[DMQ] exit receiver due to heatbeat timeout");
+			mtm_log(ERROR, "[DMQ] exit receiver due to heartbeat timeout");
 		}
 
 	}

contrib/pg_exchange/dmq.h

@@ -19,6 +19,8 @@ typedef int8 DmqSenderId;
 #define DMQ_MAX_DESTINATIONS 127
 #define DMQ_MAX_RECEIVERS 100
 
+extern int dmq_heartbeat_timeout;
+
 extern void dmq_init(const char *library_name);
 
 extern DmqDestinationId dmq_destination_add(char *connstr,

contrib/pg_exchange/pg_exchange.c

@@ -80,6 +80,18 @@ _PG_init(void)
 							 NULL,
 							 NULL);
 
+	DefineCustomIntVariable("dmq_heartbeat_timeout",
+							"Max timeout between heartbeat messages",
+							NULL,
+							&dmq_heartbeat_timeout,
+							20000,
+							1, INT_MAX,
+							PGC_USERSET,
+							GUC_UNIT_MS,
+							NULL,
+							NULL,
+							NULL);
+
 	EXCHANGE_Init_methods();
 	DUMMYSCAN_Init_methods();
 	EXEC_Hooks_init();