stdoutsub.cpp

00001 /*******************************************************************************
00002  * Copyright (c) 2012, 2013 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  *******************************************************************************/
00017 
00018 /*
00019  
00020  stdout subscriber
00021  
00022  compulsory parameters:
00023  
00024   topic to subscribe to
00025  
00026  defaulted parameters:
00027  
00028     --host localhost
00029     --port 1883
00030     --qos 2
00031     --delimiter \n
00032     --clientid stdout_subscriber
00033     
00034     --userid none
00035     --password none
00036  
00037 */
00038 #include <stdio.h>
00039 #include <memory.h>
00040 #define MQTT_DEBUG 1
00041 #include "MQTTClient.h"
00042 
00043 #define DEFAULT_STACK_SIZE -1
00044 
00045 #include "linux.cpp"
00046 
00047 #include <signal.h>
00048 #include <sys/time.h>
00049 #include <stdlib.h>
00050 
00051 
00052 volatile int toStop = 0;
00053 
00054 
00055 void usage()
00056 {
00057     printf("MQTT stdout subscriber\n");
00058     printf("Usage: stdoutsub topicname <options>, where options are:\n");
00059     printf("  --host <hostname> (default is localhost)\n");
00060     printf("  --port <port> (default is 1883)\n");
00061     printf("  --qos <qos> (default is 2)\n");
00062     printf("  --delimiter <delim> (default is \\n)\n");
00063     printf("  --clientid <clientid> (default is hostname+timestamp)\n");
00064     printf("  --username none\n");
00065     printf("  --password none\n");
00066     printf("  --showtopics <on or off> (default is on if the topic has a wildcard, else off)\n");
00067     exit(-1);
00068 }
00069 
00070 
00071 void cfinish(int sig)
00072 {
00073     signal(SIGINT, NULL);
00074     toStop = 1;
00075 }
00076 
00077 
00078 struct opts_struct
00079 {
00080     char* clientid;
00081     int nodelimiter;
00082     char* delimiter;
00083     MQTT::QoS qos;
00084     char* username;
00085     char* password;
00086     char* host;
00087     int port;
00088     int showtopics;
00089 } opts =
00090 {
00091     (char*)"stdout-subscriber", 0, (char*)"\n", MQTT::QOS2, NULL, NULL, (char*)"localhost", 1883, 0
00092 };
00093 
00094 
00095 void getopts(int argc, char** argv)
00096 {
00097     int count = 2;
00098     
00099     while (count < argc)
00100     {
00101         if (strcmp(argv[count], "--qos") == 0)
00102         {
00103             if (++count < argc)
00104             {
00105                 if (strcmp(argv[count], "0") == 0)
00106                     opts.qos = MQTT::QOS0;
00107                 else if (strcmp(argv[count], "1") == 0)
00108                     opts.qos = MQTT::QOS1;
00109                 else if (strcmp(argv[count], "2") == 0)
00110                     opts.qos = MQTT::QOS2;
00111                 else
00112                     usage();
00113             }
00114             else
00115                 usage();
00116         }
00117         else if (strcmp(argv[count], "--host") == 0)
00118         {
00119             if (++count < argc)
00120                 opts.host = argv[count];
00121             else
00122                 usage();
00123         }
00124         else if (strcmp(argv[count], "--port") == 0)
00125         {
00126             if (++count < argc)
00127                 opts.port = atoi(argv[count]);
00128             else
00129                 usage();
00130         }
00131         else if (strcmp(argv[count], "--clientid") == 0)
00132         {
00133             if (++count < argc)
00134                 opts.clientid = argv[count];
00135             else
00136                 usage();
00137         }
00138         else if (strcmp(argv[count], "--username") == 0)
00139         {
00140             if (++count < argc)
00141                 opts.username = argv[count];
00142             else
00143                 usage();
00144         }
00145         else if (strcmp(argv[count], "--password") == 0)
00146         {
00147             if (++count < argc)
00148                 opts.password = argv[count];
00149             else
00150                 usage();
00151         }
00152         else if (strcmp(argv[count], "--delimiter") == 0)
00153         {
00154             if (++count < argc)
00155                 opts.delimiter = argv[count];
00156             else
00157                 opts.nodelimiter = 1;
00158         }
00159         else if (strcmp(argv[count], "--showtopics") == 0)
00160         {
00161             if (++count < argc)
00162             {
00163                 if (strcmp(argv[count], "on") == 0)
00164                     opts.showtopics = 1;
00165                 else if (strcmp(argv[count], "off") == 0)
00166                     opts.showtopics = 0;
00167                 else
00168                     usage();
00169             }
00170             else
00171                 usage();
00172         }
00173         count++;
00174     }
00175     
00176 }
00177 
00178 
00179 void myconnect(IPStack& ipstack, MQTT::Client<IPStack, Countdown, 1000>& client, MQTTPacket_connectData& data)
00180 {
00181     printf("Connecting to %s:%d\n", opts.host, opts.port);
00182     int rc = ipstack.connect(opts.host, opts.port);
00183     if (rc != 0)
00184         printf("rc from TCP connect is %d\n", rc);
00185 
00186     rc = client.connect(data);
00187     if (rc != 0)
00188     {
00189         printf("Failed to connect, return code %d\n", rc);
00190         exit(-1);   
00191     }
00192     printf("Connected\n");
00193 }
00194 
00195 
00196 void messageArrived(MQTT::MessageData& md)
00197 {
00198     MQTT::Message &message = md.message;
00199 
00200     if (opts.showtopics)
00201         printf("%.*s\t", md.topicName.lenstring.len, md.topicName.lenstring.data);
00202     if (opts.nodelimiter)
00203         printf("%.*s", (int)message.payloadlen, (char*)message.payload);
00204     else
00205         printf("%.*s%s", (int)message.payloadlen, (char*)message.payload, opts.delimiter);
00206     fflush(stdout);
00207 }
00208 
00209 
00210 int main(int argc, char** argv)
00211 {
00212     int rc = 0;
00213     
00214     if (argc < 2)
00215         usage();
00216     
00217     char* topic = argv[1];
00218 
00219     if (strchr(topic, '#') || strchr(topic, '+'))
00220         opts.showtopics = 1;
00221     if (opts.showtopics)
00222         printf("topic is %s\n", topic);
00223 
00224     getopts(argc, argv);    
00225 
00226     IPStack ipstack = IPStack();
00227     MQTT::Client<IPStack, Countdown, 1000> client = MQTT::Client<IPStack, Countdown, 1000>(ipstack);
00228 
00229     signal(SIGINT, cfinish);
00230     signal(SIGTERM, cfinish);
00231  
00232     MQTTPacket_connectData data = MQTTPacket_connectData_initializer;       
00233     data.willFlag = 0;
00234     data.MQTTVersion = 3;
00235     data.clientID.cstring = opts.clientid;
00236     data.username.cstring = opts.username;
00237     data.password.cstring = opts.password;
00238 
00239     data.keepAliveInterval = 10;
00240     data.cleansession = 1;
00241     printf("will flag %d\n", data.willFlag);
00242     
00243     myconnect(ipstack, client, data);
00244     
00245     rc = client.subscribe(topic, opts.qos, messageArrived);
00246     printf("Subscribed %d\n", rc);
00247 
00248     while (!toStop)
00249     {
00250         client.yield(1000); 
00251 
00252         //if (!client.isconnected)
00253         //  myconnect(ipstack, client, data);
00254     }
00255     
00256     printf("Stopping\n");
00257 
00258     rc = client.disconnect();
00259 
00260     ipstack.disconnect();
00261 
00262     return 0;
00263 }
00264 
00265