[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.1.2.32 and 1.2.2.7

version 1.1.2.32, 2012/01/25 10:34:14 version 1.2.2.7, 2012/04/15 22:06:10
Line 27  initSession(int sock, ait_val_t * __restrict v) Line 27  initSession(int sock, ait_val_t * __restrict v)
         } else          } else
                 memset(sess, 0, sizeof(struct tagSession));                  memset(sess, 0, sizeof(struct tagSession));
   
        TAILQ_INIT(&sess->sess_sndqueue);        pthread_mutex_init(&sess->sess_mtx, NULL);
   
        str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry"));        SLIST_INIT(&sess->sess_txque[MQTT_QOS_ONCE]);
         SLIST_INIT(&sess->sess_txque[MQTT_QOS_ACK]);
         SLIST_INIT(&sess->sess_txque[MQTT_QOS_EXACTLY]);
 
         str = cfg_getAttribute(&cfg, "mqttd", "retry");
         if (!str)          if (!str)
                 sess->sess_retry = DEFAULT_RETRY;                  sess->sess_retry = DEFAULT_RETRY;
         else          else
Line 43  initSession(int sock, ait_val_t * __restrict v) Line 47  initSession(int sock, ait_val_t * __restrict v)
                 return NULL;                  return NULL;
         }          }
   
           sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf);
           if (!sess->sess_srv) {
                   ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
                   mqtt_msgFree(&sess->sess_buf, 42);
                   free(sess);
                   io_freeVar(v);
                   return NULL;
           } else {
                   mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT);
                   mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBLISH, cmdPUBLISH);
                   mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBREL, cmdPUBREL);
                   mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_SUBSCRIBE, cmdSUBSCRIBE);
                   mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_UNSUBSCRIBE, cmdUNSUBSCRIBE);
                   mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PINGREQ, cmdPINGREQ);
                   mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_DISCONNECT, cmdDISCONNECT);
           }
   
         sess->sess_sock = sock;          sess->sess_sock = sock;
         strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr);          strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr);
         io_freeVar(v);          io_freeVar(v);
Line 53  static void Line 74  static void
 finiSession(struct tagSession *sess, int preservSock)  finiSession(struct tagSession *sess, int preservSock)
 {  {
         struct tagStore *store;          struct tagStore *store;
           register int i;
   
         ioTRACE(5);          ioTRACE(5);
   
Line 62  finiSession(struct tagSession *sess, int preservSock) Line 84  finiSession(struct tagSession *sess, int preservSock)
         if (call.FiniSessPUB)          if (call.FiniSessPUB)
                 call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");                  call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
   
        while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) {        SESS_ELEM_LOCK(sess);
                TAILQ_REMOVE(&sess->sess_sndqueue, store, st_node);        for (i = 0; i < MQTT_QOS_RESERVED; i++)
                 while ((store = SLIST_FIRST(&sess->sess_txque[i]))) {
                         SLIST_REMOVE_HEAD(&sess->sess_txque[i], st_node);
   
                if (store->st_subscr.sub_topic._base)                        if (store->st_subscr.sub_topic.msg_base)
                        free(store->st_subscr.sub_topic._base);                                free(store->st_subscr.sub_topic.msg_base);
                if (store->st_subscr.sub_value._base)                        if (store->st_subscr.sub_value.msg_base)
                        free(store->st_subscr.sub_value._base);                                free(store->st_subscr.sub_value.msg_base);
   
                free(store);                        free(store);
        }                }
         SESS_ELEM_UNLOCK(sess);
         pthread_mutex_destroy(&sess->sess_mtx);
   
         if (sess->sess_will.msg)          if (sess->sess_will.msg)
                 free(sess->sess_will.msg);                  free(sess->sess_will.msg);
Line 81  finiSession(struct tagSession *sess, int preservSock) Line 107  finiSession(struct tagSession *sess, int preservSock)
         if (sess->sess_sock > STDERR_FILENO && !preservSock)          if (sess->sess_sock > STDERR_FILENO && !preservSock)
                 srv_Close(sess->sess_sock);                  srv_Close(sess->sess_sock);
   
           mqtt_srv_Fini(&sess->sess_srv);
         mqtt_msgFree(&sess->sess_buf, 42);          mqtt_msgFree(&sess->sess_buf, 42);
   
         free(sess);          free(sess);
Line 96  stopSession(struct tagSession *sess) Line 123  stopSession(struct tagSession *sess)
   
         assert(sess);          assert(sess);
   
        pthread_mutex_lock(&mtx_sess);        SESS_LOCK;
         TAILQ_REMOVE(&Sessions, sess, sess_node);          TAILQ_REMOVE(&Sessions, sess, sess_node);
        pthread_mutex_unlock(&mtx_sess);        SESS_UNLOCK;
   
         ret = mqtt_msgDISCONNECT(&msg);          ret = mqtt_msgDISCONNECT(&msg);
         if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1)          if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1)
Line 165  KASession(struct tagSession *sess) Line 192  KASession(struct tagSession *sess)
 static void *  static void *
 thrSession(struct tagSession *sess)  thrSession(struct tagSession *sess)
 {  {
         mqtt_msg_t msg = { NULL, 0 };  
         int ret, locKill = 42;          int ret, locKill = 42;
         struct pollfd pfd;          struct pollfd pfd;
         struct mqtthdr *hdr;          struct mqtthdr *hdr;
         ait_val_t *v;          ait_val_t *v;
         struct tagStore *store;  
   
         pthread_cleanup_push((void(*)(void*)) stopSession, sess);          pthread_cleanup_push((void(*)(void*)) stopSession, sess);
         ioTRACE(2);          ioTRACE(2);
Line 198  thrSession(struct tagSession *sess) Line 223  thrSession(struct tagSession *sess)
                         hdr = (struct mqtthdr*) sess->sess_buf->msg_base;                          hdr = (struct mqtthdr*) sess->sess_buf->msg_base;
   
                 /* dispatch message type */                  /* dispatch message type */
                   if (mqtt_srv_Dispatch(sess->sess_srv, sess))
                           ioLIBERR(mqtt);
                 switch (hdr->mqtt_msg.type) {                  switch (hdr->mqtt_msg.type) {
                         case MQTT_TYPE_CONNECT:                          case MQTT_TYPE_CONNECT:
                                 ioDEBUG(5, "Exec CONNECT session");                                  ioDEBUG(5, "Exec CONNECT session");
                                 pthread_mutex_lock(&mtx_sess);  
                                 TAILQ_REMOVE(&Sessions, sess, sess_node);  
                                 pthread_mutex_unlock(&mtx_sess);  
   
                                 if (call.FiniSessPUB)  
                                         call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");  
   
                                 while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) {  
                                         TAILQ_REMOVE(&sess->sess_sndqueue, store, st_node);  
   
                                         if (store->st_subscr.sub_topic._base)  
                                                 free(store->st_subscr.sub_topic._base);  
                                         if (store->st_subscr.sub_value._base)  
                                                 free(store->st_subscr.sub_value._base);  
   
                                         free(store);  
                                 }  
   
                                 if (sess->sess_will.msg)  
                                         free(sess->sess_will.msg);  
                                 if (sess->sess_will.topic)  
                                         free(sess->sess_will.topic);  
   
                                 if ((v = io_allocVar())) {                                  if ((v = io_allocVar())) {
                                         AIT_SET_STR(v, sess->sess_addr);                                          AIT_SET_STR(v, sess->sess_addr);
                                         if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret))                                          if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret))
                                                 io_freeVar(v);                                                  io_freeVar(v);
                                 } else                                  } else
                                         ioLIBERR(mqtt);                                          ioLIBERR(mqtt);
   
                                 locKill ^= locKill;                                  locKill ^= locKill;
   
                                 call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid,   
                                                 sess->sess_addr, sess->sess_user);  
                                 continue;                                  continue;
                         case MQTT_TYPE_DISCONNECT:                          case MQTT_TYPE_DISCONNECT:
                                 ioDEBUG(5, "Exec DISCONNECT session");                                  ioDEBUG(5, "Exec DISCONNECT session");
                                 pthread_mutex_lock(&mtx_sess);  
                                 TAILQ_REMOVE(&Sessions, sess, sess_node);  
                                 pthread_mutex_unlock(&mtx_sess);  
   
                                 finiSession(sess, 0);                                  finiSession(sess, 0);
                                 locKill ^= locKill;                                  locKill ^= locKill;
   
                                 call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid,   
                                                 sess->sess_addr, sess->sess_user);  
                                 continue;                                  continue;
                         case MQTT_TYPE_PUBLISH:                          case MQTT_TYPE_PUBLISH:
                                 ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos);                                  ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos);
                                if (Publish(sess))                                /*
                                 if (cmdPUBLISH(sess))
                                         locKill ^= locKill;                                          locKill ^= locKill;
                                           */
                                 break;                                  break;
                         case MQTT_TYPE_PUBREL:                          case MQTT_TYPE_PUBREL:
                                 break;                                  break;
Line 260  thrSession(struct tagSession *sess) Line 255  thrSession(struct tagSession *sess)
                         case MQTT_TYPE_UNSUBSCRIBE:                          case MQTT_TYPE_UNSUBSCRIBE:
                                 break;                                  break;
                         case MQTT_TYPE_PINGREQ:                          case MQTT_TYPE_PINGREQ:
                                   ioDEBUG(5, "Exec PINGREQ session");
                                 break;                                  break;
                         default:                          default:
                                 ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED",                                   ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", 
Line 368  startSession(sched_task_t *task) Line 364  startSession(sched_task_t *task)
         pthread_attr_init(&attr);          pthread_attr_init(&attr);
         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);          pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
   
        pthread_mutex_lock(&mtx_sess);        SESS_LOCK;
         TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);          TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);
         pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess);          pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess);
        pthread_mutex_unlock(&mtx_sess);        SESS_UNLOCK;
   
         call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid,           call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid, 
                         sess->sess_addr, sess->sess_user, sess->sess_ka);                          sess->sess_addr, sess->sess_user, sess->sess_ka);
Line 425  Run(int sock) Line 421  Run(int sock)
   
         ioTRACE(1);          ioTRACE(1);
   
           /* start alternative thread for scheduler */
         if (pthread_create(&tid, NULL, thrSched, NULL)) {          if (pthread_create(&tid, NULL, thrSched, NULL)) {
                 ioSYSERR(0);                  ioSYSERR(0);
                 return -1;                  return -1;
Line 446  Run(int sock) Line 443  Run(int sock)
                 }                  }
                 v = io_allocVar();                  v = io_allocVar();
                 if (!v) {                  if (!v) {
                        ioLIBERR(mqtt);                        ioLIBERR(io);
                         break;                          break;
                 } else {                  } else {
                         memset(str, 0, sizeof str);                          memset(str, 0, sizeof str);
                         snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));                          snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));
                           AIT_FREE_VAL(v);
                         AIT_SET_STR(v, str);                          AIT_SET_STR(v, str);
                 }                  }
                 ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));                  ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));
Removed from v.1.1.2.32  
changed lines
  Added in v.1.2.2.7

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