stdoutsub.c

00001 /*******************************************************************************
00002  * Copyright (c) 2012, 2016 IBM Corp.
00003  *
00004  * All rights reserved. This program and the accompanying materials
00005  * are made available under the terms of the Eclipse Public License v1.0
00006  * and Eclipse Distribution License v1.0 which accompany this distribution. 
00007  *
00008  * The Eclipse Public License is available at 
00009  *   http://www.eclipse.org/legal/epl-v10.html
00010  * and the Eclipse Distribution License is available at 
00011  *   http://www.eclipse.org/org/documents/edl-v10.php.
00012  *
00013  * Contributors:
00014  *    Ian Craggs - initial contribution
00015  *    Ian Craggs - change delimiter option from char to string
00016  *    Al Stockdill-Mander - Version using the embedded C client
00017  *    Ian Craggs - update MQTTClient function names
00018  *******************************************************************************/
00019 
00020 /*
00021  
00022  stdout subscriber
00023  
00024  compulsory parameters:
00025  
00026   topic to subscribe to
00027  
00028  defaulted parameters:
00029  
00030     --host localhost
00031     --port 1883
00032     --qos 2
00033     --delimiter \n
00034     --clientid stdout_subscriber
00035     
00036     --userid none
00037     --password none
00038 
00039  for example:
00040 
00041     stdoutsub topic/of/interest --host iot.eclipse.org
00042 
00043 */
00044 #include <stdio.h>
00045 #include <memory.h>
00046 #include "MQTTClient.h"
00047 
00048 #include <stdio.h>
00049 #include <signal.h>
00050 
00051 #include <sys/time.h>
00052 
00053 
00054 volatile int toStop = 0;
00055 
00056 
00057 void usage()
00058 {
00059     printf("MQTT stdout subscriber\n");
00060     printf("Usage: stdoutsub topicname <options>, where options are:\n");
00061     printf("  --host <hostname> (default is localhost)\n");
00062     printf("  --port <port> (default is 1883)\n");
00063     printf("  --qos <qos> (default is 2)\n");
00064     printf("  --delimiter <delim> (default is \\n)\n");
00065     printf("  --clientid <clientid> (default is hostname+timestamp)\n");
00066     printf("  --username none\n");
00067     printf("  --password none\n");
00068     printf("  --showtopics <on or off> (default is on if the topic has a wildcard, else off)\n");
00069     exit(-1);
00070 }
00071 
00072 
00073 void cfinish(int sig)
00074 {
00075     signal(SIGINT, NULL);
00076     toStop = 1;
00077 }
00078 
00079 
00080 struct opts_struct
00081 {
00082     char* clientid;
00083     int nodelimiter;
00084     char* delimiter;
00085     enum QoS qos;
00086     char* username;
00087     char* password;
00088     char* host;
00089     int port;
00090     int showtopics;
00091 } opts =
00092 {
00093     (char*)"stdout-subscriber", 0, (char*)"\n", QOS2, NULL, NULL, (char*)"localhost", 1883, 0
00094 };
00095 
00096 
00097 void getopts(int argc, char** argv)
00098 {
00099     int count = 2;
00100     
00101     while (count < argc)
00102     {
00103         if (strcmp(argv[count], "--qos") == 0)
00104         {
00105             if (++count < argc)
00106             {
00107                 if (strcmp(argv[count], "0") == 0)
00108                     opts.qos = QOS0;
00109                 else if (strcmp(argv[count], "1") == 0)
00110                     opts.qos = QOS1;
00111                 else if (strcmp(argv[count], "2") == 0)
00112                     opts.qos = QOS2;
00113                 else
00114                     usage();
00115             }
00116             else
00117                 usage();
00118         }
00119         else if (strcmp(argv[count], "--host") == 0)
00120         {
00121             if (++count < argc)
00122                 opts.host = argv[count];
00123             else
00124                 usage();
00125         }
00126         else if (strcmp(argv[count], "--port") == 0)
00127         {
00128             if (++count < argc)
00129                 opts.port = atoi(argv[count]);
00130             else
00131                 usage();
00132         }
00133         else if (strcmp(argv[count], "--clientid") == 0)
00134         {
00135             if (++count < argc)
00136                 opts.clientid = argv[count];
00137             else
00138                 usage();
00139         }
00140         else if (strcmp(argv[count], "--username") == 0)
00141         {
00142             if (++count < argc)
00143                 opts.username = argv[count];
00144             else
00145                 usage();
00146         }
00147         else if (strcmp(argv[count], "--password") == 0)
00148         {
00149             if (++count < argc)
00150                 opts.password = argv[count];
00151             else
00152                 usage();
00153         }
00154         else if (strcmp(argv[count], "--delimiter") == 0)
00155         {
00156             if (++count < argc)
00157                 opts.delimiter = argv[count];
00158             else
00159                 opts.nodelimiter = 1;
00160         }
00161         else if (strcmp(argv[count], "--showtopics") == 0)
00162         {
00163             if (++count < argc)
00164             {
00165                 if (strcmp(argv[count], "on") == 0)
00166                     opts.showtopics = 1;
00167                 else if (strcmp(argv[count], "off") == 0)
00168                     opts.showtopics = 0;
00169                 else
00170                     usage();
00171             }
00172             else
00173                 usage();
00174         }
00175         count++;
00176     }
00177     
00178 }
00179 
00180 
00181 void messageArrived(MessageData* md)
00182 {
00183     MQTTMessage* message = md->message;
00184 
00185     if (opts.showtopics)
00186         printf("%.*s\t", md->topicName->lenstring.len, md->topicName->lenstring.data);
00187     if (opts.nodelimiter)
00188         printf("%.*s", (int)message->payloadlen, (char*)message->payload);
00189     else
00190         printf("%.*s%s", (int)message->payloadlen, (char*)message->payload, opts.delimiter);
00191     //fflush(stdout);
00192 }
00193 
00194 
00195 int main(int argc, char** argv)
00196 {
00197     int rc = 0;
00198     unsigned char buf[100];
00199     unsigned char readbuf[100];
00200     
00201     if (argc < 2)
00202         usage();
00203     
00204     char* topic = argv[1];
00205 
00206     if (strchr(topic, '#') || strchr(topic, '+'))
00207         opts.showtopics = 1;
00208     if (opts.showtopics)
00209         printf("topic is %s\n", topic);
00210 
00211     getopts(argc, argv);    
00212 
00213     Network n;
00214     MQTTClient c;
00215 
00216     signal(SIGINT, cfinish);
00217     signal(SIGTERM, cfinish);
00218 
00219     NetworkInit(&n);
00220     NetworkConnect(&n, opts.host, opts.port);
00221     MQTTClientInit(&c, &n, 1000, buf, 100, readbuf, 100);
00222  
00223     MQTTPacket_connectData data = MQTTPacket_connectData_initializer;       
00224     data.willFlag = 0;
00225     data.MQTTVersion = 3;
00226     data.clientID.cstring = opts.clientid;
00227     data.username.cstring = opts.username;
00228     data.password.cstring = opts.password;
00229 
00230     data.keepAliveInterval = 10;
00231     data.cleansession = 1;
00232     printf("Connecting to %s %d\n", opts.host, opts.port);
00233     
00234     rc = MQTTConnect(&c, &data);
00235     printf("Connected %d\n", rc);
00236     
00237     printf("Subscribing to %s\n", topic);
00238     rc = MQTTSubscribe(&c, topic, opts.qos, messageArrived);
00239     printf("Subscribed %d\n", rc);
00240 
00241     while (!toStop)
00242     {
00243         MQTTYield(&c, 1000);    
00244     }
00245     
00246     printf("Stopping\n");
00247 
00248     MQTTDisconnect(&c);
00249     NetworkDisconnect(&n);
00250 
00251     return 0;
00252 }
00253 
00254