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
|
// Standard C
#include <stdio.h>
// System
#include <unistd.h>
// Project
#include "config.h"
#include "mqtt.h"
struct mosquitto *
mqtt_connect(const char *clientid, // NOLINT
const char *username, // NOLINT
const char *password, // NOLINT
const char *hostname, // NOLINT
int port,
void *userdata)
{
struct mosquitto *client = mosquitto_new(clientid, true, userdata);
if ( client == NULL ) {
(void) fprintf(stderr, "error on new client...\n");
return NULL;
}
int err = mosquitto_username_pw_set(client, username, password);
if ( err != MOSQ_ERR_SUCCESS ) {
(void) fprintf(stderr, "error: %s\n", mosquitto_strerror(err));
mosquitto_destroy(client);
return NULL;
}
err = mosquitto_connect(client, hostname, port, KEEPALIVE);
if ( err != MOSQ_ERR_SUCCESS ) {
(void) fprintf(stderr, "error: %s\n", mosquitto_strerror(err));
mosquitto_destroy(client);
return NULL;
}
return client;
}
void
mqtt_handle_messages(struct mosquitto *client,
const char *topics[],
void callback(struct mosquitto *, void *, const struct mosquitto_message *),
const volatile bool *quit)
{
if ( callback ) {
mosquitto_message_callback_set(client, callback);
}
int *mid = NULL;
for ( size_t idx = 0; topics[idx] != NULL; ++idx ) {
mosquitto_subscribe(client, mid, topics[idx], 0);
}
while ( !*quit ) {
static const int defaultTimeout = -1; // 1000ms
int err = mosquitto_loop(client, defaultTimeout, 1);
if ( !*quit && err ) {
(void) fprintf(stderr, "connection error: %s\n", mosquitto_strerror(err));
sleep(WAIT_UNTIL_RECONNECT); // NOLINT
mosquitto_reconnect(client);
}
}
for ( size_t idx = 0; topics[idx] != NULL; ++idx ) {
mosquitto_unsubscribe(client, mid, topics[idx]);
}
}
void
mqtt_shutdown(struct mosquitto *client)
{
mosquitto_disconnect(client);
mosquitto_destroy(client);
}
|
