mqtt/src/daemon.c - diff

Return to daemon.c CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / mqtt / src

Diff for /mqtt/src/daemon.c between versions 1.1.2.25 and 1.1.2.27

-version 1.1.2.25, 2011/12/16 12:22:26
+version 1.1.2.27, 2011/12/16 13:08:46
  Line 79  stopSession(struct tagSession *sess)
          mqtt_msg_t msg = { NULL, 0 };
          int ret;
+         ioTRACE(4);
+         assert(sess);
          pthread_mutex_lock(&mtx_sess);
          TAILQ_REMOVE(&Sessions, sess, sess_node);
          pthread_mutex_unlock(&mtx_sess);
- Line 89  stopSession(struct tagSession *sess)
+ Line 93  stopSession(struct tagSession *sess)
          else {
                  ioDEBUG(5, "Sended %d bytes for disconnect", ret);
                  free(msg.msg_base);
-                msg.msg_len = 0;
+                memset(&msg, 0, sizeof msg);
          }
          ioDEBUG(1, "Close socket=%d", sess->sess_sock);
          finiSession(sess);
+         call.LOG(logg, "Session %s stopped from %s for user %s OK!\n", sess->sess_cid,
+                         sess->sess_addr, sess->sess_user);
  }
+ static int
+ KASession(struct tagSession *sess)
+ {
+         u_char basebuf[USHRT_MAX];
+         mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf };
+         int ret;
+         struct pollfd pfd;
+         ioTRACE(4);
+         assert(sess);
+         /* ping request */
+         ret = mqtt_msgPINGREQ(&msg);
+         if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {
+                 ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
+                 stopSession(sess);
+                 return -1;
+         } else {
+                 ioDEBUG(5, "Sended %d bytes for ping request", ret);
+                 free(msg.msg_base);
+                 memset(&msg, 0, sizeof msg);
+         }
+         pfd.fd = sess->sess_sock;
+         pfd.events = POLLIN | POLLPRI;
+         if ((ret = poll(&pfd, 1, sess->sess_ka)) == -1) {
+                 ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
+                 stopSession(sess);
+                 return -1;
+         } else if (!ret) {
+                 /* problem!!! session is abandoned ... must be disconnect! */
+                 stopSession(sess);
+                 return 1;
+         }
+         /* receive & decode packet */
+         if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {
+                 ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
+                 stopSession(sess);
+                 return -1;
+         }
+         if (mqtt_readPINGRESP(&buf)) {
+                 /* problem!!! session is broken, not hear ping response ... must be disconnect! */
+                 stopSession(sess);
+                 return 2;
+         }
+         /* Keep Alive is OK! */
+         return 0;
+ }
  static void *
  thrSession(struct tagSession *sess)
  {
- Line 120  startSession(sched_task_t *task)
+ Line 178  startSession(sched_task_t *task)
          ioTRACE(4);
+         assert(task);
          sess = initSession(TASK_FD(task), TASK_ARG(task));
          if (!sess) {
                  io_freeVar(TASK_ARG(task));
- Line 187  startSession(sched_task_t *task)
+ Line 247  startSession(sched_task_t *task)
          } else {
                  ioDEBUG(5, "Sended %d bytes", ret);
                  free(msg.msg_base);
-                msg.msg_len = 0;
+                memset(&msg, 0, sizeof msg);
          }
          /* Start session thread OK ... */
- Line 211  end: /* close client connection */
+ Line 271  end: /* close client connection */
          } else {
                  ioDEBUG(5, "Sended %d bytes", ret);
                  free(msg.msg_base);
-                msg.msg_len = 0;
+                memset(&msg, 0, sizeof msg);
          }
          ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock);

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

Removed from v.1.1.2.25
changed lines
	Added in v.1.1.2.27