-
Notifications
You must be signed in to change notification settings - Fork 2.4k
/
sub_client.c
367 lines (318 loc) · 12.3 KB
/
sub_client.c
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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
/*
Copyright (c) 2009-2019 Roger Light <[email protected]>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License v1.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
http:https://www.eclipse.org/legal/epl-v10.html
and the Eclipse Distribution License is available at
http:https://www.eclipse.org/org/documents/edl-v10.php.
Contributors:
Roger Light - initial implementation and documentation.
*/
#include "config.h"
#include <assert.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#ifndef WIN32
#include <unistd.h>
#include <signal.h>
#else
#include <process.h>
#include <winsock2.h>
#define snprintf sprintf_s
#endif
#include <mosquitto.h>
#include <mqtt_protocol.h>
#include "client_shared.h"
struct mosq_config cfg;
bool process_messages = true;
int msg_count = 0;
struct mosquitto *mosq = NULL;
int last_mid = 0;
#ifndef WIN32
void my_signal_handler(int signum)
{
if(signum == SIGALRM){
process_messages = false;
mosquitto_disconnect_v5(mosq, MQTT_RC_DISCONNECT_WITH_WILL_MSG, cfg.disconnect_props);
}
}
#endif
void print_message(struct mosq_config *cfg, const struct mosquitto_message *message);
void my_publish_callback(struct mosquitto *mosq, void *obj, int mid, int reason_code, const mosquitto_property *properties)
{
UNUSED(obj);
UNUSED(reason_code);
UNUSED(properties);
if(process_messages == false && (mid == last_mid || last_mid == 0)){
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
}
void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message, const mosquitto_property *properties)
{
int i;
bool res;
UNUSED(obj);
UNUSED(properties);
if(process_messages == false) return;
if(cfg.remove_retained && message->retain){
mosquitto_publish(mosq, &last_mid, message->topic, 0, NULL, 1, true);
}
if(cfg.retained_only && !message->retain && process_messages){
process_messages = false;
if(last_mid == 0){
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
return;
}
if(message->retain && cfg.no_retain) return;
if(cfg.filter_outs){
for(i=0; i<cfg.filter_out_count; i++){
mosquitto_topic_matches_sub(cfg.filter_outs[i], message->topic, &res);
if(res) return;
}
}
print_message(&cfg, message);
if(cfg.msg_count>0){
msg_count++;
if(cfg.msg_count == msg_count){
process_messages = false;
if(last_mid == 0){
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
}
}
}
void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flags, const mosquitto_property *properties)
{
int i;
UNUSED(obj);
UNUSED(flags);
UNUSED(properties);
if(!result){
mosquitto_subscribe_multiple(mosq, NULL, cfg.topic_count, cfg.topics, cfg.qos, cfg.sub_opts, cfg.subscribe_props);
for(i=0; i<cfg.unsub_topic_count; i++){
mosquitto_unsubscribe_v5(mosq, NULL, cfg.unsub_topics[i], cfg.unsubscribe_props);
}
}else{
if(result){
if(cfg.protocol_version == MQTT_PROTOCOL_V5){
err_printf(&cfg, "%s\n", mosquitto_reason_string(result));
}else{
err_printf(&cfg, "%s\n", mosquitto_connack_string(result));
}
}
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
}
void my_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
int i;
UNUSED(obj);
if(!cfg.quiet) printf("Subscribed (mid: %d): %d", mid, granted_qos[0]);
for(i=1; i<qos_count; i++){
if(!cfg.quiet) printf(", %d", granted_qos[i]);
}
if(!cfg.quiet) printf("\n");
if(cfg.exit_after_sub){
mosquitto_disconnect_v5(mosq, 0, cfg.disconnect_props);
}
}
void my_log_callback(struct mosquitto *mosq, void *obj, int level, const char *str)
{
UNUSED(mosq);
UNUSED(obj);
UNUSED(level);
printf("%s\n", str);
}
void print_usage(void)
{
int major, minor, revision;
mosquitto_lib_version(&major, &minor, &revision);
printf("mosquitto_sub is a simple mqtt client that will subscribe to a set of topics and print all messages it receives.\n");
printf("mosquitto_sub version %s running on libmosquitto %d.%d.%d.\n\n", VERSION, major, minor, revision);
printf("Usage: mosquitto_sub {[-h host] [-p port] [-u username] [-P password] -t topic | -L URL [-t topic]}\n");
printf(" [-c] [-k keepalive] [-q qos]\n");
printf(" [-C msg_count] [-E] [-R] [--retained-only] [--remove-retained] [-T filter_out] [-U topic ...]\n");
printf(" [-F format]\n");
#ifndef WIN32
printf(" [-W timeout_secs]\n");
#endif
#ifdef WITH_SRV
printf(" [-A bind_address] [-S]\n");
#else
printf(" [-A bind_address]\n");
#endif
printf(" [-i id] [-I id_prefix]\n");
printf(" [-d] [-N] [--quiet] [-v]\n");
printf(" [--will-topic [--will-payload payload] [--will-qos qos] [--will-retain]]\n");
#ifdef WITH_TLS
printf(" [{--cafile file | --capath dir} [--cert file] [--key file]\n");
printf(" [--ciphers ciphers] [--insecure]\n");
printf(" [--tls-alpn protocol]\n");
printf(" [--tls-engine engine] [--keyform keyform] [--tls-engine-kpass-sha1]]\n");
#ifdef FINAL_WITH_TLS_PSK
printf(" [--psk hex-key --psk-identity identity [--ciphers ciphers]]\n");
#endif
#endif
#ifdef WITH_SOCKS
printf(" [--proxy socks-url]\n");
#endif
printf(" [-D command identifier value]\n");
printf(" mosquitto_sub --help\n\n");
printf(" -A : bind the outgoing socket to this host/ip address. Use to control which interface\n");
printf(" the client communicates over.\n");
printf(" -c : disable 'clean session' (store subscription and pending messages when client disconnects).\n");
printf(" -C : disconnect and exit after receiving the 'msg_count' messages.\n");
printf(" -d : enable debug messages.\n");
printf(" -D : Define MQTT v5 properties. See the documentation for more details.\n");
printf(" -E : Exit once all subscriptions have been acknowledged by the broker.\n");
printf(" -F : output format.\n");
printf(" -h : mqtt host to connect to. Defaults to localhost.\n");
printf(" -i : id to use for this client. Defaults to mosquitto_sub_ appended with the process id.\n");
printf(" -I : define the client id as id_prefix appended with the process id. Useful for when the\n");
printf(" broker is using the clientid_prefixes option.\n");
printf(" -k : keep alive in seconds for this client. Defaults to 60.\n");
printf(" -L : specify user, password, hostname, port and topic as a URL in the form:\n");
printf(" mqtt(s):https://[username[:password]@]host[:port]/topic\n");
printf(" -N : do not add an end of line character when printing the payload.\n");
printf(" -p : network port to connect to. Defaults to 1883 for plain MQTT and 8883 for MQTT over TLS.\n");
printf(" -P : provide a password\n");
printf(" -q : quality of service level to use for the subscription. Defaults to 0.\n");
printf(" -R : do not print stale messages (those with retain set).\n");
#ifdef WITH_SRV
printf(" -S : use SRV lookups to determine which host to connect to.\n");
#endif
printf(" -t : mqtt topic to subscribe to. May be repeated multiple times.\n");
printf(" -T : topic string to filter out of results. May be repeated.\n");
printf(" -u : provide a username\n");
printf(" -U : unsubscribe from a topic. May be repeated.\n");
printf(" -v : print published messages verbosely.\n");
printf(" -V : specify the version of the MQTT protocol to use when connecting.\n");
printf(" Can be mqttv5, mqttv311 or mqttv31. Defaults to mqttv311.\n");
#ifndef WIN32
printf(" -W : Specifies a timeout in seconds how long to process incoming MQTT messages.\n");
#endif
printf(" --help : display this message.\n");
printf(" --quiet : don't print error messages.\n");
printf(" --retained-only : only handle messages with the retained flag set, and exit when the\n");
printf(" first non-retained message is received.\n");
printf(" --remove-retained : send a message to the server to clear any received retained messages\n");
printf(" Use -T to filter out messages you do not want to be cleared.\n");
printf(" --will-payload : payload for the client Will, which is sent by the broker in case of\n");
printf(" unexpected disconnection. If not given and will-topic is set, a zero\n");
printf(" length message will be sent.\n");
printf(" --will-qos : QoS level for the client Will.\n");
printf(" --will-retain : if given, make the client Will retained.\n");
printf(" --will-topic : the topic on which to publish the client Will.\n");
#ifdef WITH_TLS
printf(" --cafile : path to a file containing trusted CA certificates to enable encrypted\n");
printf(" certificate based communication.\n");
printf(" --capath : path to a directory containing trusted CA certificates to enable encrypted\n");
printf(" communication.\n");
printf(" --cert : client certificate for authentication, if required by server.\n");
printf(" --key : client private key for authentication, if required by server.\n");
printf(" --keyform : keyfile type, can be either \"pem\" or \"engine\".\n");
printf(" --ciphers : openssl compatible list of TLS ciphers to support.\n");
printf(" --tls-version : TLS protocol version, can be one of tlsv1.3 tlsv1.2 or tlsv1.1.\n");
printf(" Defaults to tlsv1.2 if available.\n");
printf(" --insecure : do not check that the server certificate hostname matches the remote\n");
printf(" hostname. Using this option means that you cannot be sure that the\n");
printf(" remote host is the server you wish to connect to and so is insecure.\n");
printf(" Do not use this option in a production environment.\n");
printf(" --tls-engine : If set, enables the use of a SSL engine device.\n");
printf(" --tls-engine-kpass-sha1 : SHA1 of the key password to be used with the selected SSL engine.\n");
#ifdef FINAL_WITH_TLS_PSK
printf(" --psk : pre-shared-key in hexadecimal (no leading 0x) to enable TLS-PSK mode.\n");
printf(" --psk-identity : client identity string for TLS-PSK mode.\n");
#endif
#endif
#ifdef WITH_SOCKS
printf(" --proxy : SOCKS5 proxy URL of the form:\n");
printf(" socks5h:https://[username[:password]@]hostname[:port]\n");
printf(" Only \"none\" and \"username\" authentication is supported.\n");
#endif
printf("\nSee https://mosquitto.org/ for more information.\n\n");
}
int main(int argc, char *argv[])
{
int rc;
#ifndef WIN32
struct sigaction sigact;
#endif
memset(&cfg, 0, sizeof(struct mosq_config));
mosquitto_lib_init();
rc = client_config_load(&cfg, CLIENT_SUB, argc, argv);
if(rc){
if(rc == 2){
/* --help */
print_usage();
}else{
fprintf(stderr, "\nUse 'mosquitto_sub --help' to see usage.\n");
}
goto cleanup;
}
if(cfg.no_retain && cfg.retained_only){
fprintf(stderr, "\nError: Combining '-R' and '--retained-only' makes no sense.\n");
goto cleanup;
}
if(client_id_generate(&cfg)){
goto cleanup;
}
mosq = mosquitto_new(cfg.id, cfg.clean_session, &cfg);
if(!mosq){
switch(errno){
case ENOMEM:
err_printf(&cfg, "Error: Out of memory.\n");
break;
case EINVAL:
err_printf(&cfg, "Error: Invalid id and/or clean_session.\n");
break;
}
goto cleanup;
}
if(client_opts_set(mosq, &cfg)){
goto cleanup;
}
if(cfg.debug){
mosquitto_log_callback_set(mosq, my_log_callback);
mosquitto_subscribe_callback_set(mosq, my_subscribe_callback);
}
mosquitto_connect_v5_callback_set(mosq, my_connect_callback);
mosquitto_message_v5_callback_set(mosq, my_message_callback);
rc = client_connect(mosq, &cfg);
if(rc){
goto cleanup;
}
#ifndef WIN32
sigact.sa_handler = my_signal_handler;
sigemptyset(&sigact.sa_mask);
sigact.sa_flags = 0;
if(sigaction(SIGALRM, &sigact, NULL) == -1){
perror("sigaction");
goto cleanup;
}
if(cfg.timeout){
alarm(cfg.timeout);
}
#endif
rc = mosquitto_loop_forever(mosq, -1, 1);
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
if(cfg.msg_count>0 && rc == MOSQ_ERR_NO_CONN){
rc = 0;
}
client_config_cleanup(&cfg);
if(rc){
fprintf(stderr, "Error: %s\n", mosquitto_strerror(rc));
}
return rc;
cleanup:
mosquitto_lib_cleanup();
client_config_cleanup(&cfg);
return 1;
}