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
|
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "mosquitto.h"
int client_id_generate(char **output)
{
int len;
char hostname[256];
*output = calloc(sizeof(char), MOSQ_MQTT_ID_MAX_LENGTH);
if (!*output) {
fprintf(stderr, "Error: Out of memory. %s\n", strerror(errno));
mosquitto_lib_cleanup();
return 1;
}
memset(hostname, 0, sizeof(hostname));
gethostname(hostname, 255);
/* Clamp length to MQTT maximum client ID length */
len = strlen("mfi|-") + 6 + strlen(hostname);
if (len > MOSQ_MQTT_ID_MAX_LENGTH - 1) {
len = MOSQ_MQTT_ID_MAX_LENGTH - 1;
}
snprintf(*output, len, "mfi|%d-%s", getpid(), hostname);
return MOSQ_ERR_SUCCESS;
}
struct mosquitto * connect_to_broker(char *host, int port) {
int rc;
char *id;
struct mosquitto *mosq;
mosquitto_lib_init();
if (client_id_generate(&id)) {
return NULL;
}
mosq = mosquitto_new(id, true, NULL);
if (!mosq) {
switch (errno) {
case ENOMEM:
fprintf(stderr, "Error: Out of memory.\n");
break;
case EINVAL:
fprintf(stderr, "Error: Invalid id and/or clean_session.\n");
break;
}
mosquitto_lib_cleanup();
return NULL;
}
int protocol_version = MQTT_PROTOCOL_V311;
mosquitto_max_inflight_messages_set(mosq, 20);
mosquitto_opts_set(mosq, MOSQ_OPT_PROTOCOL_VERSION, &protocol_version);
rc = mosquitto_connect(mosq, host, port, 60);
if (rc > 0) {
if (rc == MOSQ_ERR_ERRNO) {
fprintf(stderr, "Error: %s\n", strerror(errno));
} else {
fprintf(stderr, "Unable to connect (%s).\n", mosquitto_strerror(rc));
}
mosquitto_lib_cleanup();
return NULL;
}
return mosq;
}
void shutdown_broker(struct mosquitto *mosq) {
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
}
|
