[BACK]Return to mqtt_pub.c CVS log [TXT] [DIR] Up to [ELWIX - Embedded LightWeight unIX -] / mqtt / src

Diff for /mqtt/src/mqtt_pub.c between versions 1.1.2.2 and 1.1.2.5

version 1.1.2.2, 2011/12/20 16:04:34 version 1.1.2.5, 2012/01/05 10:01:20
Line 19  Usage(void) Line 19  Usage(void)
                 " Syntax: mqtt_pub [options] <connect_to_broker[:port]> <ConnectID> <topic> <value_for_publish>\n\n"                  " Syntax: mqtt_pub [options] <connect_to_broker[:port]> <ConnectID> <topic> <value_for_publish>\n\n"
                 "\t-f\t\t\t'value_for_publish' is file name instead text\n"                  "\t-f\t\t\t'value_for_publish' is file name instead text\n"
                 "\t-q <QoS>\t\tQoS level (0-at most 1, 1-at least 1, 2-exactly 1)\n"                  "\t-q <QoS>\t\tQoS level (0-at most 1, 1-at least 1, 2-exactly 1)\n"
                   "\t-m <message_id>\t\tMessage ID for publish message\n"
                 "\t-d\t\t\tSend duplicate message\n"                  "\t-d\t\t\tSend duplicate message\n"
                 "\t-r\t\t\tRetain message from broker\n\n"                  "\t-r\t\t\tRetain message from broker\n\n"
                 "\t-C\t\t\tNot clear before connect!!!\n"                  "\t-C\t\t\tNot clear before connect!!!\n"
Line 49  cleanArgs(struct tagArgs * __restrict args) Line 50  cleanArgs(struct tagArgs * __restrict args)
 static int  static int
 Publish(int sock)  Publish(int sock)
 {  {
        return 0;        int siz = 0;
         struct pollfd pfd;
         struct mqtthdr *hdr;
 
         siz = mqtt_msgPUBLISH(args->msg, AIT_GET_STR(&args->Publish), args->MsgID, args->Dup, 
                         args->QoS, args->Retain, AIT_GET_PTR2(&args->Value), AIT_LEN(&args->Value));
         if (siz == -1) {
                 printf("Error:: msgPUBLISH #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
                 return -1;
         }
 
         siz = send(sock, args->msg->msg_base, siz, 0);
         if (siz == -1) {
                 printf("Error:: send() #%d - %s\n", errno, strerror(errno));
                 return -1;
         } else
                 ioVERBOSE(3) printf("Sended PUBLISH %d bytes\n", siz);
 
         /* QoS == MQTT_QOS_ONCE, no wait for reply */
         if (args->QoS == MQTT_QOS_ONCE)
                 return 0;
 
         pfd.fd = sock;
         pfd.events = POLLIN | POLLPRI;
         switch (poll(&pfd, 1, args->ka * 1000)) {
                 case -1:
                         printf("Error:: poll() #%d - %s\n", errno, strerror(errno));
                         return -1;
                 case 0:
                         ioVERBOSE(3) printf("Timeout reached (%d) ...\n", args->ka * 1000);
                         return -1;
         }
         if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL))
                 return -1;
 
         siz = recv(sock, args->msg->msg_base, args->msg->msg_len, 0);
         if (siz == -1) {
                 printf("Error:: recv() #%d - %s\n", errno, strerror(errno));
                 return -1;
         } else
                 ioVERBOSE(3) printf("Received %d bytes\n", siz);
         if (!siz)
                 return -1;
 
         /* QoS == MQTT_QOS_ACK, wait for PUBACK */
         if (args->QoS == MQTT_QOS_ACK)
                 return mqtt_readPUBACK(args->msg);
 
         /* QoS == MQTT_QOS_EXACTLY */
         if (mqtt_readPUBREC(args->msg) != args->MsgID) {
                 printf("Error:: Message not delivered\n");
                 return -1;
         }
 
         siz = mqtt_msgPUBREL(args->msg, args->MsgID);
         if (siz == -1) {
                 printf("Error:: msgPUBLISH #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
                 return -1;
         }
 
         siz = send(sock, args->msg->msg_base, siz, 0);
         if (siz == -1) {
                 printf("Error:: send() #%d - %s\n", errno, strerror(errno));
                 return -1;
         } else
                 ioVERBOSE(3) printf("Sended PUBLISH %d bytes\n", siz);
 
         pfd.events = POLLIN | POLLPRI;
         switch (poll(&pfd, 1, args->ka * 1000)) {
                 case -1:
                         printf("Error:: poll() #%d - %s\n", errno, strerror(errno));
                         return -1;
                 case 0:
                         ioVERBOSE(3) printf("Timeout reached (%d) ...\n", args->ka * 1000);
                         return -1;
         }
         if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL))
                 return -1;
 
         siz = recv(sock, args->msg->msg_base, args->msg->msg_len, 0);
         if (siz == -1) {
                 printf("Error:: recv() #%d - %s\n", errno, strerror(errno));
                 return -1;
         } else
                 ioVERBOSE(3) printf("Received %d bytes\n", siz);
         if (!siz)
                 return -1;
 
         return mqtt_readPUBCOMP(args->msg);
 }  }
   
   
Line 80  main(int argc, char **argv) Line 169  main(int argc, char **argv)
         AIT_SET_STR(&args->Pass, "");          AIT_SET_STR(&args->Pass, "");
   
         args->ka = MQTT_KEEPALIVE;          args->ka = MQTT_KEEPALIVE;
        while ((ch = getopt(argc, argv, "T:U:P:p:q:drCW:M:fvh")) != -1)        while ((ch = getopt(argc, argv, "T:U:P:p:q:drCW:M:m:fvh")) != -1)
                 switch (ch) {                  switch (ch) {
                         case 'T':                          case 'T':
                                 args->ka = (u_short) strtol(optarg, NULL, 0);                                  args->ka = (u_short) strtol(optarg, NULL, 0);
Line 101  main(int argc, char **argv) Line 190  main(int argc, char **argv)
                                 AIT_FREE_VAL(&args->Pass);                                  AIT_FREE_VAL(&args->Pass);
                                 AIT_SET_STR(&args->Pass, optarg);                                  AIT_SET_STR(&args->Pass, optarg);
                                 break;                                  break;
                           case 'm':
                                   args->MsgID = (u_short) strtol(optarg, NULL, 0);
                                   break;
                         case 'p':                          case 'p':
                                 port = (u_short) strtol(optarg, NULL, 0);                                  port = (u_short) strtol(optarg, NULL, 0);
                                 break;                                  break;
Line 158  main(int argc, char **argv) Line 250  main(int argc, char **argv)
                 Usage();                  Usage();
                 return 1;                  return 1;
         }          }
           if (args->QoS && !args->MsgID)
                   args->MsgID = MQTT_DEFAULT_MSGID;
         ioVERBOSE(1) printf("Connecting to %s:%d ...\n", io_n2addr(&args->addr, &val), io_n2port(&args->addr));          ioVERBOSE(1) printf("Connecting to %s:%d ...\n", io_n2addr(&args->addr, &val), io_n2port(&args->addr));
   
         if ((sock = InitClient()) == -1) {          if ((sock = InitClient()) == -1) {
Line 166  main(int argc, char **argv) Line 260  main(int argc, char **argv)
                 return 2;                  return 2;
         }          }
   
           if (args->isFile && !OpenFile()) {
                   args->free(args);
                   free(args);
                   return 3;
           }
   
         printf("Connected ... ");          printf("Connected ... ");
        switch ((ret = try2Connect(sock))) {        switch ((ret = ConnectClient(sock))) {
                 case -1:                  case -1:
                         printf(">> FAILED!\n");                          printf(">> FAILED!\n");
                         break;                          break;
Line 192  main(int argc, char **argv) Line 292  main(int argc, char **argv)
         }          }
   
         if (ret == MQTT_RETCODE_ACCEPTED) {          if (ret == MQTT_RETCODE_ACCEPTED) {
                ret = Publish(sock);                ret = !(Publish(sock) == args->MsgID);
                shutdown(sock, SHUT_RDWR);                CloseClient(sock);
        } else        } else {
                ret = 3;                close(sock);
                 ret = 4;
         }
   
        close(sock);        CloseFile();
         args->free(args);          args->free(args);
         free(args);          free(args);
         return ret;          return ret;
Removed from v.1.1.2.2  
changed lines
  Added in v.1.1.2.5

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>