Add --watch to mosquitto_sub.

ChangeLog.txt

@@ -55,6 +55,9 @@ Clients:
   that instance. This is useful for negating TLS options provided in a config
   file, or to disable the automatic use of TLS when using port 8883.
   Closes #2180.
+- Add `--watch` to mosquitto_sub which means messages will be printed on a
+  fixed line number based on the topic and order in which messages were
+  received. Requires ANSI escape code support in the terminal.
 
 
 2.0.10 - 2021-04-03

client/client_shared.c

@@ -1198,6 +1198,11 @@ int client_config_line_proc(struct mosq_config *cfg, int pub_or_sub, int argc, c
 				}
 				i++;
 			}
+		}else if(!strcmp(argv[i], "--watch")){
+			if(pub_or_sub != CLIENT_SUB){
+				goto unknown_option;
+			}
+			cfg->watch = true;
 		}else if(!strcmp(argv[i], "--will-payload")){
 			if(i==argc-1){
 				fprintf(stderr, "Error: --will-payload argument given but no will payload specified.\n\n");

client/client_shared.h

@@ -115,6 +115,7 @@ struct mosq_config {
 	int sub_opts; /* sub */
 	long session_expiry_interval;
 	int random_filter; /* sub */
+	bool watch; /* sub */
 #ifdef WITH_SOCKS
 	char *socks5_host;
 	int socks5_port;

client/sub_client.c

@@ -205,7 +205,7 @@ static void print_usage(void)
 	printf("                     [-A bind_address] [--nodelay]\n");
 #endif
 	printf("                     [-i id] [-I id_prefix]\n");
-	printf("                     [-d] [-N] [--quiet] [-v]\n");
+	printf("                     [-d] [-N] [--quiet] [-v] [--watch]\n");
 	printf("                     [--will-topic [--will-payload payload] [--will-qos qos] [--will-retain]]\n");
 #ifdef WITH_TLS
 	printf("                     [--no-tls]\n");
@@ -282,6 +282,9 @@ static void print_usage(void)
 	printf("                     Use -T to filter out messages you do not want to be cleared.\n");
 	printf(" --unix : connect to a broker through a unix domain socket instead of a TCP socket,\n");
 	printf("          e.g. /tmp/mosquitto.sock\n");
+	printf(" --watch : messages will be printed on a fixed line number based on the topic and order in\n");
+	printf("           which topics are received. Useful for monitoring multiple topics that have\n");
+	printf("           single line payloads.\n");
 	printf(" --will-payload : payload for the client Will, which is sent by the broker in case of\n");
 	printf("                  unexpected disconnection. If not given and will-topic is set, a zero\n");
 	printf("                  length message will be sent.\n");
@@ -328,8 +331,6 @@ int main(int argc, char *argv[])
 
 	mosquitto_lib_init();
 
-	rand_init();
-
 	rc = client_config_load(&cfg, CLIENT_SUB, argc, argv);
 	if(rc){
 		if(rc == 2){
@@ -375,6 +376,8 @@ int main(int argc, char *argv[])
 	mosquitto_connect_v5_callback_set(g_mosq, my_connect_callback);
 	mosquitto_message_v5_callback_set(g_mosq, my_message_callback);
 
+	output_init(&cfg);
+
 	rc = client_connect(g_mosq, &cfg);
 	if(rc){
 		goto cleanup;

client/sub_client_output.c

@@ -37,6 +37,10 @@ Contributors:
 #define snprintf sprintf_s
 #endif
 
+#undef uthash_malloc
+#undef uthash_free
+#include <uthash.h>
+
 #ifdef WITH_CJSON
 #  include <cjson/cJSON.h>
 #endif
@@ -52,6 +56,14 @@ Contributors:
 
 extern struct mosq_config cfg;
 
+struct watch_topic{
+	UT_hash_handle hh;
+	char *topic;
+	int line;
+};
+static int watch_max = 2;
+static struct watch_topic *watch_items = NULL;
+
 static int get_time(struct tm **ti, long *ns)
 {
 #ifdef WIN32
@@ -764,7 +776,7 @@ static void formatted_print(const struct mosq_config *lcfg, const struct mosquit
 }
 
 
-void rand_init(void)
+static void rand_init(void)
 {
 #ifndef WIN32
 	struct tm *ti = NULL;
@@ -777,6 +789,28 @@ void rand_init(void)
 }
 
 
+void watch_print(const struct mosquitto_message *message)
+{
+	struct watch_topic *item = NULL;
+
+	HASH_FIND(hh, watch_items, message->topic, strlen(message->topic), item);
+	if(item == NULL){
+		item = calloc(1, sizeof(struct watch_topic));
+		if(item == NULL){
+			return;
+		}
+		item->line = watch_max++;
+		item->topic = strdup(message->topic);
+		if(item->topic == NULL){
+			free(item);
+			return;
+		}
+		HASH_ADD_KEYPTR(hh, watch_items, item->topic, strlen(item->topic), item);
+	}
+	printf("\e[%d;1H", item->line);
+}
+
+
 void print_message(struct mosq_config *lcfg, const struct mosquitto_message *message, const mosquitto_property *properties)
 {
 #ifdef WIN32
@@ -785,6 +819,9 @@ void print_message(struct mosq_config *lcfg, const struct mosquitto_message *mes
 	long r = 0;
 #endif
 
+	if(lcfg->watch){
+		watch_print(message);
+	}
 	if(lcfg->random_filter < 10000){
 #ifdef WIN32
 		rand_s(&r);
@@ -819,5 +856,16 @@ void print_message(struct mosq_config *lcfg, const struct mosquitto_message *mes
 			fflush(stdout);
 		}
 	}
+	if(lcfg->watch){
+		printf("\e[%d;1H\n", watch_max-1);
+	}
 }
 
+void output_init(struct mosq_config *lcfg)
+{
+	rand_init();
+	if(lcfg->watch){
+		printf("\e[2J\e[1;1H");
+		printf("Broker: %s\n", lcfg->host);
+	}
+}

client/sub_client_output.h

@@ -22,7 +22,7 @@ Contributors:
 #include "mosquitto.h"
 #include "client_shared.h"
 
-void rand_init(void);
+void output_init(struct mosq_config *cfg);
 void print_message(struct mosq_config *cfg, const struct mosquitto_message *message, const mosquitto_property *properties);
 
 #endif

man/mosquitto_sub.1.xml

@@ -60,6 +60,7 @@
 			<arg><option>-x</option> <replaceable>session-expiry-interval</replaceable></arg>
 			<arg><option>--proxy</option> <replaceable>socks-url</replaceable></arg>
 			<arg><option>--quiet</option></arg>
+			<arg><option>--watch</option></arg>
 			<arg>
 				<option>--will-topic</option> <replaceable>topic</replaceable>
 				<arg><option>--will-payload</option> <replaceable>payload</replaceable></arg>
@@ -801,6 +802,24 @@ mosquitto_sub -t 'bbc/#' -T bbc/bbc1 --remove-retained</programlisting>
 						connected to the broker.</para>
 				</listitem>
 			</varlistentry>
+			<varlistentry>
+				<term><option>--watch</option></term>
+				<listitem>
+                    <para>
+                        Messages will be printed on a fixed line number based on
+                        the topic and order in which topics are received. Useful
+                        for monitoring multiple topics that have single line
+                        payloads. Unexpected behaviour will occur if there are
+                        more topics than lines in the terminal, or if the
+                        payload occupies more than a single line.
+						This can be used in conjuction with other output options
+						e.g. <option>-F</option>.
+					</para>
+					<para>
+						Requires ANSI escape code support in the terminal.
+					</para>
+				</listitem>
+			</varlistentry>
 			<varlistentry>
 				<term><option>--will-payload</option></term>
 				<listitem>