Correct stdoutsub.c source - bug #422540

src/samples/stdoutsub.c

@@ -12,29 +12,29 @@
  *
  * Contributors:
  *    Ian Craggs - initial contribution
+ *    Ian Craggs - change delimiter option from char to string
  *******************************************************************************/
 
- /*
- stdin publisher
+/*
+ 
+ stdout subscriber
  
  compulsory parameters:
  
-  --topic topic to publish on
+  --topic topic to subscribe to
  
  defaulted parameters:
  
 	--host localhost
 	--port 1883
-	--qos 0
-	--delimiters \n
-	--clientid stdin_publisher
-	--maxdatalen 100
+	--qos 2
+	--delimiter \n
+	--clientid stdout_subscriber
 	
 	--userid none
 	--password none
  
 */
-
 #include "MQTTClient.h"
 #include "MQTTClientPersistence.h"
 
@@ -57,27 +57,26 @@ volatile int toStop = 0;
 
 void usage()
 {
-	printf("MQTT stdin publisher\n");
-	printf("Usage: stdinpub topicname <options>, where options are:\n");
+	printf("MQTT stdout subscriber\n");
+	printf("Usage: stdoutsub topicname <options>, where options are:\n");
 	printf("  --host <hostname> (default is localhost)\n");
 	printf("  --port <port> (default is 1883)\n");
-	printf("  --qos <qos> (default is 0)\n");
-	printf("  --retained (default is off)\n");
-	printf("  --delimiter <delim> (default is \\n)");
-	printf("  --clientid <clientid> (default is hostname+timestamp)");
-	printf("  --maxdatalen 100\n");
+	printf("  --qos <qos> (default is 2)\n");
+	printf("  --delimiter <delim> (default is \\n)\n");
+	printf("  --clientid <clientid> (default is hostname+timestamp)\n");
 	printf("  --username none\n");
 	printf("  --password none\n");
+  printf("  --showtopics <on or off> (default is on if the topic has a wildcard, else off)\n");
 	exit(-1);
 }
 
 
 void myconnect(MQTTClient* client, MQTTClient_connectOptions* opts)
 {
-	printf("Connecting\n");
-	if (MQTTClient_connect(*client, opts) != 0)
+	int rc = 0;
+	if ((rc = MQTTClient_connect(*client, opts)) != 0)
 	{
-		printf("Failed to connect\n");
+		printf("Failed to connect, return code %d\n", rc);
 		exit(-1);	
 	}
 }
@@ -90,59 +89,50 @@ void cfinish(int sig)
 }
 
 
-struct
+struct opts_struct
 {
 	char* clientid;
+  int nodelimiter;
 	char* delimiter;
-	int maxdatalen;
 	int qos;
-	int retained;
 	char* username;
 	char* password;
 	char* host;
 	char* port;
-  int verbose;
+  int showtopics;
 } opts =
 {
-	"publisher", "\n", 100, 0, 0, NULL, NULL, "localhost", "1883", 0
+	"stdout-subscriber", 0, "\n", 2, NULL, NULL, "localhost", "1883", 0
 };
 
 void getopts(int argc, char** argv);
 
-int messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* m)
-{
-	/* not expecting any messages */
-	return 1;
-}
-
 int main(int argc, char** argv)
 {
 	MQTTClient client;
 	MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
 	char* topic = NULL;
-	char* buffer = NULL;
 	int rc = 0;
 	char url[100];
 	
 	if (argc < 2)
 		usage();
 	
-	getopts(argc, argv);
+	topic = argv[1];
 
-	sprintf(url, "%s:%s", opts.host, opts.port);
-  if (opts.verbose)
-		printf("URL is %s\n", url);
+  if (strchr(topic, '#') || strchr(topic, '+'))
+		opts.showtopics = 1;
+  if (opts.showtopics)
+		printf("topic is %s\n", topic);
 
-	topic = argv[1];
-	printf("Using topic %s\n", topic);
+	getopts(argc, argv);	
+	sprintf(url, "%s:%s", opts.host, opts.port);
 
 	rc = MQTTClient_create(&client, url, opts.clientid, MQTTCLIENT_PERSISTENCE_NONE, NULL);
 
 	signal(SIGINT, cfinish);
 	signal(SIGTERM, cfinish);
 
-	rc = MQTTClient_setCallbacks(client, NULL, NULL, messageArrived, NULL);
-
 	conn_opts.keepAliveInterval = 10;
 	conn_opts.reliable = 0;
 	conn_opts.cleansession = 1;
@@ -151,41 +141,33 @@ int main(int argc, char** argv)
 	
 	myconnect(&client, &conn_opts);
 	
-	buffer = malloc(opts.maxdatalen);
+	rc = MQTTClient_subscribe(client, topic, opts.qos);
 
 	while (!toStop)
 	{
-		int data_len = 0;
-		int delim_len = 0;
+		char* topicName = NULL;
+		int topicLen;
+		MQTTClient_message* message = NULL;
 		
-		delim_len = strlen(opts.delimiter);
-		do
-		{
-			buffer[data_len++] = getchar();
-			if (data_len > delim_len)
+		rc = MQTTClient_receive(client, &topicName, &topicLen, &message, 1000);
+		if (message)
 		{
-			//printf("comparing %s %s\n", opts.delimiter, &buffer[data_len - delim_len]);
-			if (strncmp(opts.delimiter, &buffer[data_len - delim_len], delim_len) == 0)
-				break;
+			if (opts.showtopics)
+				printf("%s\t", topicName);
+      if (opts.nodelimiter)
+				printf("%.*s", message->payloadlen, (char*)message->payload);
+			else
+				printf("%.*s%s", message->payloadlen, (char*)message->payload, opts.delimiter);
+			fflush(stdout);
+			MQTTClient_freeMessage(&message);
+			MQTTClient_free(topicName);
 		}
-		} while (data_len < opts.maxdatalen);
-				
-		if (opts.verbose)
-				printf("Publishing data of length %d\n", data_len);
-		rc = MQTTClient_publish(client, topic, data_len, buffer, opts.qos, opts.retained, NULL);
 		if (rc != 0)
-		{
 			myconnect(&client, &conn_opts);
-			rc = MQTTClient_publish(client, topic, data_len, buffer, opts.qos, opts.retained, NULL);
-		}
-		if (opts.qos > 0)
-			MQTTClient_yield();
 	}
 	
 	printf("Stopping\n");
 
-	free(buffer);
-
 	MQTTClient_disconnect(client, 0);
 
  	MQTTClient_destroy(&client);
@@ -199,11 +181,7 @@ void getopts(int argc, char** argv)
 	
 	while (count < argc)
 	{
-		if (strcmp(argv[count], "--retained") == 0)
-			opts.retained = 1;
-		if (strcmp(argv[count], "--verbose") == 0)
-			opts.verbose = 1;
-		else if (strcmp(argv[count], "--qos") == 0)
+		if (strcmp(argv[count], "--qos") == 0)
 		{
 			if (++count < argc)
 			{
@@ -254,17 +232,24 @@ void getopts(int argc, char** argv)
 			else
 				usage();
 		}
-		else if (strcmp(argv[count], "--maxdatalen") == 0)
+		else if (strcmp(argv[count], "--delimiter") == 0)
 		{
 			if (++count < argc)
-				opts.maxdatalen = atoi(argv[count]);
+				opts.delimiter = argv[count];
 			else
-				usage();
+				opts.nodelimiter = 1;
 		}
-		else if (strcmp(argv[count], "--delimiter") == 0)
+		else if (strcmp(argv[count], "--showtopics") == 0)
 		{
 			if (++count < argc)
-				opts.delimiter = argv[count];
+			{
+				if (strcmp(argv[count], "on") == 0)
+					opts.showtopics = 1;
+				else if (strcmp(argv[count], "off") == 0)
+					opts.showtopics = 0;
+				else
+					usage();
+			}
 			else
 				usage();
 		}
@@ -272,4 +257,3 @@ void getopts(int argc, char** argv)
 	}
 	
 }
-