version 1.2.2.22, 2012/06/19 08:32:17
|
version 1.2.2.24, 2012/06/20 14:51:38
|
Line 87 pubOnce(struct tagSession *sess, char * __restrict psT
|
Line 87 pubOnce(struct tagSession *sess, char * __restrict psT
|
} |
} |
if (!regexec(&re, psTopic, 1, &match, 0)) { |
if (!regexec(&re, psTopic, 1, &match, 0)) { |
/* MATCH */ |
/* MATCH */ |
|
ioDEBUG(1, "+++ dlen=%d\n", datlen); |
p = mkPkt(sess->sess_buf->msg_base, datlen); |
p = mkPkt(sess->sess_buf->msg_base, datlen); |
schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); | schedWrite(root, sendPacket, p, s->sess_sock, NULL, 0); |
} |
} |
|
|
regfree(&re); |
regfree(&re); |
Line 145 cmdPUBLISH(void *srv, int len, void *arg)
|
Line 146 cmdPUBLISH(void *srv, int len, void *arg)
|
hdr = (struct mqtthdr*) sess->sess_buf->msg_base; |
hdr = (struct mqtthdr*) sess->sess_buf->msg_base; |
switch (hdr->mqtt_msg.qos) { |
switch (hdr->mqtt_msg.qos) { |
case MQTT_QOS_ACK: |
case MQTT_QOS_ACK: |
pubAck(sess, szTopic, siz); | pubAck(sess, szTopic, mqtt_pktLen(hdr)); |
siz = mqtt_msgPUBACK(sess->sess_buf, mid); |
siz = mqtt_msgPUBACK(sess->sess_buf, mid); |
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", |
ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", |
Line 154 cmdPUBLISH(void *srv, int len, void *arg)
|
Line 155 cmdPUBLISH(void *srv, int len, void *arg)
|
} |
} |
break; |
break; |
case MQTT_QOS_EXACTLY: |
case MQTT_QOS_EXACTLY: |
pubExactly(sess, szTopic, siz); | pubExactly(sess, szTopic, mqtt_pktLen(hdr)); |
siz = mqtt_msgPUBREC(sess->sess_buf, mid); |
siz = mqtt_msgPUBREC(sess->sess_buf, mid); |
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", |
ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", |
Line 163 cmdPUBLISH(void *srv, int len, void *arg)
|
Line 164 cmdPUBLISH(void *srv, int len, void *arg)
|
} |
} |
break; |
break; |
case MQTT_QOS_ONCE: |
case MQTT_QOS_ONCE: |
pubOnce(sess, szTopic, siz); | pubOnce(sess, szTopic, mqtt_pktLen(hdr)); |
default: |
default: |
return 0; |
return 0; |
} |
} |
Line 256 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
Line 257 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); |
SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); |
|
|
/* convert topic to regexp */ |
/* convert topic to regexp */ |
if (mqtt_expandTopic(subs[i].sub_topic.msg_base, buf, sizeof buf, 1, 0) == -1) { | if (mqtt_expandTopic(subs[i].sub_topic.msg_base, buf, sizeof buf, 1, 1) == -1) { |
ioDEBUG(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
goto end; | |
| subs[i].sub_ret = MQTT_QOS_DENY; |
} else { |
} else { |
ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); |
ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); |
if (!ptr) { |
if (!ptr) { |
ioSYSERR(0); |
ioSYSERR(0); |
goto end; | |
| subs[i].sub_ret = MQTT_QOS_DENY; |
} else { |
} else { |
store->st_subscr.sub_topic.msg_base = ptr; |
store->st_subscr.sub_topic.msg_base = ptr; |
store->st_subscr.sub_topic.msg_len = strlen(buf) + 1; |
store->st_subscr.sub_topic.msg_len = strlen(buf) + 1; |
memcpy(store->st_subscr.sub_topic.msg_base, buf, |
memcpy(store->st_subscr.sub_topic.msg_base, buf, |
store->st_subscr.sub_topic.msg_len); |
store->st_subscr.sub_topic.msg_len); |
} |
} |
} |
|
|
|
call.LOG(logg, "Added [%s] SUBSCRIBE '%s'(%d) from %s\n", sess->sess_cid, | call.LOG(logg, "Added [%s] SUBSCRIBE '%s'(%d) from %s\n", sess->sess_cid, |
store->st_subscr.sub_topic.msg_base, | store->st_subscr.sub_topic.msg_base, |
store->st_subscr.sub_topic.msg_len, sess->sess_addr); | store->st_subscr.sub_topic.msg_len, sess->sess_addr); |
|
|
subs[i].sub_ret = MQTT_QOS_PASS; | subs[i].sub_ret = MQTT_QOS_PASS; |
| } |
} else |
} else |
subs[i].sub_ret = MQTT_QOS_DENY; |
subs[i].sub_ret = MQTT_QOS_DENY; |
} |
} |