David Fletcher
Part of TI's mqtt
Dependents: mqtt_V1 cc3100_Test_mqtt_CM3
Diff: mqtt_client.cpp
- Revision:
- 0:087b5655778d
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/mqtt_client.cpp Sat Jun 06 13:28:41 2015 +0000
@@ -0,0 +1,2317 @@
+/******************************************************************************
+*
+* 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_client.c
+
+ The module provides implementation to the public interface of the MQTT
+ Client Library.
+*/
+#include "FreeRTOS.h"
+#include "mqtt_client.h"
+#include "semphr.h"
+
+#include "cli_uart.h"
+
+#define PRINT_BUF_LEN 128
+extern int8_t print_buf[PRINT_BUF_LEN];
+
+namespace mbed_mqtt
+{
+
+//void createMutex(void);
+
+static void *mutex = NULL;
+static void (*mutex_lockin)(void*) = NULL;
+static void (*mutex_unlock)(void*) = NULL;
+
+#define MUTEX_LOCKIN() if(mutex_lockin) mutex_lockin(mutex);
+#define MUTEX_UNLOCK() if(mutex_unlock) mutex_unlock(mutex);
+
+static bool aux_dbg_enbl = true;
+int32_t (*debug_printf)(const char *fmt, ...) = NULL;
+
+#define USR_INFO debug_printf
+#define DBG_INFO(I, ...) if(aux_dbg_enbl) debug_printf(I, ##__VA_ARGS__)
+
+static const struct device_net_services *net_ops = NULL;
+
+static uint16_t msg_id = 0xFFFF;
+static inline uint16_t assign_new_msg_id()
+{
+ return msg_id += 2;
+}
+
+SemaphoreHandle_t xSemaphore = NULL;
+void createMutex()
+{
+
+ xSemaphore = xSemaphoreCreateMutex();
+}
+
+/*-----------------------------------------------------------------------------
+ * Data structure for managing the client and its nuances
+ *---------------------------------------------------------------------------*/
+
+/* Construct to manage TX for network that requires LIB to send a partial and
+ incremental data to support restrictive TCP segments. Specifically, for the
+ deployments, in which the network layer supports small segments, there can
+ be only one in-flight message.
+*/
+struct tx_part_pkt {
+
+ /* Refers to MQP, if TX for it is active, otherwise, set it to NULL */
+ struct mqtt_packet *tx_mqp;
+
+ union {
+#define VH_MSG_LEN 4
+ uint8_t vh_msg[VH_MSG_LEN]; /* VH msg for MQP = NULL */
+ const uint8_t *buffer; /* Refers to data in MQP */
+ };
+
+ uint32_t length; /* Length of entire data */
+ uint32_t offset; /* Next data for sending */
+};
+
+static bool tx_part_setup(struct tx_part_pkt *tx_part, const uint8_t *buffer,
+ uint32_t length, struct mqtt_packet *tx_mqp)
+{
+
+ if(tx_mqp) {
+ tx_part->buffer = buffer;
+ } else {
+ if(VH_MSG_LEN < length)
+ return false;
+
+ buf_wr_nbytes(tx_part->vh_msg, buffer, length);
+ }
+
+ tx_part->length = length;
+ tx_part->tx_mqp = tx_mqp;
+
+ return true;
+}
+
+static void tx_part_reset(struct tx_part_pkt *tx_part)
+{
+
+ struct mqtt_packet *tx_mqp = tx_part->tx_mqp;
+ if(tx_mqp) {
+ mqp_free(tx_mqp);
+ }
+ tx_part->vh_msg[0] = 0x00;
+ tx_part->tx_mqp = NULL;
+ tx_part->length = 0;
+ tx_part->offset = 0;
+
+ return;
+}
+
+static const uint8_t *tx_part_buf_p(struct tx_part_pkt *tx_part)
+{
+ struct mqtt_packet *tx_mqp = tx_part->tx_mqp;
+ uint32_t offset = tx_part->offset;
+
+ return tx_mqp?
+ tx_part->buffer + offset :
+ tx_part->vh_msg + offset;
+}
+
+static void tx_part_addup(struct tx_part_pkt *tx_part, uint32_t offset)
+{
+ tx_part->offset += offset;
+}
+
+#define TX_PART_BUFFER(tx_part) tx_part_buf_p(tx_part)
+#define TX_PART_BUF_SZ(tx_part) (tx_part->length - tx_part->offset)
+#define TX_PART_IN_USE(tx_part) TX_PART_BUF_SZ(tx_part)
+
+enum module_state {
+
+ WAIT_INIT_STATE,
+ INIT_DONE_STATE = 0x01,
+};
+
+static enum module_state cl_lib_state = WAIT_INIT_STATE;
+
+static uint16_t loopb_portid = 0;
+static bool grp_has_cbfn = false;
+
+#define USE_PROTO_V31_FLAG MQTT_CFG_PROTOCOL_V31
+#define APP_RECV_TASK_FLAG MQTT_CFG_APP_HAS_RTSK
+#define GROUP_CONTEXT_FLAG MQTT_CFG_MK_GROUP_CTX
+
+#define CLEAN_SESSION_FLAG 0x00010000
+#define CONNACK_AWAIT_FLAG 0x00020000
+#define NOW_CONNECTED_FLAG 0x00040000
+#define KA_PINGER_RSP_FLAG 0x00080000
+#define USER_PING_RSP_FLAG 0x00100000
+#define NETWORK_CLOSE_FLAG 0x00200000
+#define DO_CONNACK_TO_FLAG 0x00400000
+
+static struct client_ctx *free_ctxs = NULL; /* CTX construct available */
+static struct client_ctx *used_ctxs = NULL; /* Relevant only for group */
+static struct client_ctx *conn_ctxs = NULL; /* Relevant only for group */
+
+static void cl_ctx_freeup(struct client_ctx *cl_ctx)
+{
+ cl_ctx->next = free_ctxs;
+ free_ctxs = cl_ctx;
+
+ return;
+}
+
+#define IS_PROTO_VER31(cl_ctx) (cl_ctx->flags & USE_PROTO_V31_FLAG)
+#define AWAITS_CONNACK(cl_ctx) (cl_ctx->flags & CONNACK_AWAIT_FLAG)
+#define HAS_CONNECTION(cl_ctx) (cl_ctx->flags & NOW_CONNECTED_FLAG)
+#define AWAITS_KA_PING(cl_ctx) (cl_ctx->flags & KA_PINGER_RSP_FLAG)
+#define AWAITS_PINGRSP(cl_ctx) (cl_ctx->flags & USER_PING_RSP_FLAG)
+#define IS_CLN_SESSION(cl_ctx) (cl_ctx->flags & CLEAN_SESSION_FLAG)
+#define RECV_TASK_AVBL(cl_ctx) (cl_ctx->flags & APP_RECV_TASK_FLAG)
+#define A_GROUP_MEMBER(cl_ctx) (cl_ctx->flags & GROUP_CONTEXT_FLAG)
+#define NEED_NET_CLOSE(cl_ctx) (cl_ctx->flags & NETWORK_CLOSE_FLAG)
+#define CFG_CONNACK_TO(cl_ctx) (cl_ctx->flags & DO_CONNACK_TO_FLAG)
+
+#ifndef CFG_CL_MQTT_CTXS
+#define MAX_NWCONN 4
+#else
+#define MAX_NWCONN CFG_CL_MQTT_CTXS
+#endif
+
+static struct client_desc {
+
+ /* ALERT: "context" must be first elem in this structure, do not move */
+ struct client_ctx context;
+
+#define CLIENT(cl_ctx) ((struct client_desc*) cl_ctx)
+#define CL_CTX(client) ((struct client_ctx*) client)
+
+ /* Order/Sequence: Client ID, Will Topic, Will Msg, Username, Password */
+ const struct utf8_string *conn_pl_utf8s[5]; /* Ref: CONNECT Payload */
+ uint8_t will_opts;
+
+ /* Wait-List for Level 1 ACK(s), which are PUBACK, PUBREC, UN/SUBACK */
+ struct mqtt_ack_wlist qos_ack1_wl;
+
+ /* Circular queue to track QOS2 PUBLISH packets from the server. They
+ are tracked for the duration of PUBLISH-RX to PUBREL-RX.
+ */
+ struct pub_qos2_cq qos2_rx_cq;
+
+ /* Circular queue to track QOS2 PUBLISH packets from the client. They
+ are tracked for the duration of PUBREC-RX to PUBOMP-RX.
+ */
+ struct pub_qos2_cq qos2_tx_cq;
+
+ struct mqtt_client_ctx_cbs app_ctx_cbs; /* Callback funcs from app */
+#define CTX_CBS_PTR(cl_ctx) &(CLIENT(cl_ctx)->app_ctx_cbs)
+
+ struct tx_part_pkt tx_part;/* Reference to partial TX PKT */
+ struct mqtt_packet *rx_mqp; /* Reference to partial RX PKT */
+ void *app;
+
+ uint32_t nwconn_opts;
+ char *server_addr;
+ uint16_t port_number;
+ struct secure_conn nw_security;
+
+} clients[MAX_NWCONN];
+
+static void client_reset(struct client_desc *client)
+{
+
+ struct mqtt_client_ctx_cbs *ctx_cbs = &client->app_ctx_cbs;
+ int32_t i = 0;
+
+ cl_ctx_reset(CL_CTX(client));
+
+ for(i = 0; i < 5; i++) {
+ client->conn_pl_utf8s[i] = NULL;
+ }
+ client->will_opts = 0;
+
+ mqp_ack_wlist_purge(&client->qos_ack1_wl);
+
+ qos2_pub_cq_reset(&client->qos2_rx_cq);
+ qos2_pub_cq_reset(&client->qos2_tx_cq);
+
+ ctx_cbs->publish_rx = NULL;
+ ctx_cbs->ack_notify = NULL;
+ ctx_cbs->disconn_cb = NULL;
+
+ tx_part_reset(&client->tx_part);
+ client->rx_mqp = NULL;
+ client->app = NULL;
+
+ client->nwconn_opts = 0;
+ client->server_addr = NULL;
+ client->port_number = 0;
+
+ secure_conn_struct_init(&client->nw_security);
+
+ return;
+}
+
+static void client_desc_init(void)
+{
+ int32_t i = 0;
+ for(i = 0; i < MAX_NWCONN; i++) {
+ struct client_desc *client = clients + i;
+ struct client_ctx *cl_ctx = CL_CTX(client);
+
+ /* Initialize certain fields to defaults */
+ client->qos_ack1_wl.head = NULL;
+ client->qos_ack1_wl.tail = NULL;
+ client->tx_part.tx_mqp = NULL;
+
+ client_reset(client); /* Reset remaining */
+
+ cl_ctx->next = free_ctxs;
+ free_ctxs = cl_ctx;
+ }
+
+ return;
+}
+
+static void mqp_free_locked(struct mqtt_packet *mqp)
+{
+
+// MUTEX_LOCKIN();
+
+// MUTEX_UNLOCK();
+
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 40 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+ mqp_free(mqp);
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ xSemaphoreGive( xSemaphore );
+
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+}
+
+/*----------------------------------------------------------------------------
+ * Fix-up to prevent certain good and non-callback MQP being reported to app
+ *----------------------------------------------------------------------------
+ */
+/* cor --> clear on read. */
+static bool mqp_do_not_report_cor(struct mqtt_packet *mqp)
+{
+ bool rv = (1 == mqp->private_)? true : false;
+ mqp->private_ = 0;
+ return rv;
+}
+
+#define MQP_RX_DO_NOT_RPT_COR(mqp) mqp_do_not_report_cor(mqp)
+
+/* Only if MQP has good RX data i.e. this macro shouldn't be used for bad RX */
+#define MQP_RX_DO_NOT_RPT_SET(mqp) (mqp->private_ = 1)
+
+#define MQP_TX_DONE_LEN_ADD(mqp, add) (mqp->private_ += add)
+#define MQP_TX_DONE_LEN_GET(mqp) (mqp->private_)
+
+/*---------------------------Fix-Up Ends ------------------------------------*/
+
+static int32_t loopb_net = -1;
+static const uint8_t LOOP_DATA[] = {0x00, 0x01};
+#define LOOP_DLEN sizeof(LOOP_DATA)
+
+static int32_t loopb_trigger(void)
+{
+
+
+ uint8_t ip_addr[] = {127,0,0,1};
+
+ return (-1 != loopb_net)?
+ net_ops->send_dest(loopb_net, LOOP_DATA, LOOP_DLEN,
+ loopb_portid, ip_addr, 4) : MQP_ERR_LIBQUIT;
+}
+
+static void session_311fix(struct client_ctx *cl_ctx)
+{
+ struct mqtt_ack_wlist *wl = &CLIENT(cl_ctx)->qos_ack1_wl;
+ struct mqtt_packet *elem = wl->head;
+
+ while(elem) {
+ struct mqtt_packet *next = elem->next;
+ if(MQTT_PUBLISH != elem->msg_type)
+ mqp_ack_wlist_remove(wl, elem->msg_id);
+
+ elem = next;
+ }
+
+ return;
+}
+
+static void session_delete(struct client_ctx *cl_ctx)
+{
+ struct client_desc *client = CLIENT(cl_ctx);
+
+ DBG_INFO("C: Cleaning session for net %d\n\r", cl_ctx->net);
+
+ qos2_pub_cq_reset(&client->qos2_rx_cq);
+ qos2_pub_cq_reset(&client->qos2_tx_cq);
+
+ mqp_ack_wlist_purge(&client->qos_ack1_wl);
+
+ return;
+}
+
+/*------------------------------------------------------------------------------
+ * Routine to manage error conditions in client - close the network connection
+ *----------------------------------------------------------------------------*/
+static void do_net_close(struct client_ctx *cl_ctx)
+{
+ int32_t net = cl_ctx->net;
+
+ if(-1 == net)
+ return; /* network closed already, must not happen */
+
+ if(IS_CLN_SESSION(cl_ctx)) {
+ session_delete(cl_ctx);
+ } else if(!IS_PROTO_VER31(cl_ctx)) {
+ /* Version 3.1.1 doesn't need SUB and UNSUB re-send */
+ session_311fix(cl_ctx);
+ }
+
+ tx_part_reset(&CLIENT(cl_ctx)->tx_part); /* Part TX, if any */
+
+ cl_ctx->flags &= ~(CONNACK_AWAIT_FLAG | NOW_CONNECTED_FLAG |
+ KA_PINGER_RSP_FLAG | USER_PING_RSP_FLAG |
+ NETWORK_CLOSE_FLAG | DO_CONNACK_TO_FLAG);
+
+ cl_ctx->net = -1;
+ net_ops->close(net);
+
+ USR_INFO("C: Net %d now closed\n\r", net);
+
+ return;
+}
+
+static void do_net_close_rx(struct client_ctx *cl_ctx, int32_t cause)
+{
+ struct mqtt_client_ctx_cbs *ctx_cbs = CTX_CBS_PTR(cl_ctx);
+
+ DBG_INFO("C: RX closing Net %d [%d]\n\r", cl_ctx->net, cause);
+
+ do_net_close(cl_ctx);
+ if(ctx_cbs->disconn_cb)
+ ctx_cbs->disconn_cb(CLIENT(cl_ctx)->app, cause);
+
+ if(A_GROUP_MEMBER(cl_ctx))
+ cl_ctx_remove(&used_ctxs, cl_ctx);
+
+ return;
+}
+
+static void do_net_close_tx(struct client_ctx *cl_ctx, char *cause)
+{
+ DBG_INFO("C: TX closing Net %d [%s]\n\r", cl_ctx->net, cause);
+
+ if(RECV_TASK_AVBL(cl_ctx)) {
+ cl_ctx->flags |= NETWORK_CLOSE_FLAG;
+ if(A_GROUP_MEMBER(cl_ctx))
+ loopb_trigger();
+ } else {
+ struct mqtt_packet *rx_mqp = CLIENT(cl_ctx)->rx_mqp;
+
+ do_net_close(cl_ctx); /* No RX Task, close now */
+
+ /* Release partial MQP, if any, for a CTX w/ CB */
+ if((NULL != rx_mqp) && (NULL != rx_mqp->free))
+ mqp_free(rx_mqp);
+
+ CLIENT(cl_ctx)->rx_mqp = NULL;
+ }
+
+ return;
+}
+
+/*----------------------------------------------------------------------------
+ * QoS2 PUB RX Message handling mechanism and associated house-keeping
+ *--------------------------------------------------------------------------*/
+static bool qos2_pub_rx_logup(struct client_ctx *cl_ctx, uint16_t msg_id)
+{
+ return qos2_pub_cq_logup(&CLIENT(cl_ctx)->qos2_rx_cq, msg_id);
+}
+
+static bool ack2_msg_id_logup(struct client_ctx *cl_ctx, uint16_t msg_id)
+{
+ return qos2_pub_cq_logup(&CLIENT(cl_ctx)->qos2_tx_cq, msg_id);
+}
+
+static bool qos2_pub_rx_unlog(struct client_ctx *cl_ctx, uint16_t msg_id)
+{
+ return qos2_pub_cq_unlog(&CLIENT(cl_ctx)->qos2_rx_cq, msg_id);
+}
+
+static bool ack2_msg_id_unlog(struct client_ctx *cl_ctx, uint16_t msg_id)
+{
+ struct client_desc *client = CLIENT(cl_ctx);
+ if(qos2_pub_cq_unlog(&client->qos2_tx_cq, msg_id)) {
+ struct mqtt_client_ctx_cbs *ctx_cbs = CTX_CBS_PTR(cl_ctx);
+ if(ctx_cbs->ack_notify)
+ ctx_cbs->ack_notify(client->app, MQTT_PUBCOMP,
+ msg_id, NULL, 0);
+ return true;
+ }
+
+ return false;
+}
+
+static bool qos2_pub_rx_is_done(struct client_ctx *cl_ctx, uint16_t msg_id)
+{
+ return qos2_pub_cq_check(&CLIENT(cl_ctx)->qos2_rx_cq, msg_id);
+}
+
+static bool awaits_pkts(struct client_ctx *cl_ctx)
+{
+
+ struct client_desc *client = CLIENT(cl_ctx);
+
+ return client->qos_ack1_wl.head ||
+ qos2_pub_cq_count(&client->qos2_rx_cq) ||
+ qos2_pub_cq_count(&client->qos2_tx_cq)?
+ true : false;
+}
+
+static inline int32_t len_err_free_mqp(struct mqtt_packet *mqp)
+{
+ mqp_free(mqp);
+ return MQP_ERR_PKT_LEN;
+}
+
+static int32_t is_valid_utf8_string(const struct utf8_string *utf8)
+{
+ /* Valid topic should be > 0 byte and must hosted in usable buffer */
+ return ((utf8->length > 0) && (NULL != utf8->buffer))? true : false;
+}
+
+#define RET_IF_INVALID_UTF8(utf8) \
+ if(false == is_valid_utf8_string(utf8)) \
+ return -1;
+
+static bool is_connected(struct client_ctx *cl_ctx)
+{
+
+ return (HAS_CONNECTION(cl_ctx) && !NEED_NET_CLOSE(cl_ctx))?
+ true : false;
+}
+
+uint16_t mqtt_client_new_msg_id()
+{
+ return assign_new_msg_id();
+}
+
+bool mqtt_client_is_connected(void *ctx)
+{
+ return is_connected(CL_CTX(ctx));
+}
+
+/*----------------------------------------------------------------------------
+ * MQTT TX Routines
+ *--------------------------------------------------------------------------*/
+static void used_ctxs_TO_sort(struct client_ctx *cl_ctx_TO)
+{
+ cl_ctx_remove(&used_ctxs, cl_ctx_TO);
+ cl_ctx_timeout_insert(&used_ctxs, cl_ctx_TO);
+}
+
+static inline int32_t net_send(int32_t net, const uint8_t *buf, uint32_t len, void *ctx)
+{
+
+ int32_t rv = net_ops->send(net, buf, len, ctx);
+ if(rv <= 0) {
+ memset(print_buf, 0x00, PRINT_BUF_LEN);
+ sprintf((char*) print_buf, "MQP_ERR_NETWORK %i\r\n", MQP_ERR_NETWORK);
+ Uart_Write((uint8_t *) print_buf);
+ rv = MQP_ERR_NETWORK;
+ }
+ return rv;
+}
+
+#if 0
+static int32_t cl_ctx_send(struct client_ctx *cl_ctx, const uint8_t *buf, uint32_t len,
+ bool is_conn_msg)
+{
+
+ int32_t rv = MQP_ERR_NOTCONN;
+
+ /* For CONNECT msg, a client context mustn't be already connected.
+ For others msgs, a client context must be conected to server */
+ if(false == (is_conn_msg ^ is_connected(cl_ctx)))
+ goto cl_ctx_send_exit1;
+
+ rv = net_send(cl_ctx->net, buf, len);
+ if(rv > 0) { /* A good send, do follow-up */
+ cl_ctx_timeout_update(cl_ctx, net_ops->time());
+ if(A_GROUP_MEMBER(cl_ctx) && HAS_CONNECTION(cl_ctx)) {
+ /* With update to timeout,
+ a sorting is impending */
+ used_ctxs_TO_sort(cl_ctx);
+ }
+
+ goto cl_ctx_send_exit1; /* A Good Send */
+ }
+
+ do_net_close_tx(cl_ctx, "snd-err");
+
+cl_ctx_send_exit1:
+ USR_INFO("C: FH-B1 0x%02x, len %u bytes, to net %d: %s\n\r",
+ *buf, len, cl_ctx->net, (rv > 0)? "Sent" : "Fail");
+ return rv;
+}
+#endif
+
+static int32_t cl_ctx_part_send(struct client_ctx *cl_ctx)
+{
+
+ struct tx_part_pkt *tx_part = &CLIENT(cl_ctx)->tx_part;
+ const uint8_t *buf = TX_PART_BUFFER(tx_part);
+ uint32_t len = TX_PART_BUF_SZ(tx_part);
+ uint32_t ofs = tx_part->offset;
+ uint8_t B1 = *buf;
+
+ int32_t rv = net_send(cl_ctx->net, buf, len, (void*)cl_ctx);
+ if(rv > 0) { /* Follow-up for a good send */
+ if(HAS_CONNECTION(cl_ctx)) {
+ /* Update TX timeout, if 'CTX' is connected */
+ cl_ctx_timeout_update(cl_ctx, net_ops->time());
+
+ /* After update, 'CTX'(s) sorting is a must */
+ if(A_GROUP_MEMBER(cl_ctx))
+ used_ctxs_TO_sort(cl_ctx);
+ }
+
+ if(rv != len)
+ /* Partial data was sent */
+ tx_part_addup(tx_part, rv);
+ else
+ tx_part_reset(tx_part);
+
+ goto cl_ctx_send_exit1; /* A Good Send */
+ }
+
+ do_net_close_tx(cl_ctx, "snd-err");
+
+cl_ctx_send_exit1:
+ USR_INFO("C: %s 0x%02x to net %d, %s (%d Bytes) [@ %u]\n\r",
+ ofs? "PartN" : "FH-B1", B1, cl_ctx->net,
+ (rv > 0)? "Sent" : "Fail", rv, net_ops->time());
+
+ return rv;
+}
+
+static int32_t cl_ctx_seg1_send(struct client_ctx *cl_ctx, const uint8_t *buf, uint32_t len,
+ bool is_conn_msg, struct mqtt_packet *tx_mqp)
+{
+
+ struct tx_part_pkt *tx_part = &CLIENT(cl_ctx)->tx_part;
+
+ /* For CONNECT msg, a client context mustn't be already connected.
+ For others msgs, a client context must be conected to server */
+ if(false == (is_conn_msg ^ is_connected(cl_ctx)))
+ return MQP_ERR_NOTCONN;
+
+ if(TX_PART_IN_USE(tx_part))
+ return MQP_ERR_BADCALL;
+
+ tx_part_setup(tx_part, buf, len, tx_mqp);
+
+ return cl_ctx_part_send(cl_ctx);
+}
+
+int32_t mqtt_client_send_progress(void *ctx)
+{
+ struct client_ctx *cl_ctx = CL_CTX(ctx);
+ struct tx_part_pkt *tx_part = NULL;
+ int32_t rv = MQP_ERR_BADCALL;
+
+ if(NULL == ctx)
+ return MQP_ERR_FNPARAM;
+
+ tx_part = &CLIENT(cl_ctx)->tx_part;
+
+ if(!TX_PART_IN_USE(tx_part))
+ return rv;
+
+ rv = cl_ctx_part_send(cl_ctx);
+ if(rv > 0)
+ rv = TX_PART_BUF_SZ(tx_part);
+
+ return rv;
+}
+
+static int32_t wr_connect_pl(struct client_ctx *cl_ctx, uint8_t *buf,
+ uint32_t fsz, uint8_t *conn_opts)
+{
+
+ /* UTF8 usage selection order: Client, W-Topic, W-Msg, User-Name, Pwd */
+ uint8_t utf8_sel[] = {0x00, WILL_CONFIG_VAL, 0x00,
+ USER_NAME_OPVAL, PASS_WORD_OPVAL
+ };
+ struct client_desc *client = CLIENT(cl_ctx);
+ uint8_t *ref = buf;
+
+ int32_t i = 0;
+
+ for(i = 0; i < 5; i++) { /* TBD 5 --> macro */
+ uint16_t len = 2;
+ const struct utf8_string *utf8 = client->conn_pl_utf8s[i];
+ if(NULL == utf8) {
+ /* UTF8 absent: Client ID (i = 0) and Will MSG (i = 2)
+ set zero byte length in the CONNECT message */
+ if(0 != i)
+ if(!((2 == i) && (*conn_opts & WILL_CONFIG_VAL)))
+ continue; /* Others, just pass */
+ } else {
+ len += utf8->length;
+ }
+
+ if(fsz < (buf - ref + len)) { /* TBD end = ref + fsz */
+ Uart_Write((uint8_t*)"Payload: no space left fail\r\n");
+ return MQP_ERR_PKT_LEN; /* Payload: no space left */
+ }
+ if(2 == len) {
+ buf += buf_wr_nbo_2B(buf, 0); /* WR 0 byte length */
+ } else {
+ buf += mqp_buf_wr_utf8(buf, utf8);
+ }
+ *conn_opts |= utf8_sel[i]; /* Enable message flags */
+ }
+
+ return buf - ref;
+}
+
+
+/* Define protocol information for the supported versions */
+static uint8_t mqtt310[] = {0x00, 0x06, 'M', 'Q', 'I', 's', 'd', 'p', 0x03};
+static uint8_t mqtt311[] = {0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04};
+
+static inline uint16_t get_connect_vh_len(struct client_ctx *cl_ctx)
+{
+
+ return (IS_PROTO_VER31(cl_ctx)? sizeof(mqtt310) : sizeof(mqtt311))
+ + 3;
+}
+
+static int32_t wr_connect_vh(struct client_ctx *cl_ctx, uint8_t *buf,
+ uint16_t ka_secs, uint8_t conn_opts)
+{
+
+ uint8_t *ref = buf;
+
+ if(IS_PROTO_VER31(cl_ctx))
+ buf += buf_wr_nbytes(buf, mqtt310, sizeof(mqtt310));
+ else
+ buf += buf_wr_nbytes(buf, mqtt311, sizeof(mqtt311));
+
+ *buf++ = conn_opts;
+ buf += buf_wr_nbo_2B(buf, ka_secs);
+
+ return buf - ref;
+}
+
+static int32_t net_connect(struct client_ctx *cl_ctx)
+{
+
+ struct client_desc *client = CLIENT(cl_ctx);
+
+ if(NEED_NET_CLOSE(cl_ctx)) {
+ Uart_Write((uint8_t*)"Return MQP_ERR_NOT_DEF\r\n");
+ return MQP_ERR_NOT_DEF;
+ }
+ if(NULL == net_ops) {
+ Uart_Write((uint8_t*)"Return MQP_ERR_NET_OPS\r\n");
+ return MQP_ERR_NET_OPS;
+ }
+ cl_ctx->net = net_ops->open(client->nwconn_opts | DEV_NETCONN_OPT_TCP,
+ client->server_addr,
+ client->port_number,
+ &client->nw_security);
+
+ return (-1 == cl_ctx->net)? MQP_ERR_NETWORK : 0;
+}
+
+static
+int32_t cl_ctx_conn_state_try_locked(struct client_ctx *cl_ctx, const uint8_t *buf,
+ uint32_t len, uint16_t ka_secs, bool clean_session,
+ struct mqtt_packet *tx_mqp)
+{
+
+ int32_t rv = 0;
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 100 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+
+ rv = net_connect(cl_ctx);
+ if(rv < 0) {
+ Uart_Write((uint8_t*)"net_connect failed\r\n");
+ goto cl_ctx_conn_state_try_locked_exit1;
+ }
+ /* Ensure LIB is initialized & CTX isn't awaiting CONNACK */
+ rv = MQP_ERR_BADCALL;
+ if(false == ((INIT_DONE_STATE != cl_lib_state) || (AWAITS_CONNACK(cl_ctx)))) {
+ rv = cl_ctx_seg1_send(cl_ctx, buf, len, true, tx_mqp);
+ }
+ if(rv < 0) {
+ Uart_Write((uint8_t*)"cl_ctx_seg1_send failed\r\n");
+ goto cl_ctx_conn_state_try_locked_exit1; /* Fail */
+ }
+ /* Successfully sent CONNECT msg - let's do housekeeping */
+ cl_ctx->timeout = net_ops->time();/* Fixup: CONN TX Time */
+ cl_ctx->flags |= DO_CONNACK_TO_FLAG | CONNACK_AWAIT_FLAG;
+ cl_ctx->flags |= clean_session? CLEAN_SESSION_FLAG : 0;
+
+ cl_ctx->ka_secs = ka_secs;
+
+ if(A_GROUP_MEMBER(cl_ctx)) {
+ cl_ctx->next = conn_ctxs;
+ conn_ctxs = cl_ctx;
+
+ /* First entry in 'conn_ctxs': schedule a move to
+ 'used_conn' (for house-keeping and tracking) */
+ if(NULL == cl_ctx->next) {
+ rv = loopb_trigger();
+ }
+ }
+
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ //xSemaphoreGive( xSemaphore );
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+cl_ctx_conn_state_try_locked_exit1:
+// MUTEX_UNLOCK();
+ xSemaphoreGive(xSemaphore);
+
+ return rv;
+}
+
+static
+int32_t connect_msg_send(struct client_ctx *cl_ctx, bool clean_session, uint16_t ka_secs)
+{
+
+ struct mqtt_packet *mqp = mqp_client_send_alloc(MQTT_CONNECT);
+ uint8_t *buf, *ref, conn_opts = clean_session? CLEAN_START_VAL : 0;
+ int32_t rv = MQP_ERR_PKT_LEN;
+ uint32_t fsz; /* Free buffer size in PKT */
+ uint16_t vhl = get_connect_vh_len(cl_ctx);
+
+ if(NULL == mqp) {
+ Uart_Write((uint8_t*)"MQP_ERR_PKT_AVL\r\n");
+ return MQP_ERR_PKT_AVL;
+ }
+ fsz = MQP_FREEBUF_LEN(mqp);
+ if(fsz < vhl) {
+ Uart_Write((uint8_t*)"No space for VAR HDR\r\n");
+ goto connect_msg_send_exit1; /* No space for VAR HDR */
+ }
+ mqp->vh_len = vhl; /* Reserve buffer for variable header */
+ buf = ref = MQP_PAYLOAD_BUF(mqp);/* Get started to incorporate payload */
+
+ rv = wr_connect_pl(cl_ctx, buf, fsz - vhl, &conn_opts);/* Payload data */
+ if(rv < 0) {
+ memset(print_buf, 0x00, PRINT_BUF_LEN);
+ sprintf((char*) print_buf, "Payload WR failed %i\r\n",rv);
+ Uart_Write((uint8_t *) print_buf);
+ goto connect_msg_send_exit1; /* Payload WR failed */
+ }
+ buf += rv;
+ mqp->pl_len = buf - ref;
+
+ wr_connect_vh(cl_ctx, ref - vhl, ka_secs,
+ CLIENT(cl_ctx)->will_opts | conn_opts); /* Var Header */
+
+ mqp_prep_fh(mqp, MAKE_FH_FLAGS(false, MQTT_QOS0, false));/* Fix Header */
+ ref = MQP_FHEADER_BUF(mqp);
+
+ /* Following routine frees up MQP - whether error or not */
+ return cl_ctx_conn_state_try_locked(cl_ctx, ref, buf - ref,
+ ka_secs, clean_session,
+ mqp);
+connect_msg_send_exit1:
+
+ if(mqp) {
+ mqp_free_locked(mqp);
+ }
+ return rv;
+}
+
+int32_t mqtt_connect_msg_send(void *ctx, bool clean_session, uint16_t ka_secs)
+{
+
+ return ctx?
+ connect_msg_send(CL_CTX(ctx), clean_session, ka_secs) : -1;
+}
+
+/*
+ To be used for the following messages: PUBLISH, SUBSCRIBE, UNSUBSCRIBE
+ Dispatches msg to broker over socket. Frees-up MQP, in case, MSG has QoS0 or
+ if client-lib allocated MQP encounters an error in dispatch.
+ Returns, on success, number of bytes transfered, otherwise -1
+*/
+static int32_t _msg_dispatch(struct client_ctx *cl_ctx, struct mqtt_packet *mqp,
+ enum mqtt_qos qos, bool retain)
+{
+
+ bool not_qos0 = (MQTT_QOS0 != qos)? true : false;
+ uint16_t msg_id = mqp->msg_id;
+ int32_t rv = MQP_ERR_NETWORK;
+
+ mqp_prep_fh(mqp, MAKE_FH_FLAGS(false, qos, retain));
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 200 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+
+ if(not_qos0) {
+ mqp->n_refs++; /* Need to enlist, do not free-up MQP */
+ }
+ /* Tries to free-up MQP either on error or if full pkt is sent */
+ rv = cl_ctx_seg1_send(cl_ctx, MQP_FHEADER_BUF(mqp),
+ MQP_CONTENT_LEN(mqp), false,
+ mqp);
+
+ /* At this point, error or not, QoS0 MQP would have been freed */
+
+ if((rv <= 0) && not_qos0) {
+ mqp_free(mqp); /* Err: Explicitly free-up non QoS0 MQP */
+ Uart_Write((uint8_t*)"cl_ctx_seg1_send failed\r\n");
+ goto _msg_dispatch_exit1;
+ }
+
+ rv = msg_id; /* Make progress for a good send to the server */
+
+ if(not_qos0) { /* Enlist non QOS0 MQP to await ACK from server */
+ mqp_ack_wlist_append(&CLIENT(cl_ctx)->qos_ack1_wl, mqp);
+ }
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ //xSemaphoreGive( xSemaphore );
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+_msg_dispatch_exit1:
+
+// MUTEX_UNLOCK();
+ xSemaphoreGive(xSemaphore);
+
+ return rv;
+}
+
+static
+int32_t msg_dispatch_no_free(struct client_ctx *cl_ctx, struct mqtt_packet *mqp,
+ enum mqtt_qos qos, bool retain)
+{
+ if((NULL == mqp) || (NULL == cl_ctx))
+ return MQP_ERR_FNPARAM;
+
+ mqp->n_refs++; /* Ensures caller that MQP is not freed-up */
+
+ return _msg_dispatch(cl_ctx, mqp, qos, retain);
+}
+
+int32_t mqtt_client_pub_msg_send(void *ctx, const struct utf8_string *topic,
+ const uint8_t *data_buf, uint32_t data_len,
+ enum mqtt_qos qos, bool retain)
+{
+
+ struct mqtt_packet *mqp = NULL;
+
+ if((NULL == ctx) ||
+ (NULL == topic) ||
+ ((data_len > 0) && (NULL == data_buf))) {
+ Uart_Write((uint8_t*)"MQP_ERR_FNPARAM\n\r");
+ return MQP_ERR_FNPARAM;
+ }
+ if(false == is_valid_utf8_string(topic)) {
+ Uart_Write((uint8_t*)"MQP_ERR_CONTENT\n\r");
+ return MQP_ERR_CONTENT;
+ }
+ mqp = mqp_client_send_alloc(MQTT_PUBLISH);
+ if(NULL == mqp) {
+ Uart_Write((uint8_t*)"MQP_ERR_PKT_AVL\n\r");
+ return MQP_ERR_PKT_AVL;
+ }
+ if((0 > mqp_pub_append_topic(mqp, topic, qos? assign_new_msg_id(): 0)) ||
+ (data_len && (0 > mqp_pub_append_data(mqp, data_buf, data_len)))) {
+ Uart_Write((uint8_t*)"len_err\n\r");
+ return len_err_free_mqp(mqp);
+ }
+
+ return _msg_dispatch(CL_CTX(ctx), mqp, qos, retain);
+}
+
+int32_t mqtt_client_pub_dispatch(void *ctx, struct mqtt_packet *mqp,
+ enum mqtt_qos qos, bool retain)
+{
+ return msg_dispatch_no_free(CL_CTX(ctx), mqp, qos, retain);
+}
+
+static int32_t tail_incorp_msg_id(struct mqtt_packet *mqp)
+{
+ uint8_t *buf = MQP_FHEADER_BUF(mqp) + mqp->vh_len;
+
+ if(0 == mqp->msg_id) {
+ mqp->msg_id = assign_new_msg_id();
+ buf += buf_wr_nbo_2B(buf, mqp->msg_id);
+ mqp->vh_len += 2;
+
+ return 2;
+ }
+
+ return 0;
+}
+
+static int32_t buf_utf8_wr_try(uint8_t *buf, uint32_t fsz, const struct utf8_string *topic,
+ uint8_t qid)
+{
+ uint8_t *ref = buf;
+
+ if(fsz < (topic->length + 2 + (QFL_VALUE == qid)? 0 : 1))
+ return MQP_ERR_PKT_LEN; /* No buf */
+
+ if(false == is_valid_utf8_string(topic))
+ return MQP_ERR_CONTENT;/* Invalid */
+
+ buf += mqp_buf_wr_utf8(buf, topic);
+ if(QFL_VALUE != qid)
+ *buf++ = qid;
+
+ return buf - ref;
+}
+
+static int32_t utf8_array_send(struct client_ctx *cl_ctx,
+ const struct utf8_strqos *subsc_vec,
+ const struct utf8_string *unsub_vec,
+ uint32_t n_elem)
+{
+ struct mqtt_packet *mqp;
+ uint8_t *ref, *buf, *end;
+ uint32_t i;
+
+ if((NULL == cl_ctx) || !((!!subsc_vec) ^ (!!unsub_vec)) || (0 == n_elem))
+ return MQP_ERR_FNPARAM;
+
+ mqp = mqp_client_send_alloc(subsc_vec?
+ MQTT_SUBSCRIBE : MQTT_UNSUBSCRIBE);
+ if(NULL == mqp)
+ return MQP_ERR_PKT_AVL;
+
+ buf = MQP_VHEADER_BUF(mqp);
+ end = MQP_FREEBUF_LEN(mqp) + buf; /* End of free buffer */
+ if((end - buf) < 2)
+ return len_err_free_mqp(mqp);/* MSG-ID: no space */
+
+ buf += tail_incorp_msg_id(mqp);
+ ref = buf;
+
+ for(i = 0; i < n_elem; i++) {
+ const struct utf8_string *topic;
+ struct utf8_string topreq;
+ int32_t rv;
+
+ if(subsc_vec) {
+ topreq.length = subsc_vec[i].length;
+ topreq.buffer = subsc_vec[i].buffer;
+ topic = &topreq;
+ } else
+ topic = unsub_vec + i;
+
+ rv = buf_utf8_wr_try(buf, end - buf, topic, subsc_vec?
+ (uint8_t)subsc_vec[i].qosreq : QFL_VALUE);
+ if(rv < 0) {
+ mqp_free(mqp);
+ return rv;
+ }
+
+ buf += rv;
+ }
+
+ mqp->pl_len = buf - ref; /* Total length of topics data */
+
+ return _msg_dispatch(cl_ctx, mqp, MQTT_QOS1, false);
+}
+
+int32_t mqtt_sub_msg_send(void *ctx, const struct utf8_strqos *qos_topics, uint32_t count)
+{
+ return utf8_array_send(CL_CTX(ctx), qos_topics, NULL, count);
+}
+
+int32_t mqtt_sub_dispatch(void *ctx, struct mqtt_packet *mqp)
+{
+ return msg_dispatch_no_free(CL_CTX(ctx), mqp, MQTT_QOS1, false);
+}
+
+int32_t mqtt_unsub_msg_send(void *ctx, const struct utf8_string *topics, uint32_t count)
+{
+ return utf8_array_send(CL_CTX(ctx), NULL, topics, count);
+}
+
+int32_t mqtt_unsub_dispatch(void *ctx, struct mqtt_packet *mqp)
+{
+ return msg_dispatch_no_free(CL_CTX(ctx), mqp, MQTT_QOS1, false);
+}
+
+/* Note: in this revision of implementation, vh_msg_send() is being invoked
+ from a locked RX context. Should this situation change, so should the
+ 'locking' considerations in the routine. */
+static int32_t vh_msg_send(struct client_ctx *cl_ctx, uint8_t msg_type,
+ enum mqtt_qos qos, bool has_vh,
+ uint16_t vh_data)
+{
+ uint8_t buf[4];
+ uint32_t len = 2;
+
+ buf[0] = MAKE_FH_BYTE1(msg_type, MAKE_FH_FLAGS(false, qos, false));
+ buf[1] = has_vh ? 2 : 0;
+
+ if(has_vh)
+ len += buf_wr_nbo_2B(buf + 2, vh_data);
+
+ return cl_ctx_seg1_send(cl_ctx, buf, len, false, NULL);
+}
+
+static int32_t pingreq_send(struct client_ctx *cl_ctx, uint32_t rsp_flag)
+{
+
+ int32_t rv = 0;
+ uint8_t buf[2];
+
+ buf[0] = MAKE_FH_BYTE1(MQTT_PINGREQ,
+ MAKE_FH_FLAGS(false, MQTT_QOS0, false));
+ buf[1] = 0;
+
+ /* Note: in case of error in network send, cl_ctx_send() may
+ try to terminate connection with server. */
+ rv = cl_ctx_seg1_send(cl_ctx, buf, 2, false, NULL);
+ if(rv > 0)
+ cl_ctx->flags |= rsp_flag;
+
+ return rv;
+}
+
+int32_t mqtt_pingreq_send(void *ctx)
+{
+ Uart_Write((uint8_t*)"mqtt_pingreq_send\r\n");
+ int32_t rv = 0;
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 40 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+ rv = pingreq_send(CL_CTX(ctx), USER_PING_RSP_FLAG);
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ xSemaphoreGive( xSemaphore );
+
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+// MUTEX_UNLOCK();
+ return rv;
+}
+
+int32_t mqtt_disconn_send(void *ctx)
+{
+ Uart_Write((uint8_t*)"mqtt_disconn_send\r\n");
+ uint8_t buf[2];
+
+ buf[0] = MAKE_FH_BYTE1(MQTT_DISCONNECT,
+ MAKE_FH_FLAGS(false, MQTT_QOS0, false));
+ buf[1] = 0;
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 40 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+ /* Note: in case of error in network send, cl_ctx_send() may
+ try to terminate connection with server. */
+ if(cl_ctx_seg1_send(CL_CTX(ctx), buf, 2, false, NULL) > 0) {
+ /* Terminate connection on application's request */
+ do_net_close_tx(CL_CTX(ctx), "DISCONN");
+ }
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ xSemaphoreGive( xSemaphore );
+
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+// MUTEX_UNLOCK();
+ return 0;
+}
+
+/*------------------------------------------------------------------------------
+ * MQTT RX Routines
+ *------------------------------------------------------------------------------
+ */
+static bool ack1_wl_mqp_dispatch(struct client_ctx *cl_ctx)
+{
+ struct mqtt_ack_wlist *wlist = &CLIENT(cl_ctx)->qos_ack1_wl;
+ struct mqtt_packet *mqp = NULL;
+ bool rv = true;
+
+ for(mqp = wlist->head; mqp && (true == rv); mqp = mqp->next) {
+ uint8_t *buf = MQP_FHEADER_BUF(mqp);
+ mqp->fh_byte1 = *buf |= DUP_FLAG_VAL(true);
+
+ mqp->n_refs++; /* Ensures MQP is not freed by following */
+
+ /* Error or not, following routine tries to free up MQP */
+ if(cl_ctx_seg1_send(cl_ctx, buf, MQP_CONTENT_LEN(mqp),
+ false, mqp) <= 0)
+ rv = false;
+ }
+
+ return rv;
+}
+
+/* TBD candidate for common */
+static bool ack2_msg_id_dispatch(struct client_ctx *cl_ctx)
+{
+ struct pub_qos2_cq *tx_cq = &CLIENT(cl_ctx)->qos2_tx_cq;
+ uint8_t rd_idx = tx_cq->rd_idx;
+ uint8_t n_free = tx_cq->n_free;
+ bool rv = true;
+ uint8_t i = 0;
+
+ for(i = rd_idx; i < (MAX_PUBREL_INFLT - n_free) && (true == rv); i++) {
+ if(vh_msg_send(cl_ctx, MQTT_PUBREL, MQTT_QOS1,
+ true, tx_cq->id_vec[i]) <= 0)
+ rv = false;
+ }
+
+ return rv;
+}
+
+static void session_resume(struct client_ctx *cl_ctx)
+{
+ DBG_INFO("C: Re-send ACK awaited QoS1/2 msgs to net %d\n\r",
+ cl_ctx->net);
+
+ if(ack1_wl_mqp_dispatch(cl_ctx))
+ ack2_msg_id_dispatch(cl_ctx);
+
+ return;
+}
+
+static bool ack1_wl_rmfree(struct mqtt_ack_wlist *wl, uint16_t msg_id)
+{
+ struct mqtt_packet *mqp = mqp_ack_wlist_remove(wl, msg_id);
+ if(NULL != mqp) {
+ mqp_free(mqp);
+ return true;
+ }
+
+ USR_INFO("Err: Unexpected ACK w/ ID 0x%04x\n\r", msg_id);
+
+ return false;
+}
+
+static bool _proc_pub_rec_rx(struct client_ctx *cl_ctx, uint16_t msg_id)
+{
+ /* Follow-up messages for QOS2 PUB must be transacted in the
+ same order as the initial sequence of QOS2 PUB dispatches.
+ Therefore, checking the first entry should be OK
+ */
+ struct mqtt_packet *mqp = CLIENT(cl_ctx)->qos_ack1_wl.head;
+
+ if((msg_id == mqp->msg_id) && ack2_msg_id_logup(cl_ctx, msg_id)) {
+
+ ack1_wl_rmfree(&CLIENT(cl_ctx)->qos_ack1_wl, msg_id);
+
+ vh_msg_send(cl_ctx, MQTT_PUBREL, MQTT_QOS1,
+ true, msg_id);
+
+ return true;
+ }
+
+ return false; /* Unexpected PUBREC or QOS2 store exceeded */
+}
+
+static bool _proc_pub_rel_rx(struct client_ctx *cl_ctx, uint16_t msg_id)
+{
+ /* For a PUB-REL RX, send PUBCOMP to let server make progress */
+ vh_msg_send(cl_ctx, MQTT_PUBCOMP, MQTT_QOS0, true, msg_id);
+
+ if(qos2_pub_rx_is_done(cl_ctx, msg_id))
+ qos2_pub_rx_unlog(cl_ctx, msg_id); /* Expunge record */
+
+ return true;
+}
+
+/*
+ Process ACK Message from Broker.
+ Returns true on success, otherwise false.
+ Used for: PUBACK, SUBACK and UNSUBACK
+*/
+static
+bool _proc_ack_msg_rx(struct client_ctx *cl_ctx, struct mqtt_packet *mqp_raw)
+{
+ struct mqtt_client_ctx_cbs *ctx_cbs = CTX_CBS_PTR(cl_ctx);
+ struct client_desc *client = CLIENT(cl_ctx);
+ uint16_t msg_id = mqp_raw->msg_id;
+ uint32_t len = mqp_raw->pl_len;
+
+ /* Caters to SUB-ACK, UNSUB-ACK and PUB-ACK Messages */
+ if(false == ack1_wl_rmfree(&client->qos_ack1_wl, msg_id))
+ return false; /* Err: MSG_ID was not awaited */
+
+ if(ctx_cbs->ack_notify)
+ ctx_cbs->ack_notify(client->app, mqp_raw->msg_type, msg_id,
+ len? MQP_PAYLOAD_BUF(mqp_raw): NULL,
+ len);
+ return true;
+}
+
+static
+bool proc_ack_msg_rx(struct client_ctx *cl_ctx, struct mqtt_packet *mqp_raw)
+{
+ uint8_t msg_type = mqp_raw->msg_type;
+ bool rv = false;
+ uint16_t msg_id = 0;
+
+ if(false == mqp_proc_msg_id_ack_rx(mqp_raw, MQTT_SUBACK == msg_type))
+ return rv; /* Problem in contents received from server */
+
+ msg_id = mqp_raw->msg_id;
+
+ if(MQTT_PUBREC == msg_type) {
+ rv = _proc_pub_rec_rx(cl_ctx, msg_id);
+ if(rv)
+ MQP_RX_DO_NOT_RPT_SET(mqp_raw); /* Don't report to App */
+
+ } else if(MQTT_PUBREL == msg_type) {
+ rv = _proc_pub_rel_rx(cl_ctx, msg_id);
+ if(rv)
+ MQP_RX_DO_NOT_RPT_SET(mqp_raw); /* Don't report to App */
+ } else if(MQTT_PUBCOMP == msg_type) {
+ rv = ack2_msg_id_unlog(cl_ctx, msg_id);
+ } else {
+ rv = _proc_ack_msg_rx(cl_ctx, mqp_raw);
+ }
+
+ return rv;
+}
+
+static
+bool proc_pub_msg_rx(struct client_ctx *cl_ctx, struct mqtt_packet *mqp_raw)
+{
+ struct mqtt_client_ctx_cbs *ctx_cbs = CTX_CBS_PTR(cl_ctx);
+ bool good_pub = mqp_proc_pub_rx(mqp_raw);
+ uint8_t B = mqp_raw->fh_byte1;
+ enum mqtt_qos qos = ENUM_QOS(B);
+ uint16_t msg_id = 0;
+
+ if(false == good_pub)
+ return false; /* Didn't get nicely composed PUB Packet */
+
+ msg_id = mqp_raw->msg_id;
+
+ /* Irrespective of the result of the ACK through vh_msg_send(),
+ the implementation has chosen to process the good PUB packet.
+ Any error will be handled in next iteration of rx processing.
+ */
+ if(MQTT_QOS1 == qos)
+ vh_msg_send(cl_ctx, MQTT_PUBACK, MQTT_QOS0, true, msg_id);
+
+ if(MQTT_QOS2 == qos) {
+ /* Ensuring "only once" philosophy for MQTT QoS2 PUBs */
+ if(qos2_pub_rx_is_done(cl_ctx, msg_id)) {
+ /* Already delivered. Drop it & do not report */
+ MQP_RX_DO_NOT_RPT_SET(mqp_raw);
+ return true; /* No more follow-up; all's good */
+ }
+
+ if(false == qos2_pub_rx_logup(cl_ctx, msg_id))
+ return false; /* Failed to record New RX PUB */
+
+ vh_msg_send(cl_ctx, MQTT_PUBREC, MQTT_QOS0, true, msg_id);
+ }
+
+ /* QoS obliations completed, present PUBLISH RX packet to app */
+ if(ctx_cbs->publish_rx) {
+ /* App has chosen the callback method to receive PKT */
+ mqp_raw->n_refs++; /* Make app owner of this packet */
+ if(ctx_cbs->publish_rx(CLIENT(cl_ctx)->app, BOOL_DUP(B),
+ qos, BOOL_RETAIN(B), mqp_raw)) {
+ /* App has no use of PKT any more, so free it */
+ mqp_raw->n_refs--; /* Take back ownership */
+ }
+ }
+
+ return true;
+}
+
+static
+bool proc_connack_rx(struct client_ctx *cl_ctx, struct mqtt_packet *mqp_raw)
+{
+ struct mqtt_client_ctx_cbs *ctx_cbs = CTX_CBS_PTR(cl_ctx);
+ uint8_t *buf = MQP_VHEADER_BUF(mqp_raw);
+
+ mqp_raw->vh_len += 2;
+ mqp_raw->pl_len -= 2;
+
+ if(0 != mqp_raw->pl_len)
+ return false; /* There is no payload in message */
+
+ cl_ctx->flags &= ~(DO_CONNACK_TO_FLAG | CONNACK_AWAIT_FLAG);
+
+ if(VHB_CONNACK_RC(buf))
+ /* Server has refused the connection, inform the app */
+ goto proc_connack_rx_exit1;
+
+ cl_ctx->flags |= NOW_CONNECTED_FLAG;
+ cl_ctx_timeout_update(cl_ctx, net_ops->time()); /* start KA */
+
+ if(IS_CLN_SESSION(cl_ctx))
+ session_delete(cl_ctx);
+ else
+ session_resume(cl_ctx);
+
+proc_connack_rx_exit1:
+ if(ctx_cbs->ack_notify)
+ ctx_cbs->ack_notify(CLIENT(cl_ctx)->app, mqp_raw->msg_type,
+ 0, buf, 2);
+
+ return true;
+}
+
+static
+bool proc_pingrsp_rx(struct client_ctx *cl_ctx, struct mqtt_packet *mqp_raw)
+{
+ struct mqtt_client_ctx_cbs *ctx_cbs = CTX_CBS_PTR(cl_ctx);
+
+ if(0 != mqp_raw->pl_len)
+ return false;
+
+ if(AWAITS_KA_PING(cl_ctx)) {
+ cl_ctx->flags &= ~KA_PINGER_RSP_FLAG;
+ return true;
+ }
+
+ if(AWAITS_PINGRSP(cl_ctx)) {
+ cl_ctx->flags &= ~USER_PING_RSP_FLAG;
+ if(ctx_cbs->ack_notify)
+ ctx_cbs->ack_notify(CLIENT(cl_ctx)->app,
+ mqp_raw->msg_type,
+ 0, NULL, 0);
+ return true;
+ }
+
+ return false;
+}
+
+static
+bool conn_sent_state_rx(struct client_ctx *cl_ctx, struct mqtt_packet *mqp_raw)
+{
+ bool rv = false;
+
+ switch(mqp_raw->msg_type) {
+
+ case MQTT_CONNACK:
+ /* Changes client_ctx->flags to CONNECTED */
+ rv = proc_connack_rx(cl_ctx, mqp_raw);
+ break;
+
+ default:
+ break;
+ }
+
+ return rv;
+}
+
+static
+bool connected_state_rx(struct client_ctx *cl_ctx, struct mqtt_packet *mqp_raw)
+{
+ bool rv = false;
+
+ switch(mqp_raw->msg_type) {
+
+ case MQTT_SUBACK:
+ case MQTT_PUBACK:
+ case MQTT_PUBREC:
+ case MQTT_PUBREL:
+ case MQTT_PUBCOMP:
+ case MQTT_UNSUBACK:
+ rv = proc_ack_msg_rx(cl_ctx, mqp_raw);
+ break;
+
+ case MQTT_PINGRSP:
+ rv = proc_pingrsp_rx(cl_ctx, mqp_raw);
+ break;
+
+ case MQTT_PUBLISH:
+ rv = proc_pub_msg_rx(cl_ctx, mqp_raw);
+ break;
+
+ case MQTT_CONNACK: /* not expected */
+ default:
+ break;
+ }
+
+ return rv;
+}
+
+static bool process_recv(struct client_ctx *cl_ctx,
+ struct mqtt_packet *mqp_raw)
+{
+ bool rv;
+
+ USR_INFO("C: Rcvd msg Fix-Hdr (Byte1) 0x%02x from net %d [@ %u]\n\r",
+ mqp_raw->fh_byte1, cl_ctx->net, net_ops->time());
+
+ /* Working Principle: Only RX processing errors should be
+ reported as 'false'. Status of TX as a follow-up to RX
+ messages need not be reported by the xyz_rx() routines.
+ Error observed in TX is either dealt in next iteration
+ of RX loop (in case, there is a dedicated RX task for
+ the CTX) or in TX routine itself (in case, there is no
+ dedicated RX task for the CTX).
+ */
+ rv = AWAITS_CONNACK(cl_ctx)?
+ conn_sent_state_rx(cl_ctx, mqp_raw) :
+ connected_state_rx(cl_ctx, mqp_raw);
+
+ DBG_INFO("C: Msg w/ ID 0x%04x, processing status: %s\n\r",
+ mqp_raw->msg_id, rv? "Good" : "Fail");
+
+ return rv;
+}
+
+static int32_t net_recv(int32_t net, struct mqtt_packet *mqp, uint32_t wait_secs, void *ctx)
+{
+ bool timed_out = false;
+ int32_t rv = mqp_recv(net, net_ops, mqp, wait_secs, &timed_out, ctx);
+ if(rv <= 0) {
+ USR_INFO("C: Net %d, Raw Error %d, Time Out: %c\n\r",
+ net, rv, timed_out? 'Y' : 'N');
+
+ if(timed_out)
+ rv = MQP_ERR_TIMEOUT;
+ }
+
+ return rv;
+}
+
+/*
+ MQTT 3.1.1 implementation
+ -------------------------
+
+ Keep Alive Time is maxmimum interval within which a client should send a
+ packet to broker. If there are either no packets to be sent to broker or
+ no retries left, then client is expected to a send a PINGREQ within Keep
+ Alive Time. Broker should respond by sending PINGRSP with-in reasonable
+ time of 'wait_secs'. If Keep Alive Time is set as 0, then client is not
+ expected to be disconnected due to in-activity of MQTT messages. Value
+ of 'wait_secs' is assumed to be quite smaller than (non-zero) 'ka_secs'.
+*/
+static void conn2used_ctxs(uint32_t wait_secs)
+{
+
+ while(conn_ctxs) {
+ struct client_ctx *cl_ctx = conn_ctxs;
+ conn_ctxs = conn_ctxs->next;
+
+ cl_ctx_timeout_insert(&used_ctxs, cl_ctx);
+ }
+}
+
+static int32_t single_ctx_ka_sequence(struct client_ctx *cl_ctx, uint32_t wait_secs)
+{
+
+ uint32_t now_secs = net_ops->time();
+
+ if(AWAITS_CONNACK(cl_ctx) && CFG_CONNACK_TO(cl_ctx)) {
+ cl_ctx->timeout += wait_secs; /* Set CONNACK timeout value */
+ cl_ctx->flags &= ~DO_CONNACK_TO_FLAG;
+ }
+
+ if(cl_ctx->timeout > now_secs) {
+ return 1; /* Still have time for next message transaction */
+ }
+ if(is_connected(cl_ctx)) {
+ /* Timeout has happened. Check for PINGRESP if PINGREQ done.
+ Otherwise, send PINGREQ (Are You there?) to the server. */
+ if(AWAITS_KA_PING(cl_ctx)) {
+ goto single_ctx_ka_sequence_exit1; /* No PINGRESP */
+ }
+ return pingreq_send(cl_ctx, KA_PINGER_RSP_FLAG); /* Hello! */
+ }
+
+single_ctx_ka_sequence_exit1:
+
+ USR_INFO("C: Net %d, no RX MSG in reasonable time\n\r", cl_ctx->net);
+ return -1;
+}
+
+static uint32_t single_ctx_adj_wait_secs_get(struct client_ctx *cl_ctx, uint32_t wait_secs)
+{
+
+ return (KA_TIMEOUT_NONE != cl_ctx->timeout)?
+ MIN(cl_ctx->timeout - net_ops->time(), wait_secs) : wait_secs;
+}
+
+static int32_t single_ctx_rx_prep(struct client_ctx *cl_ctx, uint32_t *secs2wait)
+{
+
+ int32_t rv;
+
+ if(-1 == cl_ctx->net)
+ return MQP_ERR_NOTCONN; /* Likely for a ctx w/o a receive task */
+
+ if(NEED_NET_CLOSE(cl_ctx))
+ rv = MQP_ERR_NOTCONN;
+ else if(0 > single_ctx_ka_sequence(cl_ctx, *secs2wait))
+ rv = MQP_ERR_NETWORK;
+ else {
+ *secs2wait = single_ctx_adj_wait_secs_get(cl_ctx, *secs2wait);
+ return 1;
+ }
+
+ do_net_close_rx(cl_ctx, rv);
+ return rv;
+}
+
+static
+int32_t proc_ctx_data_recv(struct client_ctx *cl_ctx, struct mqtt_packet *mqp,
+ uint32_t wait_secs, void **app)
+{
+
+ int32_t rv = MQP_ERR_NOTCONN;
+ int32_t net = cl_ctx->net;
+
+ *app = cl_ctx->usr;
+
+ rv = net_recv(net, mqp, wait_secs, (void*)cl_ctx);
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 40 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+ if(rv > 0) {
+ if(false == process_recv(cl_ctx, mqp)) {
+ Uart_Write((uint8_t*)"MQP_ERR_CONTENT\r\n");
+ rv = MQP_ERR_CONTENT;
+ }
+ }
+ /* RX: close the network connection to the server for this context, if
+ (a) there is a processing / protocol error other than time-out
+ (b) A good MQTT CONNACK has a return code - connection refused
+ */
+ if(((rv < 0) && (rv != MQP_ERR_TIMEOUT)) ||
+ ((MQTT_CONNACK == mqp->msg_type) &&
+ MQP_CONNACK_RC(mqp))) {
+ do_net_close_rx(cl_ctx, rv);
+ }
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ xSemaphoreGive( xSemaphore );
+
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+// MUTEX_UNLOCK();
+ return rv;
+}
+
+static int32_t mqp_setup_proc_ctx_data_recv(struct client_ctx *cl_ctx,
+ struct mqtt_packet *mqp,
+ uint32_t wait_secs, void **app)
+{
+ struct mqtt_packet *rx_mqp = CLIENT(cl_ctx)->rx_mqp;
+ int32_t rv;
+
+ if(NULL != mqp) {
+ /* Input MQP must be same as MQP for partial RX, if any */
+ if(rx_mqp) {
+ if(mqp != rx_mqp)
+ return MQP_ERR_FNPARAM;
+ } else
+ mqp_reset(mqp);
+ }
+
+ if(NULL == mqp) {
+ mqp = rx_mqp? rx_mqp : mqp_client_recv_alloc(0);
+ if(NULL == mqp)
+ return MQP_ERR_PKT_AVL;
+ }
+
+ rv = proc_ctx_data_recv(cl_ctx, mqp, wait_secs, app);
+ if(rv == MQP_ERR_TIMEOUT) {
+ CLIENT(cl_ctx)->rx_mqp = mqp; /* Save partial RX MQP */
+ } else {
+ /* Control reaches here due to either an error in RX or the
+ completion of RX. In both the cases, the MQP needs to be
+ detached and processed. For completion of RX:
+ callback mode: Application has used up MQP data; free it
+ Non-callback mode: Application will now use complete MQP
+ */
+ CLIENT(cl_ctx)->rx_mqp = NULL;
+ if(mqp->free)
+ mqp_free_locked(mqp); /* For only callback mode */
+ }
+
+ return rv;
+}
+
+static int32_t cl_ctx_recv(struct client_ctx *cl_ctx, struct mqtt_packet *mqp,
+ uint32_t wait_secs)
+{
+
+ void *app = NULL;
+ int32_t rv = 0;
+
+ do {
+ if(mqp && (NULL == CLIENT(cl_ctx)->rx_mqp))
+ mqp_reset(mqp);
+
+ rv = single_ctx_rx_prep(cl_ctx, &wait_secs);
+ if(rv > 0)
+ rv = mqp_setup_proc_ctx_data_recv(cl_ctx, mqp,
+ wait_secs,
+ &app);
+
+ /* 'mqp' must be valid, if rv > 0 but including further
+ & additional check for sake of static cod eanalysis.*/
+ } while((rv > 0) && mqp && MQP_RX_DO_NOT_RPT_COR(mqp));
+
+ return rv;
+}
+
+int32_t mqtt_client_ctx_await_msg(void *ctx, uint8_t msg_type, struct mqtt_packet *mqp,
+ uint32_t wait_secs)
+{
+ struct client_ctx *cl_ctx = CL_CTX(ctx);
+ int32_t rv = -1;
+
+ if((NULL == cl_ctx) || (NULL == mqp))
+ return MQP_ERR_FNPARAM;
+
+ do {
+ rv = cl_ctx_recv(cl_ctx, mqp, wait_secs);
+
+ } while((rv > 0) &&
+ (0 != msg_type) && (msg_type != mqp->msg_type));
+
+ return rv;
+}
+
+int32_t mqtt_client_ctx_run(void *ctx, uint32_t wait_secs)
+{
+
+ int32_t rv;
+
+ if(NULL == ctx)
+ return MQP_ERR_FNPARAM;
+
+ do {
+ rv = cl_ctx_recv(CL_CTX(ctx), NULL, wait_secs);
+
+ } while(rv > 0);
+
+ return rv;
+}
+
+static struct client_ctx *group_ctxs_ka_sequence(uint32_t wait_secs) {
+
+ struct client_ctx *cl_ctx = used_ctxs;
+
+ while(cl_ctx) {
+ struct client_ctx *next = cl_ctx->next;
+ if(single_ctx_rx_prep(cl_ctx, &wait_secs) < 0) {
+ /* 'CTX' no more eligible for operation
+ and has been removed from used_list */
+ if(false == grp_has_cbfn)
+ return cl_ctx;
+ }
+
+ cl_ctx = next;
+ }
+
+ return NULL;
+}
+
+#define IO_MON_NO_TIMEOUT (0xFFFFFFFF)
+
+static uint32_t group_ctxs_adj_wait_secs_get(uint32_t wait_secs)
+{
+
+ return used_ctxs?
+ single_ctx_adj_wait_secs_get(used_ctxs, wait_secs) : wait_secs;
+}
+
+static int32_t recv_hvec[MAX_NWCONN + 1 + 1]; /* GROUP LISTEN PORT + VEC END */
+static int32_t send_hvec = -1;
+static int32_t rsvd_hvec = -1;
+
+/* Caller must ensure atomic enviroment for execution of this routine */
+static void recv_hvec_load(int32_t *hvec_recv, uint32_t size, struct client_ctx *list)
+{
+
+ int32_t i = 0;
+
+ for(i = 0; (i < size) && (NULL != list); i++, list = list->next)
+ hvec_recv[i] = list->net;
+
+ hvec_recv[i] = -1;
+
+ return;
+}
+
+static int32_t group_ctxs_rx_prep(uint32_t wait_secs, void **app)
+{
+
+ /* CHK 'used ctx'(s) have live connection w/ server. If not, drop it */
+ struct client_ctx *ctx_kaTO = group_ctxs_ka_sequence(wait_secs);
+ int32_t n_hnds;
+
+ if(ctx_kaTO) {
+ *app = CLIENT(ctx_kaTO)->app;
+ return MQP_ERR_NETWORK;
+ }
+
+ conn2used_ctxs(wait_secs); /* Now, add new 'ctx'(s) to 'used ctxs' */
+
+ recv_hvec[0] = loopb_net;
+ recv_hvec_load(&recv_hvec[1], MAX_NWCONN + 1, used_ctxs);
+
+ wait_secs = group_ctxs_adj_wait_secs_get(wait_secs);
+
+ n_hnds = net_ops->io_mon(recv_hvec, &send_hvec,
+ &rsvd_hvec, wait_secs);
+ if(0 == n_hnds)
+ n_hnds = MQP_ERR_TIMEOUT;
+ else if(n_hnds < 0)
+ n_hnds = MQP_ERR_LIBQUIT;
+
+ return n_hnds;
+}
+
+static int32_t proc_loopback_recv(int32_t net)
+{
+
+ int32_t rv = 0;
+ uint8_t buf[LOOP_DLEN];
+
+ /* Thanks for waking-up thread, but ain't got much to do now */
+ rv = net_ops->recv_from(net, buf, LOOP_DLEN, NULL, NULL, 0);
+ if(rv <= 0) {
+ memset(print_buf, 0x00, PRINT_BUF_LEN);
+ sprintf((char*) print_buf, "MQP_ERR_LIBQUIT %i\r\n",MQP_ERR_LIBQUIT);
+ Uart_Write((uint8_t *) print_buf);
+ net_ops->close(net);
+ return MQP_ERR_LIBQUIT;
+ }
+
+ return rv;
+}
+
+static struct client_ctx *net_cl_ctx_find(int32_t net) {
+ struct client_ctx *cl_ctx = used_ctxs;
+
+ while(cl_ctx && (net != cl_ctx->net))
+ cl_ctx = cl_ctx->next;
+
+ return cl_ctx;
+}
+
+static int32_t proc_net_data_recv(int32_t net, struct mqtt_packet *mqp, void **app)
+{
+
+ /* Note: used_ctxs are always managed by a single RX task */
+ struct client_ctx *cl_ctx = net_cl_ctx_find(net);
+ int32_t rv = MQP_ERR_NOTCONN;
+
+ if(NULL == cl_ctx) {
+ return rv; /* TX removed it interim, mustn't happen */
+ }
+ return mqp_setup_proc_ctx_data_recv(cl_ctx, mqp, 1, app);
+}
+
+static int32_t cl_recv(struct mqtt_packet *mqp, uint32_t wait_secs, void **app)
+{
+
+ int32_t rv = MQP_ERR_NETWORK;
+ int32_t n_hnds = 0, idx = 0;
+
+ rv = group_ctxs_rx_prep(wait_secs, app);
+ if(rv > 0)
+ n_hnds = rv;
+
+ for(idx = 0; (idx < n_hnds) && (rv > 0); idx++) {
+ int32_t net = recv_hvec[idx];
+ if(loopb_net == net)
+ rv = proc_loopback_recv(net); /* UDP Packet */
+ else {
+ rv = proc_net_data_recv(net, mqp, app);
+ if(false == grp_has_cbfn)
+ break; /* 'CTX': inform application */
+ }
+ }
+
+ return rv;
+}
+
+static int32_t grp_net_setup_create()
+{
+
+ if(0 == loopb_portid) {
+ return MQP_ERR_NOT_DEF;
+ }
+ if(NULL == net_ops) {
+ return MQP_ERR_NET_OPS;
+ }
+ if(-1 == loopb_net) {
+ loopb_net = net_ops->open(DEV_NETCONN_OPT_UDP, NULL, loopb_portid, NULL);
+
+ if(-1 == loopb_net) {
+ return MQP_ERR_LIBQUIT;
+ }
+ }
+
+ return 1;
+}
+
+int32_t mqtt_client_await_msg(struct mqtt_packet *mqp, uint32_t wait_secs, void **app)
+{
+
+ int32_t rv = MQP_ERR_NOTCONN;
+ *app = NULL;
+
+ if(NULL == mqp)
+ return MQP_ERR_FNPARAM; /* Didn't get a valid MQP */
+
+ if(true == grp_has_cbfn)
+ return MQP_ERR_BADCALL; /* Err: LIB has CB config */
+
+ rv = grp_net_setup_create();
+ if(rv <= 0)
+ return rv;
+
+ do {
+ rv = cl_recv(mqp, wait_secs, app);
+
+ } while((rv > 0) && MQP_RX_DO_NOT_RPT_COR(mqp));
+
+ return rv;
+}
+
+int32_t mqtt_client_run(uint32_t wait_secs)
+{
+
+ void *app = NULL;
+ int32_t rv = -1;
+
+ if(false == grp_has_cbfn) {
+ return MQP_ERR_BADCALL; /* Err: LIB has no CB config */
+ }
+ rv = grp_net_setup_create();
+ if(rv <= 0) {
+ return rv;
+ }
+ do {
+ rv = cl_recv(NULL, wait_secs, &app);
+
+ } while((rv > 0) ||
+ /* 'ctx' related errors are handled by the callbacks */
+ ((rv != MQP_ERR_LIBQUIT) && (rv != MQP_ERR_TIMEOUT)));
+
+ return rv;
+}
+
+/*------------------------------------------------------------------------------
+ * Buffer Pool and management, other registrations and initialization.
+ *------------------------------------------------------------------------------
+ */
+static struct mqtt_packet *free_list = NULL;
+
+static struct mqtt_packet *mqp_alloc_atomic(void) {
+
+ struct mqtt_packet *mqp = NULL;
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 40 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+ mqp = free_list;
+ if(mqp) {
+ free_list = mqp->next;
+ }
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ xSemaphoreGive( xSemaphore );
+
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+// MUTEX_UNLOCK();
+ return mqp;
+}
+
+struct mqtt_packet *mqp_client_alloc(uint8_t msg_type, uint8_t offset) {
+
+ struct mqtt_packet *mqp = mqp_alloc_atomic();
+ if(NULL == mqp) {
+ USR_INFO("MQP alloc failed - msg type 0x%02x\n\r", msg_type);
+ return NULL;
+ }
+
+ mqp_init(mqp, offset);
+ mqp->msg_type = msg_type;
+
+ return mqp;
+}
+
+/* Do not use this routine with-in this file. */
+static void free_mqp(struct mqtt_packet *mqp)
+{
+ /* Must be used in a locked state */
+ mqp->next = free_list;
+ free_list = mqp;
+}
+
+int32_t mqtt_client_buffers_register(uint32_t num_mqp, struct mqtt_packet *mqp_vec,
+ uint32_t buf_len, uint8_t *buf_vec)
+{
+ uint32_t i, j;
+
+ if((0 == num_mqp) || (0 == buf_len) || free_list)
+ return -1;
+
+ for(i = 0, j = 0; i < num_mqp; i++, j += buf_len) {
+ struct mqtt_packet *mqp = mqp_vec + i;
+
+ mqp->buffer = buf_vec + j;
+ mqp->maxlen = buf_len;
+
+ mqp->free = free_mqp;
+ mqp->next = free_list;
+ free_list = mqp;
+ }
+
+ return 0;
+}
+
+int32_t mqtt_client_ctx_will_register(void *ctx,
+ const struct utf8_string *will_top,
+ const struct utf8_string *will_msg,
+ enum mqtt_qos will_qos, bool retain)
+{
+ uint8_t B = 0;
+
+ if((NULL == ctx) || ((NULL == will_top) && (NULL != will_msg)))
+ return -1; /* Bad Combo */
+
+ if(NULL != will_top) {
+ RET_IF_INVALID_UTF8(will_top);
+
+ B = QOS_VALUE(will_qos) << 3;
+ if(retain)
+ B |= WILL_RETAIN_VAL;
+
+ if(NULL != will_msg)
+ RET_IF_INVALID_UTF8(will_msg);
+ }
+
+ CLIENT(ctx)->conn_pl_utf8s[1] = will_top;
+ CLIENT(ctx)->conn_pl_utf8s[2] = will_msg;
+
+ CLIENT(ctx)->will_opts = B;
+
+ return 0;
+}
+
+int32_t mqtt_client_ctx_info_register(void *ctx,
+ const struct utf8_string *client_id,
+ const struct utf8_string *user_name,
+ const struct utf8_string *pass_word)
+{
+ const struct utf8_string *users_pwd = NULL;
+
+ if(NULL == ctx)
+ return -1;
+
+ if(NULL != client_id)
+ RET_IF_INVALID_UTF8(client_id);
+
+ if(NULL != user_name) {
+ RET_IF_INVALID_UTF8(user_name);
+
+ if(NULL != pass_word)
+ RET_IF_INVALID_UTF8(pass_word);
+
+ users_pwd = pass_word;
+ }
+
+ CLIENT(ctx)->conn_pl_utf8s[0] = client_id;
+ CLIENT(ctx)->conn_pl_utf8s[3] = user_name;
+ CLIENT(ctx)->conn_pl_utf8s[4] = users_pwd;
+
+ return 0;
+}
+
+int32_t mqtt_client_net_svc_register(const struct device_net_services *net)
+{
+ if(net && net->open && net->send && net->recv &&
+ net->send_dest && net->recv_from && net->close
+ && net->io_mon && net->time) {
+ net_ops = net;
+ return 0;
+ }
+
+ return -1;
+}
+
+static void cl_ctx_setup(struct client_ctx *cl_ctx, /* WR Object */
+ const struct mqtt_client_ctx_cfg *ctx_cfg,
+ const struct mqtt_client_ctx_cbs *ctx_cbs,
+ void *app)
+{
+
+ struct client_desc *client = CLIENT(cl_ctx);
+
+ cl_ctx->flags = ctx_cfg->config_opts;
+
+ client->nwconn_opts = ctx_cfg->nwconn_opts;
+ client->server_addr = ctx_cfg->server_addr;
+ client->port_number = ctx_cfg->port_number;
+
+ client->app = app;
+
+
+ if(NULL != ctx_cfg->nw_security)
+ buf_wr_nbytes((uint8_t*)&client->nw_security,
+ (uint8_t*)ctx_cfg->nw_security,
+ sizeof(struct secure_conn));
+
+ if(NULL != ctx_cbs) {
+ // set callback flag
+ struct mqtt_client_ctx_cbs *cbs_ctx = CTX_CBS_PTR(client);
+ cbs_ctx->publish_rx = ctx_cbs->publish_rx;
+ cbs_ctx->ack_notify = ctx_cbs->ack_notify;
+ cbs_ctx->disconn_cb = ctx_cbs->disconn_cb;
+ }
+
+ return;
+}
+
+int32_t mqtt_client_ctx_create(const struct mqtt_client_ctx_cfg *ctx_cfg,
+ const struct mqtt_client_ctx_cbs *ctx_cbs,
+ void *app, void **ctx)
+{
+
+ struct client_ctx *cl_ctx = NULL;
+
+ if((NULL == ctx_cfg) ||
+ (NULL == ctx_cfg->server_addr) ||
+ (0 == ctx_cfg->port_number)) {
+ return -1;
+ }
+ if(ctx_cfg->config_opts & MQTT_CFG_MK_GROUP_CTX) {
+ if(grp_has_cbfn ^ (!!ctx_cbs)) {
+ return -1;
+ }
+ }
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 40 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+ if(free_ctxs) {
+ cl_ctx = free_ctxs;
+ free_ctxs = cl_ctx->next;
+ cl_ctx->next = NULL;
+ }
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ xSemaphoreGive( xSemaphore );
+
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+// MUTEX_UNLOCK();
+
+ if(cl_ctx) {
+ cl_ctx_setup(cl_ctx, ctx_cfg, ctx_cbs, app);
+ *ctx = (void*) cl_ctx;
+ return 0;
+ }
+
+ return -1;
+}
+
+int32_t mqtt_client_ctx_delete(void *ctx)
+{
+
+ struct client_ctx *cl_ctx = (struct client_ctx*) ctx;
+ int32_t rv = -1; /* Not sure about deletion as yet */
+
+// MUTEX_LOCKIN();
+ if( xSemaphore != NULL ) {
+ // See if we can obtain the semaphore. If the semaphore is not available
+ // wait 10 ticks to see if it becomes free.
+ if( xSemaphoreTake( xSemaphore, ( TickType_t ) 40 ) == pdTRUE ) {
+ // We were able to obtain the semaphore and can now access the
+ // shared resource.
+ if((NULL == cl_ctx) ||
+ (-1 != cl_ctx->net) ||
+ (awaits_pkts(cl_ctx))) {
+ goto mqtt_client_ctx_delete_exit1;
+ }
+ rv = 0; /* OK to delete ctx */
+ client_reset(CLIENT(cl_ctx));
+ cl_ctx_freeup(cl_ctx);
+
+ // We have finished accessing the shared resource. Release the
+ // semaphore.
+ //xSemaphoreGive( xSemaphore );
+ } else {
+ // We could not obtain the semaphore and can therefore not access
+ // the shared resource safely.
+ Uart_Write((uint8_t*)"Unable to obtain the semaphore\r\n");
+ }
+ }
+
+mqtt_client_ctx_delete_exit1:
+// MUTEX_UNLOCK();
+ xSemaphoreGive(xSemaphore);
+
+ return rv;
+}
+
+int32_t mqtt_client_lib_init(const struct mqtt_client_lib_cfg *lib_cfg)
+{
+ if((NULL == lib_cfg) || (NULL == lib_cfg->debug_printf))
+ return -1;
+
+ debug_printf = lib_cfg->debug_printf; /* Facilitate debug */
+
+ if(INIT_DONE_STATE == cl_lib_state) {
+ Uart_Write((uint8_t*)"C: Error trying to re-initialize \n\r");
+ USR_INFO("C: Error trying to re-initialize \n\r");
+ return -1;
+ }
+
+ USR_INFO("Version: Client LIB %s, Common LIB %s.\n\r",
+ MQTT_CLIENT_VERSTR, MQTT_COMMON_VERSTR);
+
+ client_desc_init();
+
+ cl_lib_state = INIT_DONE_STATE;
+
+ loopb_portid = lib_cfg->loopback_port;
+ grp_has_cbfn = lib_cfg->grp_uses_cbfn;
+
+ mutex = lib_cfg->mutex;
+ mutex_lockin = lib_cfg->mutex_lockin;
+ mutex_unlock = lib_cfg->mutex_unlock;
+
+ aux_dbg_enbl = lib_cfg->aux_debug_en;
+
+ return 0;
+}
+
+int32_t mqtt_client_lib_exit()
+{
+ struct client_ctx *cl_ctx = free_ctxs;
+ int32_t count = 0;
+
+ while(cl_ctx) {
+ cl_ctx = cl_ctx->next;
+ count++;
+ }
+
+ if(MAX_NWCONN == count) {
+ cl_lib_state = WAIT_INIT_STATE;
+ free_ctxs = NULL;
+ return 0;
+ }
+
+ return -1;
+}
+
+}//namespace mbed_mqtt