[BACK]Return to daemon.c CVS log [TXT] [DIR] Up to [ELWIX - Embedded LightWeight unIX -] / mqtt / src

Diff for /mqtt/src/daemon.c between versions 1.2.2.26 and 1.2.2.30

version 1.2.2.26, 2012/05/05 15:04:29 version 1.2.2.30, 2012/05/08 14:36:10
Line 39  initSession(int sock, ait_val_t * __restrict v) Line 39  initSession(int sock, ait_val_t * __restrict v)
         }          }
   
         /* init server actor */          /* init server actor */
        sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf);        sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf, sess->sess_ka);
         if (!sess->sess_srv) {          if (!sess->sess_srv) {
                 ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());                  ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
                 mqtt_msgFree(&sess->sess_buf, 42);                  mqtt_msgFree(&sess->sess_buf, 42);
Line 70  finiSession(struct tagSession *sess) Line 70  finiSession(struct tagSession *sess)
         if (!sess)          if (!sess)
                 return;                  return;
   
        if (call.FiniSessPUB)        if (sess->sess_clean) {
                call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");                if (call.FiniSessPUB)
                         call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
                 if (call.DeletePUB_subscribe)
                         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);
         }
   
         while ((store = SLIST_FIRST(&sess->sess_subscr))) {          while ((store = SLIST_FIRST(&sess->sess_subscr))) {
                 SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);                  SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);
Line 153  dispatchSession(sched_task_t *task) Line 159  dispatchSession(sched_task_t *task)
         do {          do {
                 /* dispatch message type */                  /* dispatch message type */
                 if ((len = mqtt_srv_Dispatch(sess->sess_srv, ret, sess)) < 0) {                  if ((len = mqtt_srv_Dispatch(sess->sess_srv, ret, sess)) < 0) {
                        if (len == -1)                        if (len == -1) {
                                 ioLIBERR(mqtt);                                  ioLIBERR(mqtt);
                        TAILQ_REMOVE(&Sessions, sess, sess_node);                                finiSession(sess);
                        finiSession(sess);                        } else if (len == -2) {
                                 TAILQ_REMOVE(&Sessions, sess, sess_node);
                                 finiSession(sess);
                         } else if (len == -3)
                                 schedEvent(root, startSession, NULL, (u_long) TASK_FD(task), sess, ret);
                 } else                  } else
                         ret -= len;                          ret -= len;
         } while (len > 0 && ret > 0);          } while (len > 0 && ret > 0);
Line 219  startSession(sched_task_t *task) Line 229  startSession(sched_task_t *task)
                 sess->sess_will.qos = flg.will_qos;                  sess->sess_will.qos = flg.will_qos;
                 sess->sess_will.retain = flg.will_retain;                  sess->sess_will.retain = flg.will_retain;
                 sess->sess_will.flag = flg.will_flg;                  sess->sess_will.flag = flg.will_flg;
   
                   sess->sess_srv->timeout = sess->sess_ka;
         }          }
   
         /* check online table for user */          /* check online table for user */
Line 232  startSession(sched_task_t *task) Line 244  startSession(sched_task_t *task)
                 ret = MQTT_RETCODE_ACCEPTED;                  ret = MQTT_RETCODE_ACCEPTED;
         }          }
   
           /* db management */
         if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) {          if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) {
                 ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid);                  ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid);
                 TAILQ_FOREACH(s, &Sessions, sess_node)                  TAILQ_FOREACH(s, &Sessions, sess_node)
Line 241  startSession(sched_task_t *task) Line 254  startSession(sched_task_t *task)
                                 break;                                  break;
                         }                          }
         }          }
   
         if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr,           if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr, 
                                 sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg,                                   sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg, 
                                 sess->sess_will.qos, sess->sess_will.retain) == -1) {                                  sess->sess_will.qos, sess->sess_will.retain) == -1) {
Line 252  startSession(sched_task_t *task) Line 264  startSession(sched_task_t *task)
                 ioDEBUG(0, "Session %s from %s and username %s is started",                   ioDEBUG(0, "Session %s from %s and username %s is started", 
                                 sess->sess_cid, sess->sess_addr, sess->sess_user);                                  sess->sess_cid, sess->sess_addr, sess->sess_user);
                 ret = MQTT_RETCODE_ACCEPTED;                  ret = MQTT_RETCODE_ACCEPTED;
           }
           /* clean/load session if requested */
           if (sess->sess_clean) {
                   if (call.DeletePUB_subscribe)
                           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
         }          }
   
         /* Start session task OK ... */          /* Start session task OK ... */
Removed from v.1.2.2.26  
changed lines
  Added in v.1.2.2.30

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>