Return to mqttd_calls.c CVS log

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

add reconnect again

    1: #include "global.h"
    2: #include "mqttd.h"
    3: #include "rtlm.h"
    4: #include "mqttd_calls.h"
    5: 
    6: 
    7: int
    8: cmdPUBLISH(void *srv, int len, void *arg)
    9: {
   10: 	struct mqtthdr *hdr;
   11: 	struct tagSession *sess = (struct tagSession*) arg;
   12: 
   13: 	ioTRACE(2);
   14: 
   15: 	if (!sess)
   16: 		return -1;
   17: 
   18: 	hdr = (struct mqtthdr*) sess->sess_buf->msg_base;
   19: 	switch (hdr->mqtt_msg.qos) {
   20: 		case MQTT_QOS_ONCE:
   21: 			break;
   22: 		case MQTT_QOS_ACK:
   23: 			break;
   24: 		case MQTT_QOS_EXACTLY:
   25: 			break;
   26: 		default:
   27: 			ioDEBUG(1, "Error:: Unknown QoS %d - rejected publishing request", 
   28: 					hdr->mqtt_msg.qos);
   29: 			return 0;
   30: 	}
   31: 
   32: 	return 0;
   33: }
   34: 
   35: int
   36: cmdPUBREL(void *srv, int len, void *arg)
   37: {
   38: 	struct mqtthdr *hdr;
   39: 	struct tagSession *sess = (struct tagSession*) arg;
   40: 
   41: 	ioTRACE(2);
   42: 
   43: 	if (!sess)
   44: 		return -1;
   45: 
   46: 	hdr = (struct mqtthdr*) sess->sess_buf->msg_base;
   47: 
   48: 	return 0;
   49: }
   50: 
   51: int
   52: cmdSUBSCRIBE(void *srv, int len, void *arg)
   53: {
   54: 	struct tagSession *sess = (struct tagSession*) arg;
   55: 	mqtt_subscr_t *subs = NULL;
   56: 	int siz = 0;
   57: 	u_short mid = 0;
   58: 	register int i;
   59: 	struct tagStore *store;
   60: 
   61: 	ioTRACE(2);
   62: 
   63: 	if (!sess)
   64: 		return -1;
   65: 
   66: 	ioDEBUG(5, "Exec SUBSCRIBE session");
   67: 	siz = mqtt_readSUBSCRIBE(sess->sess_buf, &mid, &subs);
   68: 	if (siz == -1) {
   69: 		ioDEBUG(5, "Error:: in readSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError());
   70: 		return 0;
   71: 	}
   72: 
   73: 	/* add to db */
   74: 	for (i = 0; i < siz; i++) {
   75: 		if (call.WritePUB_subscribe(&cfg, pub, mid, subs[i].sub_topic.msg_base, 
   76: 				sess->sess_user, sess->sess_addr, subs[i].sub_ret) > 0) {
   77: 			store = malloc(sizeof(struct tagStore));
   78: 			if (!store) {
   79: 				ioSYSERR(0);
   80: 				goto end;
   81: 			} else {
   82: 				store->st_msgid = mid;
   83: 				mqtt_subCopy(&store->st_subscr, &subs[i]);
   84: 			}
   85: 
   86: 			/* add to cache */
   87: 			SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node);
   88: 
   89: 			subs[i].sub_ret = MQTT_QOS_PASS;
   90: 		} else
   91: 			subs[i].sub_ret = MQTT_QOS_DENY;
   92: 	}
   93: 
   94: 	/* send acknowledge */
   95: 	siz = mqtt_msgSUBACK(sess->sess_buf, subs, mid);
   96: 	if (siz == -1) {
   97: 		ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError());
   98: 		goto end;
   99: 	}
  100: 	if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1)
  101: 		ioSYSERR(0);
  102: 	else {
  103: 		ioDEBUG(5, "Sended %d bytes.", siz);
  104: 		memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len);
  105: 	}
  106: end:
  107: 	mqtt_subFree(&subs);
  108: 	return 0;
  109: }
  110: 
  111: int
  112: cmdUNSUBSCRIBE(void *srv, int len, void *arg)
  113: {
  114: 	struct tagSession *sess = (struct tagSession*) arg;
  115: 
  116: 	ioTRACE(2);
  117: 
  118: 	if (!sess)
  119: 		return -1;
  120: 
  121: 	return 0;
  122: }
  123: 
  124: int
  125: cmdPINGREQ(void *srv, int len, void *arg)
  126: {
  127: 	struct tagSession *sess = (struct tagSession*) arg;
  128: 	int siz = 0;
  129: 
  130: 	ioTRACE(2);
  131: 
  132: 	if (!sess)
  133: 		return -1;
  134: 
  135: 	ioDEBUG(5, "Exec PINGREQ session");
  136: 	siz = mqtt_msgPINGRESP(sess->sess_buf);
  137: 	if (siz == -1) {
  138: 		ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError());
  139: 		return 0;
  140: 	}
  141: 	if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1) {
  142: 		ioSYSERR(0);
  143: 		return 0;
  144: 	} else {
  145: 		ioDEBUG(5, "Sended %d bytes.", siz);
  146: 		memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len);
  147: 	}
  148: 
  149: 	return 0;
  150: }
  151: 
  152: int
  153: cmdCONNECT(void *srv, int len, void *arg)
  154: {
  155: 	struct tagStore *store;
  156: 	struct tagSession *sess = (struct tagSession*) arg;
  157: 
  158: 	ioTRACE(2);
  159: 
  160: 	if (!sess)
  161: 		return -1;
  162: 
  163: 	ioDEBUG(5, "Exec CONNECT session");
  164: 	TAILQ_REMOVE(&Sessions, sess, sess_node);
  165: 
  166: 	if (call.FiniSessPUB)
  167: 		call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
  168: 
  169: 	while ((store = SLIST_FIRST(&sess->sess_subscr))) {
  170: 		SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);
  171: 
  172: 		if (store->st_subscr.sub_topic.msg_base)
  173: 			free(store->st_subscr.sub_topic.msg_base);
  174: 		if (store->st_subscr.sub_value.msg_base)
  175: 			free(store->st_subscr.sub_value.msg_base);
  176: 
  177: 		free(store);
  178: 	}
  179: 
  180: 	if (sess->sess_will.msg)
  181: 		free(sess->sess_will.msg);
  182: 	if (sess->sess_will.topic)
  183: 		free(sess->sess_will.topic);
  184: 
  185: 	call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, 
  186: 			sess->sess_addr, sess->sess_user);
  187: 
  188: 	return -3;	/* reconnect client */
  189: }
  190: 
  191: int
  192: cmdDISCONNECT(void *srv, int len, void *arg)
  193: {
  194: 	struct tagSession *sess = (struct tagSession*) arg;
  195: 
  196: 	ioTRACE(2);
  197: 
  198: 	if (!sess)
  199: 		return -1;
  200: 
  201: 	ioDEBUG(5, "Exec DISCONNECT session");
  202: 
  203: 	call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, 
  204: 			sess->sess_addr, sess->sess_user);
  205: 
  206: 	return -2;	/* must terminate dispatcher */
  207: }