This repository was archived by the owner on Feb 13, 2021. It is now read-only.
Repository navigation
Expand file tree
/
Copy pathmqtt_wrapper.c
More file actions
219 lines (174 loc) · 6.68 KB
/
Copy pathmqtt_wrapper.c
File metadata and controls
219 lines (174 loc) · 6.68 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
/* vim: set et fenc=utf-8 ff=unix ts=4 sw=4 sts=4 : */
#include "mqtt_wrapper.h"
int mqtt_connect(mqttcd_context_t* context) {
logger_debug(context, "connecting to mqtt broker... ");
// connect to mqtt broker
int sock = transport_open(context->option.host, context->option.port);
if(sock < 0) {
logger_error(context, "couldn't connect to mqtt broker: %d\n", sock);
return MQTTCD_OPEN_FAILED;
}
int ret = mqttcd_process_set_close_on_exec(sock);
if (ret != MQTTCD_SUCCEEDED) {
logger_error(context, "couldn't set close-on-exec\n");
return ret;
}
context->mqtt_socket = sock;
logger_debug(context, "ok\n");
return MQTTCD_SUCCEEDED;
}
void mqtt_disconnect(mqttcd_context_t* context) {
// disconnect
logger_debug(context, "disconnect from mqtt broker... ");
transport_close(context->mqtt_socket);
logger_debug(context, "ok\n");
}
int mqtt_send(mqttcd_context_t* context, unsigned char* buf, int length) {
logger_debug(context, "sending packet... ");
int result = transport_sendPacketBuffer(context->mqtt_socket, buf, length);
if (result != length) {
logger_error(context, "couldn't send packet: %d\n", result);
return MQTTCD_SEND_FAILED;
}
logger_debug(context, "ok\n");
return MQTTCD_SUCCEEDED;
}
int mqtt_recv(mqttcd_context_t* context, unsigned char* buf, int length, int* packet_type) {
logger_debug(context, "receiving packet... ");
// waiting for receiving packet
// transport_getdata() has built-in 1 second timeout
*packet_type = MQTTPacket_read(buf, length, transport_getdata);
if (*packet_type == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
logger_error(context, "timeout\n");
return MQTTCD_RECV_TIMEOUT;
}
logger_error(context, "couldn't receive packet: %d\n", *packet_type);
return MQTTCD_RECV_FAILED;
}
logger_debug(context, "ok: %d\n", *packet_type);
return MQTTCD_SUCCEEDED;
}
int mqtt_initialize_connection(mqttcd_context_t* context) {
unsigned char buf[MQTTCD_BUFFER_LENGTH];
int packet_len;
int result;
int packet_type;
// create packet for connection data
MQTTPacket_connectData conn_data = MQTTPacket_connectData_initializer;
conn_data.MQTTVersion = context->option.version;
conn_data.clientID.cstring = context->option.client_id;
conn_data.keepAliveInterval = MQTTCD_KEEP_ALIVE;
conn_data.cleansession = 1;
conn_data.username.cstring = context->option.username;
conn_data.password.cstring = context->option.password;
logger_debug(context, "serializing connect packet... ");
packet_len = MQTTSerialize_connect(buf, MQTTCD_BUFFER_LENGTH, &conn_data);
if (packet_len <= 0) {
logger_error(context, "couldn't serialize connect packet: %d\n", packet_len);
return MQTTCD_SERIALIZE_FAILED;
}
logger_debug(context, "ok\n");
// send connection data
result = mqtt_send(context, buf, packet_len);
if (result != MQTTCD_SUCCEEDED) {
return result;
}
// receive connack packet from broker
result = mqtt_recv(context, buf, MQTTCD_BUFFER_LENGTH, &packet_type);
if (result != MQTTCD_SUCCEEDED) {
return result;
}
if (packet_type != CONNACK) {
logger_error(context, "couldn't receive connack packet: %d\n", packet_type);
return MQTTCD_PACKET_TYPE_MISMATCHED;
}
// create subscribe packet
int dup = 0;
int msgid = 1;
int count = 1;
MQTTString topic= MQTTString_initializer;
topic.cstring = context->option.topic;
int qos = 0;
logger_debug(context, "serializing subscribe packet... ");
packet_len = MQTTSerialize_subscribe(buf, MQTTCD_BUFFER_LENGTH, dup, msgid, count, &topic, &qos);
if (packet_len <= 0) {
logger_error(context, "couldn't serialize subscribe packet: %d\n", packet_len);
return MQTTCD_SERIALIZE_FAILED;
}
logger_debug(context, "ok\n");
// send subscribe packet
result = mqtt_send(context, buf, packet_len);
if (result != MQTTCD_SUCCEEDED) {
return result;
}
// receive suback packet from broker
result = mqtt_recv(context, buf, MQTTCD_BUFFER_LENGTH, &packet_type);
if (result != MQTTCD_SUCCEEDED) {
return result;
}
if (packet_type != SUBACK) {
logger_error(context, "couldn't receive suback packet: %d\n", packet_type);
return MQTTCD_PACKET_TYPE_MISMATCHED;
}
return MQTTCD_SUCCEEDED;
}
int mqtt_finalize_connection(mqttcd_context_t* context) {
unsigned char buf[MQTTCD_BUFFER_LENGTH];
int packet_len;
int send_result;
// create desconnect packet
logger_debug(context, "serializing disconnect packet... ");
packet_len = MQTTSerialize_disconnect(buf, MQTTCD_BUFFER_LENGTH);
if (packet_len <= 0) {
logger_error(context, "couldn't serialize disconnect packet: %d\n", packet_len);
return MQTTCD_SERIALIZE_FAILED;
}
logger_debug(context, "ok\n");
// send disconnect packet
send_result = mqtt_send(context, buf, packet_len);
if (send_result != MQTTCD_SUCCEEDED) {
return send_result;
}
return MQTTCD_SUCCEEDED;
}
int mqtt_deserialize_publish(mqttcd_context_t* context, unsigned char* buf, int length, char** payload) {
unsigned char pub_dup;
int qos;
unsigned char retained;
unsigned short pub_msgid;
MQTTString receivedTopic;
unsigned char* payload_in;
int payloadlen_in;
logger_debug(context, "deserializing publish packet... ");
int result = MQTTDeserialize_publish(&pub_dup, &qos, &retained, &pub_msgid, &receivedTopic, &payload_in, &payloadlen_in, buf, length);
if (result != 1) {
logger_error(context, "couldn't deserialize publish packet: %d\n", result);
return MQTTCD_DESERIALIZE_FAILED;
}
logger_debug(context, "ok\n");
*payload = malloc(payloadlen_in + 1);
strncpy(*payload, (const char*)payload_in, payloadlen_in);
(*payload)[payloadlen_in] = '\0';
logger_notice(context, "%s\n", *payload);
return MQTTCD_SUCCEEDED;
}
int mqtt_send_ping(mqttcd_context_t* context) {
unsigned char buf[MQTTCD_BUFFER_LENGTH];
int packet_len;
int result;
// create ping packet
logger_debug(context, "serializing pingreq packet... ");
packet_len = MQTTSerialize_pingreq(buf, MQTTCD_BUFFER_LENGTH);
if (packet_len <= 0) {
logger_error(context, "couldn't serialize pingreq packet: %d\n", packet_len);
return MQTTCD_SERIALIZE_FAILED;
}
logger_debug(context, "ok\n");
// send ping packet for keep connection
result = mqtt_send(context, buf, packet_len);
if (result != MQTTCD_SUCCEEDED) {
return result;
}
return MQTTCD_SUCCEEDED;
}