David Fletcher
Part of TI's mqtt
Dependents: mqtt_V1 cc3100_Test_mqtt_CM3
Diff: client_mgmt.cpp
- Revision:
- 0:547251f42a60
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/client_mgmt.cpp Sat Jun 06 13:29:08 2015 +0000
@@ -0,0 +1,790 @@
+/******************************************************************************
+*
+* 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.
+*
+******************************************************************************/
+
+#include "server_util.h"
+#include "client_mgmt.h"
+#include "server_pkts.h"
+
+namespace mbed_mqtt {
+
+#ifndef CFG_SR_MAX_CL_ID_SIZE
+#define MAX_CLIENT_ID_LEN 64
+#else
+#define MAX_CLIENT_ID_LEN CFG_SR_MAX_CL_ID_SIZE
+#endif
+
+#ifndef CFG_SR_MAX_NUM_CLIENT
+#define MAX_CLIENTS 16 /* Must not exceed 32 */
+#else
+#define MAX_CLIENTS CFG_SR_MAX_NUM_CLIENT
+
+#if (CFG_SR_MAX_NUM_CLIENT > 32)
+#error "CFG_SR_MAX_NUM_CLIENT must not exceed 32"
+#endif
+#endif
+
+static struct client_usr {
+
+ void *ctx; /* Client net */
+ void *app; /* Client app */
+
+ void *will; /* Will topic */
+ uint32_t index;
+ uint32_t n_sub;
+
+#define MQ_CONNECT_FLAG 0x00000001
+#define CLEAN_SESS_FLAG 0x00000002
+#define ASSIGNMENT_FLAG 0x00000004 /* No state or connected */
+
+ uint32_t flags;
+
+ char client_id[MAX_CLIENT_ID_LEN];
+
+ struct pub_qos2_cq qos2_rx_cq;
+ struct pub_qos2_cq qos2_tx_cq;
+
+} users[MAX_CLIENTS];
+
+// TBD consistency between inline and macro functions
+
+static inline bool is_connected(struct client_usr *cl_usr)
+{
+ return (cl_usr->flags & MQ_CONNECT_FLAG)? true : false;
+}
+
+static inline bool has_clean_session(struct client_usr *cl_usr)
+{
+ return (cl_usr->flags & CLEAN_SESS_FLAG)? true : false;
+}
+
+static inline bool is_assigned(struct client_usr *cl_usr)
+{
+ return (cl_usr->flags & ASSIGNMENT_FLAG)? true : false;
+}
+
+static inline void set_connect_state(struct client_usr *cl_usr,
+ bool connected)
+{
+ if(connected)
+ cl_usr->flags |= MQ_CONNECT_FLAG;
+ else
+ cl_usr->flags &= ~MQ_CONNECT_FLAG;
+}
+
+static inline void set_clean_session(struct client_usr *cl_usr,
+ bool clean_session)
+{
+ if(clean_session)
+ cl_usr->flags |= CLEAN_SESS_FLAG;
+ else
+ cl_usr->flags &= ~CLEAN_SESS_FLAG;
+}
+
+static inline void set_assigned(struct client_usr *cl_usr,
+ bool assignment)
+{
+ if(assignment)
+ cl_usr->flags |= ASSIGNMENT_FLAG;
+ else
+ cl_usr->flags &= ~ASSIGNMENT_FLAG;
+}
+
+static void cl_usr_reset(struct client_usr *cl_usr)
+{
+ cl_usr->ctx = NULL;
+ cl_usr->app = NULL;
+ cl_usr->will = NULL;
+
+ cl_usr->n_sub = 0;
+ cl_usr->flags = 0;
+
+ cl_usr->client_id[0] = '\0';
+
+ qos2_pub_cq_reset(&cl_usr->qos2_rx_cq);
+ qos2_pub_cq_reset(&cl_usr->qos2_tx_cq);
+}
+
+static void cl_usr_init(void)
+{
+ int32_t idx = 0;
+ for(idx = 0; idx < MAX_CLIENTS; idx++) {
+ struct client_usr *cl_usr = users + idx;
+ cl_usr->index = idx;
+
+ cl_usr_reset(cl_usr);
+ }
+}
+
+static void cl_usr_free(struct client_usr *cl_usr)
+{
+ cl_usr_reset(cl_usr);
+ cl_usr->flags &= ~(MQ_CONNECT_FLAG |
+ CLEAN_SESS_FLAG |
+ ASSIGNMENT_FLAG);
+}
+
+void cl_sub_count_add(void *usr_cl)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ if(is_connected(cl_usr)) {
+ cl_usr->n_sub++;
+ USR_INFO("%s has added a new sub, now total is %u\n\r",
+ cl_usr->client_id, cl_usr->n_sub);
+
+ }
+
+ return;
+}
+
+void cl_sub_count_del(void *usr_cl)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ if(is_connected(cl_usr)) {
+ cl_usr->n_sub--;
+ USR_INFO("%s has deleted a sub, now total is %u\n\r",
+ cl_usr->client_id, cl_usr->n_sub);
+ }
+
+ return;
+}
+
+/*----------------------------------------------------------------------------
+ * QoS2 PUB RX Message handling mechanism and associated house-keeping
+ *--------------------------------------------------------------------------*/
+static inline bool qos2_pub_rx_logup(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ return qos2_pub_cq_logup(&cl_usr->qos2_rx_cq, msg_id);
+}
+
+static inline bool ack2_msg_id_logup(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ return qos2_pub_cq_logup(&cl_usr->qos2_tx_cq, msg_id);
+}
+
+static inline bool qos2_pub_rx_unlog(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ return qos2_pub_cq_unlog(&cl_usr->qos2_rx_cq, msg_id);
+}
+
+static inline bool ack2_msg_id_unlog(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ return qos2_pub_cq_unlog(&cl_usr->qos2_tx_cq, msg_id);
+}
+
+static inline bool qos2_pub_rx_is_done(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ return qos2_pub_cq_check(&cl_usr->qos2_rx_cq, msg_id);
+}
+
+bool cl_mgmt_qos2_pub_rx_update(void *usr_cl, uint16_t msg_id)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ return cl_usr && (qos2_pub_rx_is_done(cl_usr, msg_id) ||
+ qos2_pub_rx_logup(cl_usr, msg_id));
+}
+
+static void ack2_msg_id_dispatch(struct client_usr *cl_usr)
+{
+ struct pub_qos2_cq *tx_cq = &cl_usr->qos2_tx_cq;
+ uint8_t rd_idx = tx_cq->rd_idx;
+ uint8_t n_free = tx_cq->n_free;
+ uint8_t i = 0;
+
+ for(i = rd_idx; i < (MAX_PUBREL_INFLT - n_free); i++) {
+ if(mqtt_vh_msg_send(cl_usr->ctx, MQTT_PUBREL, MQTT_QOS1,
+ true, tx_cq->id_vec[i]) <= 0)
+ break;
+ }
+
+ return;
+}
+
+static void ack2_msg_id_purge(struct client_usr *cl_usr)
+{
+ qos2_pub_cq_reset(&cl_usr->qos2_tx_cq);
+ qos2_pub_cq_reset(&cl_usr->qos2_rx_cq);
+}
+
+/*
+
+*/
+static struct mqtt_ack_wlist wl_qos_ack1 = {NULL, NULL};
+static struct mqtt_ack_wlist *qos_ack1_wl = &wl_qos_ack1;
+
+/*
+
+*/
+static struct mqtt_ack_wlist wl_mqp_sess = {NULL, NULL};
+static struct mqtt_ack_wlist *sess_mqp_wl = &wl_mqp_sess;
+
+#define MQP_CL_MAP_GET(mqp) (mqp->private_)
+#define MQP_CL_MAP_SET(mqp, cl_map) (mqp->private_ |= cl_map)
+#define MQP_CL_MAP_CLR(mqp, cl_map) (mqp->private_ &= ~cl_map)
+
+#define CL_BMAP(cl_usr) (1 << cl_usr->index)
+
+static inline
+int32_t _pub_dispatch(struct client_usr *cl_usr, struct mqtt_packet *mqp,
+ bool dup)
+{
+ /* Error, if any, is handled in 'cl_net_close()' ....*/
+ return mqtt_server_pub_dispatch(cl_usr->ctx, mqp, dup);
+}
+
+static void ack1_wl_mqp_dispatch(struct client_usr *cl_usr)
+{
+ struct mqtt_packet *mqp = NULL;
+
+ for(mqp = qos_ack1_wl->head; NULL != mqp; mqp = mqp->next)
+ if(MQP_CL_MAP_GET(mqp) & CL_BMAP(cl_usr))
+ _pub_dispatch(cl_usr, mqp, true);
+}
+
+#if 0
+static struct client_usr *find_cl_usr(uint32_t index)
+{
+ struct client_usr *cl_usr = users + 0;
+
+ int32_t i = 0;
+ for(i = 0; i < MAX_CLIENTS; i++, cl_usr++) {
+ if(index == cl_usr->index)
+ break;
+ }
+
+ return (MAX_CLIENTS == i)? NULL : cl_usr;
+}
+#endif
+
+#define IS_CL_USR_FREE(cl_usr) (( 0 == cl_usr->flags) && \
+ ('\0' == cl_usr->client_id[0]))
+
+#define IS_CL_INACTIVE(cl_usr) (( 0 == cl_usr->flags) && \
+ ('\0' != cl_usr->client_id[0]))
+
+#define NEED_TO_WAIT_LIST_PUBLISH(qos, cl_usr) \
+ ((((MQTT_QOS1 == qos) && has_clean_session(cl_usr)) || \
+ ((MQTT_QOS0 == qos)))? \
+ false : true)
+
+static inline uint32_t _cl_bmap_get(void *usr_cl)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ return IS_CL_USR_FREE(cl_usr)? 0 : CL_BMAP(cl_usr);
+}
+
+uint32_t cl_bmap_get(void *usr_cl)
+{
+ return usr_cl? _cl_bmap_get(usr_cl) : 0;
+}
+
+void *cl_app_hndl_get(void *usr_cl)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ return (cl_usr && is_connected((client_usr*) usr_cl))? cl_usr->app : NULL;
+}
+
+void *cl_will_hndl_get(void *usr_cl)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ return (cl_usr && is_connected((client_usr*)usr_cl))? cl_usr->will : NULL;
+}
+
+static void pub_dispatch(uint32_t cl_map, struct mqtt_packet *mqp)
+{
+ //uint32_t n_bits = sizeof(uint32_t) << 3; /* Multiply by 8 */
+ uint32_t sp_map = 0; /* client Map for sessions present */
+ enum mqtt_qos qos = ENUM_QOS(mqp->fh_byte1);/* QOS */
+ const uint32_t cl_ref = cl_map; /* Keep ref to original */
+
+ int32_t i = 0;
+ for(i = 0; i < MAX_CLIENTS; i++) {
+ if(cl_map & (1 << i)) {
+ struct client_usr *cl_usr = users + i; //find_cl_usr(i);
+ if(is_connected(cl_usr))
+ if((_pub_dispatch(cl_usr, mqp, false) > 0) &&
+ NEED_TO_WAIT_LIST_PUBLISH(qos, cl_usr))
+ continue;/* Processing done; next CL */
+
+ /* CL: unconnected / PUB Err / QOS1 PKT (clean sess) */
+ cl_map &= ~(1 << i);
+
+ if(IS_CL_INACTIVE(cl_usr))
+ sp_map |= (1 << i); /* CL: Maintain session */
+ }
+ }
+
+ if(sp_map) {
+ struct mqtt_packet *cpy = mqp_server_copy(mqp); /* Make copy */
+ if(cpy) {
+ MQP_CL_MAP_SET(cpy, sp_map);
+ mqp_ack_wlist_append(sess_mqp_wl, cpy);
+ } else
+ sp_map = 0;
+ }
+
+ if(cl_map) { /* Wait List Publish */
+ MQP_CL_MAP_SET(mqp, cl_map);
+ mqp_ack_wlist_append(qos_ack1_wl, mqp);
+ } else
+ mqp_free(mqp); /* PUB MQP now has no more use; must be freed */
+
+ USR_INFO("PUBLISH: CL Map(0x%08x): For Ack 0x%08x, Inactive 0x%08x\n\r",
+ cl_ref, cl_map, sp_map);
+
+ return;
+}
+
+void cl_pub_dispatch(uint32_t cl_map, struct mqtt_packet *mqp)
+{
+ pub_dispatch(cl_map, mqp);
+ return;
+}
+
+/* Move to mqtt_common.h and remove it from mqtt_client.h */
+static inline int32_t len_err_free_mqp(struct mqtt_packet *mqp)
+{
+ mqp_free(mqp);
+ return MQP_ERR_PKT_LEN;
+}
+
+/* Move this to a common file */
+int32_t cl_pub_msg_send(void *usr_cl,
+ 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 == topic) ||
+ ((data_len > 0) && (NULL == data_buf)) ||
+ (NULL == usr_cl))
+ return MQP_ERR_FNPARAM;
+
+ mqp = mqp_server_alloc(MQTT_PUBLISH, 2 + topic->length + 2 + data_len);
+ if(NULL == mqp)
+ return MQP_ERR_PKT_AVL;
+
+ if((0 > mqp_pub_append_topic(mqp, topic, qos? mqp_new_id_server(): 0)) ||
+ (data_len && (0 > mqp_pub_append_data(mqp, data_buf, data_len))))
+ return len_err_free_mqp(mqp);
+
+ mqp_prep_fh(mqp, MAKE_FH_FLAGS(false, qos, retain));
+
+ pub_dispatch(_cl_bmap_get(usr_cl), mqp);
+
+ return MQP_CONTENT_LEN(mqp);
+}
+
+static void wl_remove(struct mqtt_ack_wlist *list,
+ struct mqtt_packet *prev,
+ struct mqtt_packet *elem)
+{
+ if(prev)
+ prev->next = elem->next;
+ else
+ list->head = elem->next;
+
+ if(NULL == list->head)
+ list->tail = NULL;
+
+ if(list->tail == elem)
+ list->tail = prev;
+
+ return;
+}
+
+static void sess_wl_mqp_dispatch(struct client_usr *cl_usr)
+{
+ struct mqtt_packet *mqp = NULL, *prev = NULL, *next = NULL;
+ uint32_t cl_map = CL_BMAP(cl_usr);
+
+ for(mqp = sess_mqp_wl->head; NULL != mqp; prev = mqp, mqp = next) {
+ struct mqtt_packet *cpy = NULL;
+
+ if(0 == (MQP_CL_MAP_GET(mqp) & cl_map))
+ continue; /* MQP & CL: no association */
+
+ /* MQP has associated client(s) - process it */
+
+ /* Dissociate this client from MQP */
+ MQP_CL_MAP_CLR(mqp, cl_map);
+ next = mqp->next; /* House keeping */
+
+ if(0 == MQP_CL_MAP_GET(mqp)) {
+ /* MQP w/ no clients, remove from WL */
+ wl_remove(sess_mqp_wl, prev, mqp);
+
+ /* Ensure correct arrangement for WL */
+ cpy = mqp; mqp = prev;
+ } else {
+ /* MQP is associated w/ other clients */
+ cpy = mqp_server_copy(mqp);
+ if(NULL == cpy)
+ continue;
+ }
+
+ /* Got packet from session, dispatch it to CL */
+ pub_dispatch(cl_map, cpy);
+ }
+
+ return;
+}
+
+static
+struct mqtt_packet *wl_mqp_unmap_find(struct mqtt_ack_wlist *wl,
+ struct client_usr *cl_usr, uint16_t msg_id,
+ struct mqtt_packet **prev)
+{
+ struct mqtt_packet *mqp = NULL;
+
+ *prev = NULL;
+ for(mqp = wl->head; NULL != mqp; *prev = mqp, mqp = mqp->next) {
+ if(msg_id == mqp->msg_id) {
+ /* Found a MQP whose msg_id matches with input */
+ MQP_CL_MAP_CLR(mqp, CL_BMAP(cl_usr));
+ return mqp;
+ }
+ }
+
+ return NULL;
+}
+
+static bool wl_rmfree_try(struct mqtt_ack_wlist *wl, struct mqtt_packet *prev,
+ struct mqtt_packet *mqp)
+{
+ if(0 == MQP_CL_MAP_GET(mqp)) {
+ wl_remove(wl, prev, mqp);
+ mqp_free(mqp);
+
+ return true;
+ }
+
+ return false;
+}
+
+static bool ack1_unmap_rmfree_try(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ struct mqtt_packet *prev = NULL, *mqp = NULL;
+
+ if(false == has_clean_session(cl_usr)) {
+ mqp = wl_mqp_unmap_find(qos_ack1_wl, cl_usr, msg_id, &prev);
+ if(mqp)
+ wl_rmfree_try(qos_ack1_wl, prev, mqp);
+ }
+
+ return true;
+}
+
+static
+void wl_purge(struct mqtt_ack_wlist *wl, struct client_usr *cl_usr)
+{
+ struct mqtt_packet *mqp = NULL, *prev = NULL, *next = NULL;
+ uint32_t bmap = CL_BMAP(cl_usr);
+
+ for(mqp = wl->head; NULL != mqp; prev = mqp, mqp = next) {
+ next = mqp->next;
+
+ /* Ideally, check should be done to see if cl_usr and mqp are
+ associated. If yes, then the bit should be cleared. At the
+ moment, blind clearing of the bit has been implemented and
+ it has no side effects.
+ */
+ MQP_CL_MAP_CLR(mqp, bmap);
+
+ /* MQP with no clients has no more use, so try deleting MQP */
+ if(wl_rmfree_try(wl, prev, mqp))
+ mqp = prev; /* MQP deleted, prep for next iteration */
+ }
+
+ return;
+}
+
+static void ack1_wl_mqp_purge(struct client_usr *cl_usr)
+{
+ wl_purge(qos_ack1_wl, cl_usr);
+}
+
+static void sess_wl_mqp_purge(struct client_usr *cl_usr)
+{
+ wl_purge(sess_mqp_wl, cl_usr);
+}
+
+static void session_resume(struct client_usr *cl_usr)
+{
+ ack1_wl_mqp_dispatch(cl_usr);
+ ack2_msg_id_dispatch(cl_usr);
+ sess_wl_mqp_dispatch(cl_usr);
+}
+
+static void session_delete(struct client_usr *cl_usr)
+{
+ ack1_wl_mqp_purge(cl_usr);
+ ack2_msg_id_purge(cl_usr);
+ sess_wl_mqp_purge(cl_usr);
+}
+
+static bool has_session_data(struct client_usr *cl_usr)
+{
+ uint32_t map = CL_BMAP(cl_usr);
+ struct mqtt_packet *elem;
+
+ if(cl_usr->n_sub)
+ return true;
+
+ elem = qos_ack1_wl->head;
+ while(elem) {
+ if(MQP_CL_MAP_GET(elem) & map)
+ return true;
+
+ elem = elem->next;
+ }
+
+ elem = sess_mqp_wl->head;
+ while(elem) {
+ if(MQP_CL_MAP_GET(elem) & map)
+ return true;
+
+ elem = elem->next;
+ }
+
+ return false;
+}
+
+bool cl_can_session_delete(void *usr_cl)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ return cl_usr? (has_clean_session(cl_usr) ||
+ !has_session_data(cl_usr)) : false;
+}
+
+void cl_on_net_close(void *usr_cl)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ if(is_assigned(cl_usr)) {
+ if(false == has_session_data(cl_usr))
+ cl_usr_free(cl_usr);
+ } else if(has_clean_session(cl_usr)) {
+ session_delete(cl_usr);
+ cl_usr_free(cl_usr);
+ } else {
+ set_connect_state(cl_usr, false);
+ cl_usr->ctx = NULL;
+ cl_usr->app = NULL;
+ }
+
+ return;
+}
+
+static bool proc_pub_rel_rx(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ mqtt_vh_msg_send(cl_usr->ctx, MQTT_PUBCOMP,
+ MQTT_QOS0, true, msg_id);
+
+ if(qos2_pub_rx_is_done(cl_usr, msg_id))
+ qos2_pub_rx_unlog(cl_usr, msg_id);
+
+ return true;
+}
+
+static bool proc_pub_rec_rx(struct client_usr *cl_usr, uint16_t msg_id)
+{
+ struct mqtt_packet *prev = NULL, *mqp = NULL;
+
+ mqp = wl_mqp_unmap_find(qos_ack1_wl, cl_usr, msg_id, &prev);
+ if(mqp && ack2_msg_id_logup(cl_usr, msg_id)) {
+
+ wl_rmfree_try(qos_ack1_wl, prev, mqp);
+
+ mqtt_vh_msg_send(cl_usr->ctx, MQTT_PUBREL,
+ MQTT_QOS1, true, msg_id);
+
+ return true;
+ }
+
+ return false;
+}
+
+bool cl_notify_ack(void *usr_cl, uint8_t msg_type, uint16_t msg_id)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+ bool rv = false;
+
+ if(NULL == cl_usr)
+ return rv;
+
+ switch(msg_type) {
+
+ case MQTT_PUBACK:
+ rv = ack1_unmap_rmfree_try(cl_usr, msg_id);
+ break;
+
+ case MQTT_PUBREC:
+ rv = proc_pub_rec_rx(cl_usr, msg_id);
+ break;
+
+ case MQTT_PUBREL:
+ rv = proc_pub_rel_rx(cl_usr, msg_id);
+ break;
+
+ case MQTT_PUBCOMP:
+ rv = ack2_msg_id_unlog(cl_usr, msg_id);
+ break;
+
+ default:
+ break;
+ }
+
+ return rv;
+}
+
+static void assign_cl_index_as_id(struct client_usr *cl_usr)
+{
+ /* TBD: need a better implementation */
+ char *client_id = cl_usr->client_id;
+
+ client_id[0] = 'S';
+ client_id[1] = 'e';
+ client_id[2] = 'l';
+ client_id[3] = 'f';
+ client_id[4] = '-';
+ client_id[5] = ((cl_usr->index & 0xf0) >> 4) + 0x30;
+ client_id[6] = ((cl_usr->index & 0x0f)) + 0x30;
+ client_id[7] = '\0';
+
+ /* Make sure that above array size is with in MAX_CLIENT_ID_LEN */
+
+ return;
+}
+
+static struct client_usr *assign_cl_usr(char *client_id, uint8_t *vh_buf)
+{
+ struct client_usr *cl_usr, *fr_usr = NULL;
+ int32_t idx = 0;
+ for(idx = 0; idx < MAX_CLIENTS; idx++) {
+ cl_usr = users + idx;
+ if((NULL == fr_usr) && IS_CL_USR_FREE(cl_usr))
+ fr_usr = cl_usr; /* Note 1st free cl_usr */
+
+ if((NULL == client_id) && (NULL != fr_usr)) {
+ /* Free cl_usr is present to create CL-ID */
+ break;
+
+ } else if((NULL != client_id) &&
+ (0 == strncmp(cl_usr->client_id, client_id,
+ MAX_CLIENT_ID_LEN))) {
+ /* Found a client obj with exact ID match */
+ if(is_connected(cl_usr)) {
+ /* Error: CL-ID is already active */
+ vh_buf[1] = CONNACK_RC_CLI_REJECT;
+ cl_usr = NULL;
+ }
+ break;
+ }
+ }
+
+ if(idx == MAX_CLIENTS) { /* Did not find a match */
+ cl_usr = fr_usr;
+ if(NULL == cl_usr)
+ /* Server Resource unavailable */
+ vh_buf[1] = CONNACK_RC_SVR_UNAVBL;
+ }
+
+ if(NULL != cl_usr) {
+ if(NULL == client_id)
+ assign_cl_index_as_id(cl_usr); /* Get ID */
+ else if(IS_CL_USR_FREE(cl_usr))
+ strncpy(cl_usr->client_id, client_id,
+ MAX_CLIENT_ID_LEN);
+
+ set_assigned(cl_usr, true);
+ }
+
+ USR_INFO("Assignment of cl_usr %s (detail: index %d and name %s) \n\r",
+ cl_usr? "Succeeded" : "Failed", cl_usr? (int32_t)cl_usr->index : -1,
+ cl_usr? cl_usr->client_id : "");
+
+ return cl_usr;
+}
+
+uint16_t cl_connect_rx(void *ctx_cl, bool clean_session, char *client_id,
+ void *app_cl, void *will, void **usr_cl)
+{
+ uint8_t vh_buf[] = {0x00, CONNACK_RC_REQ_ACCEPT};
+ struct client_usr *cl_usr;
+
+ if(client_id && ('\0' == client_id[0]))
+ /* Shouldn't happen: CLIENT ID with a NUL only string */
+ return CONNACK_RC_CLI_REJECT;
+
+ cl_usr = assign_cl_usr(client_id, vh_buf);
+ if(NULL == cl_usr)
+ return vh_buf[1]; /* Use vh_buf from assign_cl_usr() */
+
+ if((false == clean_session) && has_session_data(cl_usr))
+ vh_buf[0] = 0x01; /* Set Session Present Flag */
+
+#if 0
+ if(0x00 == vh_buf[1])
+ *usr_cl = (void*)cl_usr;
+#endif
+ *usr_cl = (void*)cl_usr;
+
+ cl_usr->ctx = ctx_cl;
+ cl_usr->app = app_cl;
+ cl_usr->will = will;
+
+ return ((vh_buf[0] << 8) | vh_buf[1]);
+}
+
+void cl_on_connack_send(void *usr_cl, bool clean_session)
+{
+ struct client_usr *cl_usr = (struct client_usr*) usr_cl;
+
+ if(false == is_assigned(cl_usr)) {
+ USR_INFO("Fatal: CONNACK for unassigned cl_usr Id %u, abort\n\r",
+ cl_usr->index);
+ return;
+ }
+
+ set_assigned(cl_usr, false);
+ set_clean_session(cl_usr, clean_session);
+ set_connect_state(cl_usr, true);
+
+ if(clean_session) {
+ session_delete(cl_usr);
+ } else {
+ session_resume(cl_usr);
+ }
+
+ DBG_INFO("Server is now connected to client %s\n\r", cl_usr->client_id);
+
+ return;
+}
+
+int32_t cl_mgmt_init(void)
+{
+ cl_usr_init();
+
+ return 0;
+}
+
+}//namespace mbed_mqtt