--- mqtt/src/daemon.c 2012/06/29 08:47:28 1.2.2.40 +++ mqtt/src/daemon.c 2012/06/29 13:46:37 1.2.2.41 @@ -179,6 +179,32 @@ dispatchSession(sched_task_t *task) return NULL; } +static int +loadSubscribes(struct tagSession * __restrict sess, mqtt_subscr_t * __restrict subs) +{ + register int i; + struct tagStore *store; + + if (!subs) + return -1; + + for (i = 0; subs[i].sub_topic.msg_base; i++) { + store = io_malloc(sizeof(struct tagStore)); + if (!store) { + ioSYSERR(0); + continue; + } else { + store->st_msgid = 0; + mqtt_subCopy(&store->st_subscr, &subs[i]); + } + + /* add to cache */ + SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); + } + + return 0; +} + void * startSession(sched_task_t *task) { @@ -189,6 +215,7 @@ startSession(sched_task_t *task) ait_val_t *v; struct tagSession *s, *sess = NULL; int ret, wlen; + mqtt_subscr_t *subs; ioTRACE(4); @@ -274,8 +301,10 @@ startSession(sched_task_t *task) call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); if (call.WipePUB_topic) call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); - } else { - // TODO: read_sql subs and prepare publish + } else if (call.ReadPUB_subscribe) { + subs = call.ReadPUB_subscribe(&cfg, pub, sess->sess_cid, "%"); + loadSubscribes(sess, subs); + mqtt_subFree(&subs); } /* Start session task OK ... */