D:/my_data/GIT/network_apps/netapps/mqtt/common/mqtt_common.h

Go to the documentation of this file.

/******************************************************************************
*
*   Copyright (C) 2014 Texas Instruments Incorporated
*
*   All rights reserved. Property of Texas Instruments Incorporated.
*   Restricted rights to use, duplicate or disclose this code are
*   granted through contract.
*
*   The program may not be used without the written permission of
*   Texas Instruments Incorporated or against the terms and conditions
*   stipulated in the agreement under which this program has been supplied,
*   and under no circumstances can it be used with non-TI connectivity device.
*
******************************************************************************/

/*
  mqtt_common.h

  This module outlines the interfaces that are common to both client amd
  server components. The applications are not expected to utilize the
  services outlined in this module.
*/

#ifndef __MQTT_COMMON_H__
#define __MQTT_COMMON_H__

#include <stdbool.h>
#include <stdlib.h>
#include <stdio.h>

#define MQTT_COMMON_VERSTR "1.0.0" 
typedef int            i32;
typedef unsigned int   u32;
typedef unsigned char   u8;
typedef char            i8;
typedef unsigned short u16;
typedef short          i16;

#define MIN(a, b) ((a > b)? b : a)

#define MQTT_CONNECT      0x01
#define MQTT_CONNACK      0x02
#define MQTT_PUBLISH      0x03
#define MQTT_PUBACK       0x04
#define MQTT_PUBREC       0x05
#define MQTT_PUBREL       0x06
#define MQTT_PUBCOMP      0x07
#define MQTT_SUBSCRIBE    0x08
#define MQTT_SUBACK       0x09
#define MQTT_UNSUBSCRIBE  0x0A
#define MQTT_UNSUBACK     0x0B
#define MQTT_PINGREQ      0x0C
#define MQTT_PINGRSP      0x0D
#define MQTT_DISCONNECT   0x0E

#define MAX_FH_LEN        0x05    
#define MAX_REMLEN_BYTES  (MAX_FH_LEN - 1)  

#define MAKE_FH_BYTE1(msg_type,  flags) (u8)((msg_type << 4) | flags)

#define MAKE_FH_FLAGS(bool_dup, enum_qos, bool_retain)                  \
        (u8)(((bool_dup << 3) | (enum_qos << 1) | bool_retain) & 0xF)

#define QOS_VALUE(enum_qos) (u8)(enum_qos & 0x3)
#define QFL_VALUE           0x80  
#define DUP_FLAG_VAL(bool_val) (u8)(bool_val << 3)

#define BOOL_RETAIN(fh_byte1)  ((fh_byte1 & 0x1)? true : false)
#define BOOL_DUP(fh_byte1)     ((fh_byte1 & 0x8)? true : false)
#define ENUM_QOS(fh_byte1)     (enum mqtt_qos)((fh_byte1 & 0x6) >> 1)

#define MSG_TYPE(fh_byte1)  (u8)((fh_byte1 & 0xf0) >> 4)

static inline u32 buf_wr_nbytes(u8 *dst, const u8 *src, u32 n)
{
        u32 c = n; 
        while(c--)
                *dst++ = *src++;

        return n;
}

static inline u32 buf_set(u8 *dst, u8 val, u32 n)
{
        u32 c = n; 
        while(c--)
                *dst++ = val;

        return n;
}

static inline u32 buf_wr_nbo_2B(u8 *buf, u16 val)
{
        buf[0] = (u8)((val >> 8) & 0xFF); /* MSB */
        buf[1] = (u8)((val)      & 0xFF); /* LSB */
        return 2;
}

static inline u32 buf_rd_nbo_2B(const u8 *buf, u16 *val)
{
        *val = (u16)((buf[0] << 8) | (buf[1]));
        return 2;
}

struct mqtt_packet {

        u8                     msg_type;   
        u8                     fh_byte1;   
        u16                    msg_id;     
        u8                     n_refs;     
        u8                     pad[3];

        u8                     offset;     
        u8                     fh_len;     
        u16                    vh_len;     
        u32                    pl_len;     
        u32                    private;

        u32                    maxlen;     
        u8                    *buffer;     
        void                  (*free)(struct mqtt_packet *mqp);

        struct mqtt_packet    *next;
};

#define MQP_FHEADER_BUF(mqp)  (mqp->buffer + mqp->offset)
#define MQP_VHEADER_BUF(mqp)  (MQP_FHEADER_BUF(mqp) + mqp->fh_len)
#define MQP_PAYLOAD_BUF(mqp)  (MQP_VHEADER_BUF(mqp) + mqp->vh_len)

#define MQP_CONTENT_LEN(mqp)  (mqp->fh_len + mqp->vh_len + mqp->pl_len)
#define MQP_FREEBUF_LEN(mqp)  (mqp->maxlen - mqp->offset -      \
                               MQP_CONTENT_LEN(mqp))

#define MQP_FHEADER_VAL(mqp)  (mqp->fh_byte1)
#define MQP_FHEADER_MSG(mqp)  (MSG_TYPE(MQP_FHEADER_VAL(mqp)))
#define MQP_FHEADER_FLG(mqp)  (MSG_FLAGS(MQP_FHEADER_VAL(mqp)))

#define DEFINE_MQP_VEC(num_mqp, mqp_vec)                \
        static struct mqtt_packet mqp_vec[num_mqp];

#define DEFINE_MQP_BUF_VEC(num_mqp, mqp_vec, buf_len, buf_vec)        \
        DEFINE_MQP_VEC(num_mqp, mqp_vec);                             \
        static u8 buf_vec[num_mqp][buf_len];

/*---------------------------------------------------------------------
 * Heleper MACROS for PUBLISH-RX Message Processing
 *---------------------------------------------------------------------
 */

#define MQP_PUB_TOP_BUF(mqp) (MQP_VHEADER_BUF(mqp) + 2)

#define MQP_PUB_TOP_LEN(mqp) (mqp->vh_len - 2 - (mqp->msg_id? 2 : 0))

#define MQP_PUB_PAY_BUF(mqp) (mqp->pl_len? MQP_PAYLOAD_BUF(mqp) : NULL)

#define MQP_PUB_PAY_LEN(mqp) (mqp->pl_len)

#define WILL_RETAIN_VAL 0x20
#define WILL_CONFIG_VAL 0x04
#define CLEAN_START_VAL 0x02
#define USER_NAME_OPVAL 0x80
#define PASS_WORD_OPVAL 0x40

#define MQP_ERR_NETWORK   (-1)  
#define MQP_ERR_TIMEOUT   (-2)  
#define MQP_ERR_NET_OPS   (-3)  
#define MQP_ERR_FNPARAM   (-4)  
#define MQP_ERR_PKT_AVL   (-5)  
#define MQP_ERR_PKT_LEN   (-6)  
#define MQP_ERR_NOTCONN   (-7)  
#define MQP_ERR_BADCALL   (-8)  
#define MQP_ERR_CONTENT   (-9)  
#define MQP_ERR_LIBQUIT  (-10)  
#define MQP_ERR_NOT_DEF  (-32)  
/*---------------------------------------------------------------------
 * Common Operations
 *---------------------------------------------------------------------
 */

void mqp_free(struct mqtt_packet *mqp);

void mqp_reset(struct mqtt_packet *mqp);

void mqp_init(struct mqtt_packet *mqp, u8 offset);

static 
inline void mqp_buffer_attach(struct mqtt_packet *mqp, u8 *buffer, u32 length,
                              u8 offset)
{
        mqp_init(mqp, offset);

        mqp->buffer = buffer;
        mqp->maxlen = length;
        mqp->free   = NULL;

        return;
}

struct utf8_string {
        
        i8   *buffer;   
        u16   length;   
};

i32 mqp_buf_wr_utf8(u8 *buf, const struct utf8_string *utf8);

i32 mqp_buf_tail_wr_remlen(u8 *buf, u32 remlen);

i32 mqp_buf_rd_remlen(u8 *buf, u32 *remlen);

i32 
mqp_pub_append_topic(struct mqtt_packet *mqp, const struct utf8_string *topic,
                     u16 msg_id);

i32 mqp_pub_append_data(struct mqtt_packet *mqp, const u8 *data_buf,
                        u32 data_len);

bool mqp_proc_msg_id_ack_rx(struct mqtt_packet *mqp_raw, bool has_payload);

bool mqp_proc_pub_rx(struct mqtt_packet *mqp_raw);

/*
   Wait-List of MQTT Messages for which acknoledge is pending from remote node.
*/
struct mqtt_ack_wlist {
        
        struct mqtt_packet *head;  /* Points to head of single linked-list. */
        struct mqtt_packet *tail;  /* Points to tail of single linked-list. */
};

static inline bool mqp_ack_wlist_is_empty(struct mqtt_ack_wlist *list)
{
        return (NULL == list->head) ? true : false;
}

/*
   Add specified element into trailing end of list.

   Returns, on success, true, otherwise false.
*/
bool mqp_ack_wlist_append(struct mqtt_ack_wlist *list,
                          struct mqtt_packet    *elem);

/*
   Removes element that has specified msg_id from list. 

   Returns, on success, pointer to removed element, otherwise NULL.
*/
struct mqtt_packet *mqp_ack_wlist_remove(struct mqtt_ack_wlist *list,
                                          u16 msg_id);
/* 
   Removes and frees all elements in list. 
*/
void mqp_ack_wlist_purge(struct mqtt_ack_wlist *list);

static inline bool is_wlist_empty(const struct mqtt_ack_wlist *list)
{
        return list->head? false : true;
}

i32 mqp_prep_fh(struct mqtt_packet *mqp, u8 flags);

enum mqtt_qos {
        
        MQTT_QOS0,  
        MQTT_QOS1,  
        MQTT_QOS2   
};

struct utf8_strqos {

        i8             *buffer;  
        u16             length;  
        enum mqtt_qos   qosreq;  
};


struct secure_conn {
        
        void *method;  
        void *cipher;  
        u32   n_file;  
        char **files;  
};

struct device_net_services {

#define DEV_NETCONN_OPT_TCP  0x01  
#define DEV_NETCONN_OPT_UDP  0x02  
#define DEV_NETCONN_OPT_IP6  0x04  
#define DEV_NETCONN_OPT_URL  0x08  
#define DEV_NETCONN_OPT_SEC  0x10  
        i32 (*open)(u32 nwconn_opts, const i8 *server_addr, u16 port_number,
                    const struct secure_conn *nw_security);

        i32   (*send)(i32 comm, const u8 *buf, u32 len, void *ctx);

        i32   (*recv)(i32 comm, u8 *buf, u32 len, u32 wait_secs,
                      bool *err_timeo, void *ctx);

        i32   (*send_dest)(i32 comm, const u8 *buf, u32 len, u16 dest_port,
                           const i8 *dest_ip, u32 ip_len);

        i32   (*recv_from)(i32 comm, u8 *buf, u32 len, u16 *from_port,
                           i8 *from_ip, u32 *ip_len);

        i32   (*close)(i32 comm);

        i32 (*listen)(u32 nwconn_opts, u16 port_number,
                      const struct secure_conn *nw_security);

        i32 (*accept)(i32 listen, i8 *client_ip, u32 *ip_length);

        i32   (*io_mon)(i32 *recv_cvec, i32 *send_cvec,
                        i32 *rsvd_cvec,  u32 wait_secs);

        u32   (*time)(void);
};
 /* device_net_services */

/* Receive data from the specified network and read into the 'mqp' */
i32 mqp_recv(i32  net,     const struct device_net_services *net_ops,
             struct mqtt_packet *mqp, u32 wait_secs, bool *timed_out,
             void *ctx);

/*-----------------------------------------------------------------------------
 * Data structure for managing the QoS2 PUB RX packets and follow-ups
 *---------------------------------------------------------------------------*/

#define MAX_PUBREL_INFLT 8 /* Must be kept as a value of 2^n */

struct pub_qos2_cq { /* Circular Queue CQ to track QOS2 PUB RX messages */

        u16 id_vec[MAX_PUBREL_INFLT];  /* Vector to store RX Message-IDs */
        u8  n_free;                    /* Num of free elements in vector */
        u8  rd_idx;                    /* Index to Read  next Message-ID */
        u8  wr_idx;                    /* Index to Write next Message-ID */
};

/* Reset the specified Circular Queue (CQ) */
void qos2_pub_cq_reset(struct pub_qos2_cq *cq);

/* Append the message-id into the CQ tail. Return true on success, else false */
bool qos2_pub_cq_logup(struct pub_qos2_cq *cq, u16 msg_id);

/* Remove the message-id from the CQ head. Return true on success, else false */
bool qos2_pub_cq_unlog(struct pub_qos2_cq *cq, u16 msg_id);

/* Is the message-id available in the CQ ? Return true on success, else false */
bool qos2_pub_cq_check(struct pub_qos2_cq *cq, u16 msg_id);

/* Get the count of message-ID(s) availalbe in the CQ */
static inline i32 qos2_pub_cq_count(struct pub_qos2_cq *cq)
{
        return MAX_PUBREL_INFLT - cq->n_free;
}

struct client_ctx {

        void        *usr;  /* Client Usr */
        i32          net;  /* Socket HND */

        i8           remote_ip[16];
        u32          ip_length;

        u32          timeout;
        u16          ka_secs;

        u32          flags;

        struct client_ctx *next;
};

void cl_ctx_reset(struct client_ctx *cl_ctx);
void cl_ctx_timeout_insert(struct client_ctx **head,
                           struct client_ctx *elem);

void cl_ctx_remove(struct client_ctx **head,
                   struct client_ctx *elem);

#define KA_TIMEOUT_NONE 0xffffffff  /* Different than KA SECS = 0 */
void cl_ctx_timeout_update(struct client_ctx *cl_ctx, u32 now_secs);

#endif