version 1.1.1.1.2.11, 2012/04/27 16:17:11
|
version 1.2, 2012/06/20 15:02:24
|
Line 191 mqtt_msgRealloc(mqtt_msg_t * __restrict msg, u_short l
|
Line 191 mqtt_msgRealloc(mqtt_msg_t * __restrict msg, u_short l
|
} |
} |
|
|
/* |
/* |
|
* mqtt_msgDup() - Duplicate message buffer |
|
* |
|
* @msg = Message |
|
* return: NULL error or !=NULL duplicated message, after use must call mqtt_msgFree() with all!=0 |
|
*/ |
|
inline mqtt_msg_t * |
|
mqtt_msgDup(mqtt_msg_t * __restrict msg) |
|
{ |
|
mqtt_msg_t *m = NULL; |
|
|
|
m = malloc(sizeof(mqtt_msg_t)); |
|
if (!m) { |
|
LOGERR; |
|
return NULL; |
|
} else |
|
memset(m, 0, sizeof(mqtt_msg_t)); |
|
|
|
if (msg->msg_len) { |
|
m->msg_len = msg->msg_len; |
|
m->msg_base = malloc(m->msg_len); |
|
if (!m->msg_base) { |
|
LOGERR; |
|
free(m); |
|
return NULL; |
|
} else |
|
memcpy(m->msg_base, msg->msg_base, m->msg_len); |
|
} |
|
|
|
return m; |
|
} |
|
|
|
/* |
* mqtt_encodeLen() Encode number to MQTT length field |
* mqtt_encodeLen() Encode number to MQTT length field |
* |
* |
* @num = number for encode |
* @num = number for encode |
Line 270 mqtt_sizeLen(u_int len)
|
Line 302 mqtt_sizeLen(u_int len)
|
} |
} |
|
|
/* |
/* |
|
* mqtt_pktLen() - Get total packet length |
|
* |
|
* @hdr = MQTT packet header |
|
* return: packet length |
|
*/ |
|
inline u_int |
|
mqtt_pktLen(struct mqtthdr * __restrict hdr) |
|
{ |
|
int siz, n = 0; |
|
|
|
if (!hdr) |
|
return 0; |
|
|
|
siz = mqtt_decodeLen(hdr->mqtt_len, &n); |
|
siz += sizeof(struct mqtthdr) + n - 1; |
|
|
|
return siz; |
|
} |
|
|
|
/* |
* mqtt_str2subs Create MQTT subscribe variable from string(s) |
* mqtt_str2subs Create MQTT subscribe variable from string(s) |
* |
* |
* @csStr = null terminated string array |
* @csStr = null terminated string array |
Line 411 mqtt_subCopy(mqtt_subscr_t * __restrict dst, mqtt_subs
|
Line 463 mqtt_subCopy(mqtt_subscr_t * __restrict dst, mqtt_subs
|
memcpy(dst->sub_topic.msg_base, src->sub_topic.msg_base, |
memcpy(dst->sub_topic.msg_base, src->sub_topic.msg_base, |
dst->sub_topic.msg_len); |
dst->sub_topic.msg_len); |
} |
} |
|
} else { |
|
/* |
|
if (dst->sub_topic.msg_base) |
|
free(dst->sub_topic.msg_base); |
|
*/ |
|
dst->sub_topic.msg_base = NULL; |
|
dst->sub_topic.msg_len = 0; |
} |
} |
if (src->sub_value.msg_base) { |
if (src->sub_value.msg_base) { |
dst->sub_value.msg_base = malloc(src->sub_value.msg_len + 1); |
dst->sub_value.msg_base = malloc(src->sub_value.msg_len + 1); |
Line 426 mqtt_subCopy(mqtt_subscr_t * __restrict dst, mqtt_subs
|
Line 485 mqtt_subCopy(mqtt_subscr_t * __restrict dst, mqtt_subs
|
memcpy(dst->sub_value.msg_base, src->sub_value.msg_base, |
memcpy(dst->sub_value.msg_base, src->sub_value.msg_base, |
dst->sub_value.msg_len); |
dst->sub_value.msg_len); |
} |
} |
|
} else { |
|
/* |
|
if (dst->sub_value.msg_base) |
|
free(dst->sub_value.msg_base); |
|
*/ |
|
dst->sub_value.msg_base = NULL; |
|
dst->sub_value.msg_len = 0; |
} |
} |
|
|
dst->sub_ret = src->sub_ret; |
dst->sub_ret = src->sub_ret; |
Line 590 mqtt_sqlTopic(const char *csInput, char * __restrict p
|
Line 656 mqtt_sqlTopic(const char *csInput, char * __restrict p
|
*s = *pos; |
*s = *pos; |
} |
} |
|
|
return ret; |
|
} |
|
|
|
|
|
/* |
|
* mqtt_KeepAlive() - Keep Alive check routine |
|
* |
|
* @sock = connected socket |
|
* @ka = keep alive timeout |
|
* @tries = tries for receive correct ping response, usually ==1 |
|
* return: -1 error, 0 host is alive, 1 timeout session or 2 broken session |
|
*/ |
|
int |
|
mqtt_KeepAlive(int sock, u_short ka, u_char tries) |
|
{ |
|
int ret = 0; |
|
struct pollfd pfd; |
|
mqtt_msg_t msg = { NULL, 0 }; |
|
|
|
if (sock < 3) |
|
return -1; /* error */ |
|
|
|
pfd.fd = sock; |
|
pfd.events = POLLOUT; |
|
if ((ret = poll(&pfd, 1, ka * 1000)) == -1 || |
|
pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { |
|
LOGERR; |
|
return -1; /* error */ |
|
} else if (!ret) |
|
return 1; /* session is abandoned ... must be disconnect! */ |
|
/* ping request */ |
|
if ((ret = mqtt_msgPINGREQ(&msg)) == -1) |
|
return -1; /* error */ |
|
if ((ret = send(sock, msg.msg_base, ret, MSG_NOSIGNAL)) == -1) { |
|
LOGERR; |
|
goto end; |
|
} |
|
|
|
pfd.events = POLLIN | POLLPRI; |
|
while (tries--) { |
|
if ((ret = poll(&pfd, 1, ka * 1000)) == -1 || |
|
pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { |
|
LOGERR; |
|
break; |
|
} else if (!ret) { |
|
ret = 1; /* session is abandoned ... must be disconnect! */ |
|
continue; |
|
} |
|
/* receive & decode packet */ |
|
if ((ret = recv(sock, msg.msg_base, msg.msg_len, 0)) == -1) { |
|
LOGERR; |
|
break; |
|
} |
|
if (!mqtt_readPINGRESP(&msg)) { |
|
ret = 0; /* Host is alive */ |
|
break; |
|
} else |
|
ret = 2; /* Session is broken ... must be disconnect! */ |
|
} |
|
end: |
|
free(msg.msg_base); |
|
return ret; |
return ret; |
} |
} |