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: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>