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

Diff for /libaitmqtt/src/cliside.c between versions 1.1.2.3 and 1.3.12.2

version 1.1.2.3, 2012/05/01 01:12:16 version 1.3.12.2, 2022/09/15 15:04:44
Line 12  terms: Line 12  terms:
 All of the documentation and software included in the ELWIX and AITNET  All of the documentation and software included in the ELWIX and AITNET
 Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>  Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
   
Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012Copyright 2004 - 2022
         by Michael Pounov <misho@elwix.org>.  All rights reserved.          by Michael Pounov <misho@elwix.org>.  All rights reserved.
   
 Redistribution and use in source and binary forms, with or without  Redistribution and use in source and binary forms, with or without
Line 50  SUCH DAMAGE. Line 50  SUCH DAMAGE.
  * mqtt_cli_Open() - Open client connection to MQTT broker   * mqtt_cli_Open() - Open client connection to MQTT broker
  *   *
  * @addr = brokers address   * @addr = brokers address
    * @timeout = timeout
  * return: NULL error or !=NULL connected to broker   * return: NULL error or !=NULL connected to broker
  */   */
 mqtt_cli_t *  mqtt_cli_t *
mqtt_cli_Open(struct sockaddr *addr)mqtt_cli_Open(struct sockaddr *addr, u_short timeout)
 {  {
         mqtt_cli_t *cli;          mqtt_cli_t *cli;
   
         if (!addr)          if (!addr)
                 return NULL;                  return NULL;
   
        cli = malloc(sizeof(mqtt_cli_t));        cli = e_malloc(sizeof(mqtt_cli_t));
         if (!cli) {          if (!cli) {
                 LOGERR;                  LOGERR;
                 return NULL;                  return NULL;
         } else          } else
                 memset(cli, 0, sizeof(mqtt_cli_t));                  memset(cli, 0, sizeof(mqtt_cli_t));
   
           cli->timeout = timeout;
         cli->sock = socket(addr->sa_family, SOCK_STREAM, IPPROTO_TCP);          cli->sock = socket(addr->sa_family, SOCK_STREAM, IPPROTO_TCP);
         if (cli->sock == -1) {          if (cli->sock == -1) {
                 LOGERR;                  LOGERR;
                free(cli);                e_free(cli);
                 return NULL;                  return NULL;
         }          }
   #ifndef __linux__
         if (connect(cli->sock, addr, addr->sa_len) == -1) {          if (connect(cli->sock, addr, addr->sa_len) == -1) {
   #else
           if (connect(cli->sock, addr, addr->sa_family == AF_INET6 ? 
                                   sizeof(struct sockaddr_in6) : 
                                   sizeof(struct sockaddr_in)) == -1) {
   #endif
                 LOGERR;                  LOGERR;
                 close(cli->sock);                  close(cli->sock);
                free(cli);                e_free(cli);
                 return NULL;                  return NULL;
         }          }
   
         cli->buf = mqtt_msgAlloc(USHRT_MAX);  
         if (!cli->buf) {  
                 close(cli->sock);  
                 free(cli);  
                 return NULL;  
         }  
   
         return cli;          return cli;
 }  }
   
Line 99  mqtt_cli_Open(struct sockaddr *addr) Line 100  mqtt_cli_Open(struct sockaddr *addr)
 int  int
 mqtt_cli_Close(mqtt_cli_t ** __restrict cli)  mqtt_cli_Close(mqtt_cli_t ** __restrict cli)
 {  {
        int siz = 0;        int siz;
   
         if (!cli || !*cli)          if (!cli || !*cli)
                 return -1;                  return -1;
   
         /* send disconnect */          /* send disconnect */
        siz = mqtt_msgDISCONNECT((*cli)->buf);        (*cli)->buf = mqtt_msgDISCONNECT();
        if (siz > -1) {        if ((*cli)->buf) {
                siz = send((*cli)->sock, (*cli)->buf->msg_base, siz, MSG_NOSIGNAL);                siz = send((*cli)->sock, (*cli)->buf->msg_base, (*cli)->buf->msg_len, MSG_NOSIGNAL);
                 if (siz > -1)                  if (siz > -1)
                         shutdown((*cli)->sock, SHUT_RDWR);                          shutdown((*cli)->sock, SHUT_RDWR);
         }          }
         close((*cli)->sock);          close((*cli)->sock);
   
        mqtt_msgFree(&(*cli)->buf, 42);        mqtt_msgFree(&(*cli)->buf, 0);
   
        free(*cli);        e_free(*cli);
         *cli = NULL;          *cli = NULL;
         return 0;          return 0;
 }  }
   
   /*
    * mqtt_cli_Subscribe() - Subscribe to broker
    *
    * @cli = connected client
    * @Topics = Topics for subscribes
    * @msgID = Message ID
    * return: NULL error or !=NULL allocated array with subscribed QoS responses, 
    *      must be e_free() result!
    */
   u_char *
   mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t ** __restrict Topics, u_short msgID)
   {
           int siz = 0;
           u_short mid = 0;
           u_char *qoses = NULL;
   
           if (!cli || !Topics)
                   return NULL;
   
           /* send subscribe */
           cli->buf = mqtt_msgSUBSCRIBE(Topics, msgID);
           if (!cli->buf)
                   return NULL;
           siz = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL);
           if (siz == -1) {
                   LOGERR;
                   return NULL;
           } else
                   mqtt_msgFree(&cli->buf, 0);
   
           if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) {
                   return NULL;
           } else if (siz && mqtt_KeepAlive(cli->sock, cli->timeout, 1))
                   return NULL;
   
           /* receive suback */
           siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0);
           if (siz == -1) {
                   LOGERR;
                   return NULL;
           }
           siz = mqtt_readSUBACK(cli->buf, &mid, &qoses);
           if (siz == -1)
                   return NULL;
           if (msgID != mid) {
                   free(qoses);
                   mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, mid);
                   return NULL;
           }
   
           return qoses;
   }
   
   #if 0
   /*
    * mqtt_cli_Unsubscribe() - Unsubscribe from broker
    *
    * @cli = connected client
    * @Topics = Topics for unsubscribes
    * @msgID = Message ID
    * @Dup = Duplicated request
    * @QoS = Message QoS
    * return: -1 error or 0 ok
    */
   int
   mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t * __restrict Topics, 
                   u_short msgID, u_char Dup, u_char QoS)
   {
           int siz = 0;
   
           if (!cli || !Topics)
                   return -1;
   
           /* send unsubscribe */
           siz = mqtt_msgUNSUBSCRIBE(cli->buf, Topics, msgID, Dup, QoS);
           if (siz == -1)
                   return -1;
           siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL);
           if (siz == -1) {
                   LOGERR;
                   return -1;
           } else
                   memset(cli->buf->msg_base, 0, cli->buf->msg_len);
   
           if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) {
                   return -1;
           } else if (siz && mqtt_KeepAlive(cli->sock, cli->timeout, 1))
                   return -1;
   
           /* receive unsuback */
           siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0);
           if (siz == -1) {
                   LOGERR;
                   return -1;
           }
           siz = mqtt_readUNSUBACK(cli->buf);
           if (siz == -1)
                   return -1;
           if (msgID != siz) {
                   mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz);
                   return -1;
           }
   
           return 0;
   }
   
   /*
    * mqtt_cli_Publish() - Publish message to broker
    *
    * @cli = connected client
    * @msgID = Message ID
    * @Dup = Duplicated request
    * @QoS = Message QoS
    * @Retain = Retain message
    * @csTopic = Topic
    * @pData = Data
    * @datLen = Data length
    * return: -1 error or > -1 sended bytes
    */
   int
   mqtt_cli_Publish(mqtt_cli_t * __restrict cli, u_short msgID, u_char Dup, u_char QoS, u_char Retain, 
                   const char *csTopic, const void *pData, int datLen)
   {
           int wlen = 0, siz = 0;
   
           if (!cli || !csTopic)
                   return -1;
   
           /* send publish */
           siz = mqtt_msgPUBLISH(cli->buf, csTopic, msgID, Dup, QoS, Retain, pData, datLen);
           if (siz == -1)
                   return -1;
           siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL);
           if (siz == -1) {
                   LOGERR;
                   return -1;
           } else {
                   wlen = siz;
                   memset(cli->buf->msg_base, 0, cli->buf->msg_len);
           }
   
           if (QoS == MQTT_QOS_ONCE)       /* no reply */
                   goto end;
   
           if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) {
                   return -1;
           } else if (siz && mqtt_KeepAlive(cli->sock, cli->timeout, 1))
                   return -1;
   
           /* receive PUBxxx */
           siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0);
           if (siz == -1) {
                   LOGERR;
                   return -1;
           }
   
           if (QoS == MQTT_QOS_ACK) {      /* reply with PUBACK */
                   siz = mqtt_readPUBACK(cli->buf);
                   if (siz == -1)
                           return -1;
                   if (msgID != siz) {
                           mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz);
                           return -1;
                   }
                   goto end;
           } else {                        /* reply with PUBREC */
                   siz = mqtt_readPUBREC(cli->buf);
                   if (siz == -1)
                           return -1;
                   if (msgID != siz) {
                           mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz);
                           return -1;
                   }
           }
   
           do {
                   /* send publish release QoS == 2 */
                   siz = mqtt_msgPUBREL(cli->buf, msgID);
                   if (siz == -1)
                           return -1;
                   siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL);
                   if (siz == -1) {
                           LOGERR;
                           return -1;
                   } else
                           memset(cli->buf->msg_base, 0, cli->buf->msg_len);
   
                   if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) {
                           return -1;
                   } else if (siz && mqtt_KeepAlive(cli->sock, cli->timeout, 1)) {
                           if (Dup++ > 1)
                                   return -1;
                           else
                                   continue;
                   }
   
                   /* receive PUBCOMP */
                   siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0);
                   if (siz == -1) {
                           LOGERR;
                           return -1;
                   }
   
                   siz = mqtt_readPUBCOMP(cli->buf);
                   if (siz == -1)
                           return -1;
                   if (msgID != siz) {
                           mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz);
                           if (Dup++ > 1)
                                   return -1;
                           else
                                   continue;
                   }
           } while (0);
   
   end:
           return wlen;
   }
   #endif
Removed from v.1.1.2.3  
changed lines
  Added in v.1.3.12.2

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