initial MQTT over Secure Websockets support for ACLK (#7988)
* add aclk_lws_wss_client * shorten the thread name in case more threads are necessary * Draft libmosquitto<->libwebsockets integration * use ringbuffer for recvd data * Some code cleanup * if mqtt connection fails close lws connection and reconect * clear buffers on connection closed * work on better loop integration * move mosquitto read out of loop * remove useless code when using websockets * LWS - make host and port configurable * make default port 9002 as we use MQTT over WSS now * wait for link up before subscribing start query thread after connection has been made * cleanup - remove useless var * if there is anything to write send it immediatelly * cleanup: move buffers into engine instace * allow MQTT IO from multiple threads (although preffered is MQTT IO to be done by single thread) * add warning to future self * add some comments for whoever reviews * add destroy fnc - start work on cleanup * minor - add mosquitto to .gitignore * fix codacy errors * do not reconnect automatically by default * minor - remove outdated comment * tab -> spaces Co-Authored-By: Konstantinos Natsakis <5933427+knatsakis@users.noreply.github.com> * address thiagoftsm valid comments * add usefull logs in case of trouble * fix -Wall -Wextra -Wformat-signedness warnings * log error when connection fails * update .gitignore to match new installer * Fwd LWS logs to Netdata logs * minor - tabulation fixes * fix comments from thiago * force SSL * move UNUSED to libnetdata.h @thiago correctly pointed out it might be usefull for others * minor - rename function for clarity * minor - remove commented out code Co-authored-by: Konstantinos Natsakis <5933427+knatsakis@users.noreply.github.com>
Timotej Šiškovič committed
Feb 14, 2020 at 10:54 UTC
f4e1012f5ffe1231c25e22f7350d2928b443c69f
9 files changed
+569
-12
Makefile.am
+2
@@ -463,6 +463,8 @@ ACLK_PLUGIN_FILES = \
463
aclk/agent_cloud_link.h \
464
aclk/mqtt.c \
465
aclk/mqtt.h \
466
+ aclk/aclk_lws_wss_client.c \
467
+ aclk/aclk_lws_wss_client.h \
468
$(NULL)
469
470
EXPORTING_ENGINE_FILES = \
aclk/aclk_lws_wss_client.c
new
+337
@@ -0,0 +1,337 @@
1
+#include "aclk_lws_wss_client.h"
2
+
3
+#include "libnetdata/libnetdata.h"
4
+
5
+static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len);
6
+
7
+struct aclk_lws_wss_perconnect_data {
8
+ int todo;
9
+};
10
+
11
+struct lws_wss_packet_buffer {
12
+ unsigned char* data;
13
+ size_t data_size;
14
+ struct lws_wss_packet_buffer *next;
15
+};
16
+
17
+static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_new(void* data, size_t size)
18
+{
19
+ struct lws_wss_packet_buffer *new = callocz(1, sizeof(struct lws_wss_packet_buffer));
20
+ if(data) {
21
+ new->data = mallocz(LWS_PRE+size);
22
+ memcpy(new->data+LWS_PRE, data, size);
23
+ new->data_size = size;
24
+ }
25
+ return new;
26
+}
27
+
28
+static inline void lws_wss_packet_buffer_append(struct lws_wss_packet_buffer **list, struct lws_wss_packet_buffer *item)
29
+{
30
+ struct lws_wss_packet_buffer *tail = *list;
31
+ if(!*list) {
32
+ *list = item;
33
+ return;
34
+ }
35
+ while(tail->next) {
36
+ tail = tail->next;
37
+ }
38
+ tail->next = item;
39
+}
40
+
41
+static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_pop(struct lws_wss_packet_buffer **list)
42
+{
43
+ struct lws_wss_packet_buffer *ret = *list;
44
+ if(ret != NULL)
45
+ *list = ret->next;
46
+
47
+ return ret;
48
+}
49
+
50
+static inline void lws_wss_packet_buffer_free(struct lws_wss_packet_buffer *item)
51
+{
52
+ freez(item->data);
53
+ freez(item);
54
+}
55
+
56
+static inline void _aclk_lws_wss_read_buffer_clear(struct lws_ring *ringbuffer)
57
+{
58
+ size_t elems = lws_ring_get_count_waiting_elements(ringbuffer, NULL);
59
+ if(elems > 0)
60
+ lws_ring_consume(ringbuffer, NULL, NULL, elems);
61
+}
62
+
63
+static inline void _aclk_lws_wss_write_buffer_clear(struct lws_wss_packet_buffer **list)
64
+{
65
+ struct lws_wss_packet_buffer *i;
66
+ while((i = lws_wss_packet_buffer_pop(list)) != NULL) {
67
+ lws_wss_packet_buffer_free(i);
68
+ }
69
+ *list = NULL;
70
+}
71
+
72
+static inline void aclk_lws_wss_clear_io_buffers(struct aclk_lws_wss_engine_instance *inst)
73
+{
74
+ aclk_lws_mutex_lock(&inst->read_buf_mutex);
75
+ _aclk_lws_wss_read_buffer_clear(inst->read_ringbuffer);
76
+ aclk_lws_mutex_unlock(&inst->read_buf_mutex);
77
+ aclk_lws_mutex_lock(&inst->write_buf_mutex);
78
+ _aclk_lws_wss_write_buffer_clear(&inst->write_buffer_head);
79
+ aclk_lws_mutex_unlock(&inst->write_buf_mutex);
80
+}
81
+
82
+static const struct lws_protocols protocols[] = {
83
+ {
84
+ "aclk-wss",
85
+ aclk_lws_wss_callback,
86
+ sizeof(struct aclk_lws_wss_perconnect_data),
87
+ 0, 0, 0, 0
88
+ },
89
+ { NULL, NULL, 0, 0, 0, 0, 0 }
90
+};
91
+
92
+static void aclk_lws_wss_log_divert(int level, const char *line) {
93
+ switch(level){
94
+ case LLL_ERR:
95
+ error("Libwebsockets Error: %s", line);
96
+ break;
97
+ case LLL_WARN:
98
+ debug(D_ACLK, "Libwebsockets Warn: %s", line);
99
+ break;
100
+ default:
101
+ error("Libwebsockets try to log with unknown log level (%d), msg: %s", level, line);
102
+ }
103
+}
104
+
105
+struct aclk_lws_wss_engine_instance* aclk_lws_wss_client_init (const struct aclk_lws_wss_engine_callbacks *callbacks, const char *target_hostname, int target_port) {
106
+ static int lws_logging_initialized = 0;
107
+ struct lws_context_creation_info info;
108
+ struct aclk_lws_wss_engine_instance *inst;
109
+
110
+ if(unlikely(!lws_logging_initialized)) {
111
+ lws_set_log_level(LLL_ERR | LLL_WARN, aclk_lws_wss_log_divert);
112
+ lws_logging_initialized = 1;
113
+ }
114
+
115
+ if(!callbacks || !target_hostname)
116
+ return NULL;
117
+
118
+ inst = callocz(1, sizeof(struct aclk_lws_wss_engine_instance));
119
+
120
+ inst->host = target_hostname;
121
+ inst->port = target_port;
122
+
123
+ memset(&info, 0, sizeof(struct lws_context_creation_info));
124
+ info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
125
+ info.port = CONTEXT_PORT_NO_LISTEN;
126
+ info.protocols = protocols;
127
+ info.user = inst;
128
+
129
+ inst->lws_context = lws_create_context(&info);
130
+ if(!inst->lws_context)
131
+ goto failure_cleanup_2;
132
+
133
+ inst->callbacks = *callbacks;
134
+
135
+ aclk_lws_mutex_init(&inst->write_buf_mutex);
136
+ aclk_lws_mutex_init(&inst->read_buf_mutex);
137
+
138
+ inst->read_ringbuffer = lws_ring_create(1, ACLK_LWS_WSS_RECV_BUFF_SIZE_BYTES, NULL);
139
+ if(!inst->read_ringbuffer)
140
+ goto failure_cleanup;
141
+
142
+ return inst;
143
+
144
+failure_cleanup:
145
+ lws_context_destroy(inst->lws_context);
146
+failure_cleanup_2:
147
+ freez(inst);
148
+ return NULL;
149
+}
150
+
151
+void aclk_lws_wss_client_destroy(struct aclk_lws_wss_engine_instance* inst) {
152
+ lws_context_destroy(inst->lws_context);
153
+ inst->lws_context = NULL;
154
+ inst->lws_wsi = NULL;
155
+
156
+ aclk_lws_wss_clear_io_buffers(inst);
157
+
158
+#ifdef ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED
159
+ pthread_mutex_destroy(&inst->write_buf_mutex);
160
+ pthread_mutex_destroy(&inst->read_buf_mutex);
161
+#endif
162
+}
163
+
164
+void _aclk_wss_connect(struct aclk_lws_wss_engine_instance *inst){
165
+ struct lws_client_connect_info i;
166
+
167
+ memset(&i, 0, sizeof(i));
168
+ i.context = inst->lws_context;
169
+ i.port = inst->port;
170
+ i.address = inst->host;
171
+ i.path = "/mqtt";
172
+ i.host = inst->host;
173
+ i.protocol = "mqtt";
174
+#ifdef ACLK_SSL_ALLOW_SELF_SIGNED
175
+ i.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK;
176
+#else
177
+ i.ssl_connection = LCCSCF_USE_SSL;
178
+#endif
179
+ lws_client_connect_via_info(&i);
180
+}
181
+
182
+static inline int received_data_to_ringbuff(struct lws_ring *buffer, void* data, size_t len) {
183
+ if( lws_ring_insert(buffer, data, len) != len ) {
184
+ error("ACLK_LWS_WSS_CLIENT: receive buffer full. Closing connection to prevent flooding.");
185
+ return 0;
186
+ }
187
+ return 1;
188
+}
189
+
190
+static int
191
+aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason,
192
+ void *user, void *in, size_t len)
193
+{
194
+ UNUSED(user);
195
+ struct aclk_lws_wss_engine_instance *inst = lws_context_user(lws_get_context(wsi));
196
+ struct lws_wss_packet_buffer *data;
197
+ int retval = 0;
198
+
199
+ if( !inst ) {
200
+ error("Callback received without any aclk_lws_wss_engine_instance!");
201
+ return -1;
202
+ }
203
+
204
+ if( inst->upstream_reconnect_request ) {
205
+ error("Closing lws connectino due to libmosquitto error.");
206
+ char *upstream_connection_error = "MQTT protocol error. Closing underlying wss connection.";
207
+ lws_close_reason(wsi, LWS_CLOSE_STATUS_PROTOCOL_ERR, (unsigned char*)upstream_connection_error, strlen(upstream_connection_error));
208
+ retval = -1;
209
+ inst->upstream_reconnect_request = 0;
210
+ }
211
+
212
+ switch (reason) {
213
+ case LWS_CALLBACK_CLIENT_WRITEABLE:
214
+ aclk_lws_mutex_lock(&inst->write_buf_mutex);
215
+ data = lws_wss_packet_buffer_pop(&inst->write_buffer_head);
216
+ if(likely(data)) {
217
+ lws_write(wsi, data->data + LWS_PRE, data->data_size, LWS_WRITE_BINARY);
218
+ lws_wss_packet_buffer_free(data);
219
+ if(inst->write_buffer_head)
220
+ lws_callback_on_writable(inst->lws_wsi);
221
+ }
222
+ aclk_lws_mutex_unlock(&inst->write_buf_mutex);
223
+ break;
224
+ case LWS_CALLBACK_CLIENT_RECEIVE:
225
+ aclk_lws_mutex_lock(&inst->read_buf_mutex);
226
+ if(!received_data_to_ringbuff(inst->read_ringbuffer, in, len))
227
+ retval = 1;
228
+ aclk_lws_mutex_unlock(&inst->read_buf_mutex);
229
+
230
+ if(likely(inst->callbacks.data_rcvd_callback))
231
+ // to future myself -> do not call this while read lock is active as it will eventually
232
+ // want to acquire same lock later in aclk_lws_wss_client_read() function
233
+ inst->callbacks.data_rcvd_callback();
234
+ else
235
+ inst->data_to_read = 1; //to inform logic above there is reason to call mosquitto_loop_read
236
+ break;
237
+ case LWS_CALLBACK_PROTOCOL_INIT:
238
+ //initial connection here
239
+ //later we will reconnect with delay od ACLK_LWS_WSS_RECONNECT_TIMEOUT
240
+ //in case this connection fails or drops
241
+ _aclk_wss_connect(inst);
242
+ break;
243
+ case LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED:
244
+ //TODO if already active make some error noise
245
+ //currently we expect only one connection per netdata
246
+ inst->lws_wsi = wsi;
247
+ break;
248
+#ifdef AUTO_RECONNECT_ON_LWS_LAYER
249
+ case LWS_CALLBACK_USER:
250
+ inst->reconnect_timeout_running = 0;
251
+ _aclk_wss_connect(inst);
252
+ break;
253
+#endif
254
+ case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
255
+ error("Could not connect MQTT over WSS server \"%s:%d\". LwsReason:\"%s\"", inst->host, inst->port, (in ? (char*)in : "not given"));
256
+ /* FALLTHRU */
257
+ case LWS_CALLBACK_CLIENT_CLOSED:
258
+ case LWS_CALLBACK_WS_PEER_INITIATED_CLOSE:
259
+#ifdef AUTO_RECONNECT_ON_LWS_LAYER
260
+ if(!inst->reconnect_timeout_running) {
261
+ lws_timed_callback_vh_protocol(lws_get_vhost(wsi),
262
+ lws_get_protocol(wsi),
263
+ LWS_CALLBACK_USER, ACLK_LWS_WSS_RECONNECT_TIMEOUT);
264
+ inst->reconnect_timeout_running = 1;
265
+ }
266
+ /* FALLTHRU */
267
+#endif
268
+ //no break here on purpose we want to continue with LWS_CALLBACK_WSI_DESTROY
269
+ case LWS_CALLBACK_WSI_DESTROY:
270
+ aclk_lws_wss_clear_io_buffers(inst);
271
+ inst->lws_wsi = NULL;
272
+ inst->websocket_connection_up = 0;
273
+ break;
274
+ case LWS_CALLBACK_CLIENT_ESTABLISHED:
275
+ inst->websocket_connection_up = 1;
276
+ if(inst->callbacks.connection_established_callback)
277
+ inst->callbacks.connection_established_callback();
278
+ break;
279
+ default:
280
+ break;
281
+ }
282
+ return retval; //0-OK, other connection should be closed!
283
+}
284
+
285
+int aclk_lws_wss_client_write(struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count)
286
+{
287
+ if(inst && inst->lws_wsi && inst->websocket_connection_up)
288
+ {
289
+ aclk_lws_mutex_lock(&inst->write_buf_mutex);
290
+ lws_wss_packet_buffer_append(&inst->write_buffer_head, lws_wss_packet_buffer_new(buf, count));
291
+ aclk_lws_mutex_unlock(&inst->write_buf_mutex);
292
+
293
+ lws_callback_on_writable(inst->lws_wsi);
294
+ return count;
295
+ }
296
+ return 0;
297
+}
298
+
299
+int aclk_lws_wss_client_read(struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count)
300
+{
301
+ size_t data_to_be_read = count;
302
+
303
+ aclk_lws_mutex_lock(&inst->read_buf_mutex);
304
+ size_t readable_byte_count = lws_ring_get_count_waiting_elements(inst->read_ringbuffer, NULL);
305
+ if(unlikely(readable_byte_count == 0)) {
306
+ errno = EAGAIN;
307
+ data_to_be_read = -1;
308
+ goto abort;
309
+ }
310
+
311
+ if( readable_byte_count < data_to_be_read )
312
+ data_to_be_read = readable_byte_count;
313
+
314
+ data_to_be_read = lws_ring_consume(inst->read_ringbuffer, NULL, buf, data_to_be_read);
315
+ if(data_to_be_read == readable_byte_count)
316
+ inst->data_to_read = 0;
317
+
318
+abort:
319
+ aclk_lws_mutex_unlock(&inst->read_buf_mutex);
320
+ return data_to_be_read;
321
+}
322
+
323
+int aclk_lws_wss_service_loop(struct aclk_lws_wss_engine_instance *inst)
324
+{
325
+ return lws_service(inst->lws_context, 0);
326
+}
327
+
328
+// in case the MQTT connection disconnect while lws transport is still operational
329
+// we should drop connection and reconnect
330
+// this function should be called when that happens to notify lws of that situation
331
+void aclk_lws_wss_mqtt_layer_disconect_notif(struct aclk_lws_wss_engine_instance *inst)
332
+{
333
+ if(inst->lws_wsi && inst->websocket_connection_up) {
334
+ inst->upstream_reconnect_request = 1;
335
+ lws_callback_on_writable(inst->lws_wsi); //here we just do it to ensure we get callback called from lws, we don't need any actual data to be written.
336
+ }
337
+}
\ No newline at end of file
aclk/aclk_lws_wss_client.h
new
+81
@@ -0,0 +1,81 @@
1
+#ifndef ACLK_LWS_WSS_CLIENT_H
2
+#define ACLK_LWS_WSS_CLIENT_H
3
+
4
+#include <libwebsockets.h>
5
+
6
+#include "libnetdata/libnetdata.h"
7
+
8
+#define ACLK_LWS_WSS_RECONNECT_TIMEOUT 5
9
+
10
+// This is as define because ideally the ACLK at high level
11
+// can do mosqitto writes and reads only from one thread
12
+// which is cleaner implementation IMHO
13
+// in such case this mutexes are not necessarry and life
14
+// is simpler
15
+#define ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED 1
16
+
17
+#define ACLK_LWS_WSS_RECV_BUFF_SIZE_BYTES 128*1024
18
+
19
+#ifdef ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED
20
+ #define aclk_lws_mutex_init(x) netdata_mutex_init(x)
21
+ #define aclk_lws_mutex_lock(x) netdata_mutex_lock(x)
22
+ #define aclk_lws_mutex_unlock(x) netdata_mutex_unlock(x)
23
+#else
24
+ #define aclk_lws_mutex_init(x)
25
+ #define aclk_lws_mutex_lock(x)
26
+ #define aclk_lws_mutex_unlock(x)
27
+#endif
28
+
29
+struct aclk_lws_wss_engine_callbacks {
30
+ void (*connection_established_callback)();
31
+ void (*data_rcvd_callback)();
32
+ void (*data_writable_callback)();
33
+};
34
+
35
+struct lws_wss_packet_buffer;
36
+
37
+struct aclk_lws_wss_engine_instance {
38
+ //target host/port for connection
39
+ const char *host;
40
+ int port;
41
+
42
+ //internal data
43
+ struct lws_context *lws_context;
44
+ struct lws *lws_wsi;
45
+
46
+#ifdef ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED
47
+ netdata_mutex_t write_buf_mutex;
48
+ netdata_mutex_t read_buf_mutex;
49
+#endif
50
+
51
+ struct lws_wss_packet_buffer *write_buffer_head;
52
+ struct lws_ring *read_ringbuffer;
53
+
54
+ struct aclk_lws_wss_engine_callbacks callbacks;
55
+
56
+ //flags to be readed by engine user
57
+ int websocket_connection_up;
58
+
59
+// currently this is by default disabled
60
+// as decision has been made that reconnection
61
+// will have to be done from top layer
62
+// (after getting the new MQTT auth data)
63
+// for now i keep it here as it is usefull for
64
+// some of my internall testing
65
+#ifdef AUTO_RECONNECT_ON_LWS_LAYER
66
+ int reconnect_timeout_running;
67
+#endif
68
+ int data_to_read;
69
+ int upstream_reconnect_request;
70
+};
71
+
72
+struct aclk_lws_wss_engine_instance* aclk_lws_wss_client_init (const struct aclk_lws_wss_engine_callbacks *callbacks, const char *target_hostname, int target_port);
73
+void aclk_lws_wss_client_destroy(struct aclk_lws_wss_engine_instance* inst);
74
+
75
+int aclk_lws_wss_client_write(struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count);
76
+int aclk_lws_wss_client_read (struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count);
77
+int aclk_lws_wss_service_loop(struct aclk_lws_wss_engine_instance *inst);
78
+
79
+void aclk_lws_wss_mqtt_layer_disconect_notif(struct aclk_lws_wss_engine_instance *inst);
80
+
81
+#endif
\ No newline at end of file
aclk/agent_cloud_link.c
+7
-3
@@ -94,6 +94,10 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
94
95
// Set when we have connection up and running from the connection callback
96
int aclk_connection_initialized = 0;
97
+// TODO modify previous comment if this stays this way
98
+// con_initialized means library is initialized and ready to be used
99
+// acklk_connected means there is actually an established connection
100
+int aclk_mqtt_connected = 0;
101
102
static netdata_mutex_t aclk_mutex = NETDATA_MUTEX_INITIALIZER;
103
static netdata_mutex_t query_mutex = NETDATA_MUTEX_INITIALIZER;
@@ -625,10 +629,10 @@ void *aclk_main(void *ptr)
629
continue;
630
}
631
628
- if (unlikely(!aclk_subscribed)) {
632
+ if (unlikely(!aclk_subscribed) && aclk_mqtt_connected) {
633
aclk_subscribed = !aclk_subscribe(ACLK_COMMAND_TOPIC, 2);
634
}
631
- if (unlikely(!query_thread.thread)) {
635
+ if (unlikely(!query_thread.thread && aclk_mqtt_connected)) {
636
query_thread.thread = mallocz(sizeof(netdata_thread_t));
637
netdata_thread_create(
638
query_thread.thread, "ACLKQ", NETDATA_THREAD_OPTION_DEFAULT, aclk_query_main_thread, &query_thread);
@@ -763,7 +767,7 @@ int aclk_init(ACLK_INIT_ACTION action)
767
aclk_recv_maximum = config_get_number(CONFIG_SECTION_ACLK, "agent cloud link receive maximum", 20);
768
769
aclk_hostname = config_get(CONFIG_SECTION_ACLK, "agent cloud link hostname", "localhost");
766
- aclk_port = config_get_number(CONFIG_SECTION_ACLK, "agent cloud link port", 1883);
770
+ aclk_port = config_get_number(CONFIG_SECTION_ACLK, "agent cloud link port", 9002);
771
772
info("Maximum parallel outgoing messages %d", aclk_send_maximum);
773
info("Maximum parallel incoming messages %d", aclk_recv_maximum);
aclk/agent_cloud_link.h
+1
-1
@@ -63,7 +63,7 @@ void *aclk_main(void *ptr);
63
64
#define NETDATA_ACLK_HOOK \
65
{ \
66
- .name = "AgentCloudLink", \
66
+ .name = "ACLK_Main", \
67
.config_section = NULL, \
68
.config_name = NULL, \
69
.enabled = 1, \
aclk/mqtt.c
+135
-7
@@ -3,6 +3,7 @@
3
#include <libnetdata/json/json.h>
4
#include "../daemon/common.h"
5
#include "mqtt.h"
6
+#include "aclk_lws_wss_client.h"
7
8
void (*_on_connect)(void *ptr) = NULL;
9
void (*_on_disconnect)(void *ptr) = NULL;
@@ -104,6 +105,14 @@ void mqtt_message_callback(
105
106
}
107
108
+int lws_wss_client_initialized = 0;
109
+
110
+// This is not define because in future we might want to try plain
111
+// MQTT as fallback ?
112
+// e.g. try 1st MQTT-WSS, 2nd MQTT plain, 3rd https fallback...
113
+int mqtt_over_websockets = 1;
114
+struct aclk_lws_wss_engine_instance *lws_engine_instance = NULL;
115
+
116
void connect_callback(struct mosquitto *mosq, void *obj, int rc)
117
{
118
(void) obj;
@@ -112,6 +121,7 @@ void connect_callback(struct mosquitto *mosq, void *obj, int rc)
121
info("Connection to cloud estabilished");
122
123
aclk_connection_initialized = 1;
124
+ aclk_mqtt_connected = 1;
125
_on_connect((void *) mosq);
126
127
return;
@@ -127,7 +137,12 @@ void disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
137
// TODO: Keep the connection "alive" for now. The library will reconnect.
138
139
//mqtt_connection_initialized = 0;
140
+ aclk_mqtt_connected = 0;
141
_on_disconnect((void *) mosq);
142
+
143
+ if(mqtt_over_websockets && lws_engine_instance)
144
+ aclk_lws_wss_mqtt_layer_disconect_notif(lws_engine_instance);
145
+
146
//sleep_usec(USEC_PER_SEC * 5);
147
return;
148
}
@@ -141,7 +156,17 @@ void _show_mqtt_info()
156
info("Detected libmosquitto library version %d, %d.%d.%d",libmosq_version, libmosq_major, libmosq_minor, libmosq_revision);
157
}
158
144
-int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
159
+size_t _mqtt_external_write_hook(void *buf, size_t count)
160
+{
161
+ return aclk_lws_wss_client_write(lws_engine_instance, buf, count);
162
+}
163
+
164
+size_t _mqtt_external_read_hook(void *buf, size_t count)
165
+{
166
+ return aclk_lws_wss_client_read(lws_engine_instance, buf, count);
167
+}
168
+
169
+int _mqtt_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
170
{
171
int rc;
172
int libmosq_major, libmosq_minor, libmosq_revision, libmosq_version;
@@ -194,7 +219,7 @@ int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
219
mosquitto_connect_callback_set(mosq, connect_callback);
220
mosquitto_disconnect_callback_set(mosq, disconnect_callback);
221
197
- mosquitto_username_pw_set(mosq, "anon", "anon");
222
+ mosquitto_username_pw_set(mosq, NULL, NULL);
223
224
rc = mosquitto_threaded_set(mosq, 1);
225
if (unlikely(rc != MOSQ_ERR_SUCCESS))
@@ -209,12 +234,21 @@ int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
234
info("MQTT in flight messages set to 1 -- %s", mosquitto_strerror(rc));
235
#endif
236
212
- rc = mosquitto_reconnect_delay_set(mosq, ACLK_RECONNECT_DELAY, ACLK_MAX_RECONNECT_DELAY, 1);
237
+ if(!mqtt_over_websockets) {
238
+ rc = mosquitto_reconnect_delay_set(mosq, ACLK_RECONNECT_DELAY, ACLK_MAX_RECONNECT_DELAY, 1);
239
214
- if (unlikely(rc != MOSQ_ERR_SUCCESS))
215
- error("Failed to set the reconnect delay (%d) (%s)", rc, mosquitto_strerror(rc));
240
+ if (unlikely(rc != MOSQ_ERR_SUCCESS))
241
+ error("Failed to set the reconnect delay (%d) (%s)", rc, mosquitto_strerror(rc));
242
+
243
+ mosquitto_tls_set(mosq, ca_crt, NULL, server_crt, server_key, NULL);
244
+ }
245
+
246
+ return rc;
247
+}
248
217
- mosquitto_tls_set(mosq, ca_crt, NULL, server_crt, server_key, NULL);
249
+int _link_mqtt_connect(char *aclk_hostname, int aclk_port)
250
+{
251
+ int rc;
252
253
rc = mosquitto_connect_async(mosq, aclk_hostname, aclk_port, ACLK_PING_INTERVAL);
254
@@ -226,7 +260,83 @@ int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
260
return rc;
261
}
262
229
-int _link_event_loop(int timeout)
263
+static inline void _link_mosquitto_write()
264
+{
265
+ int rc;
266
+
267
+ if(!mqtt_over_websockets)
268
+ return;
269
+
270
+ rc = mosquitto_loop_misc(mosq);
271
+ if(unlikely( rc != MOSQ_ERR_SUCCESS ))
272
+ debug(D_ACLK, "ACLK: failure during mosquitto_loop_misc %s", mosquitto_strerror(rc));
273
+
274
+ if(likely( mosquitto_want_write(mosq) )) {
275
+ rc = mosquitto_loop_write(mosq, 1);
276
+ if( rc != MOSQ_ERR_SUCCESS )
277
+ debug(D_ACLK, "ACLK: failure during mosquitto_loop_write %s", mosquitto_strerror(rc));
278
+ }
279
+}
280
+
281
+void aclk_lws_connect_notif_callback(){
282
+ //the connection is done by LWS so this parameters dont matter
283
+ //ig MQTT over LWS is used
284
+ _link_mqtt_connect("doesntmatter", 12345);
285
+ _link_mosquitto_write();
286
+}
287
+
288
+void aclk_lws_data_received_callback(){
289
+ int rc = mosquitto_loop_read(mosq, 1);
290
+ if(rc != MOSQ_ERR_SUCCESS)
291
+ debug(D_ACLK, "ACLK: failure during mosquitto_loop_read %s", mosquitto_strerror(rc));
292
+}
293
+
294
+static const struct aclk_lws_wss_engine_callbacks aclk_lws_engine_callbacks = {
295
+ .connection_established_callback = aclk_lws_connect_notif_callback,
296
+ .data_rcvd_callback = aclk_lws_data_received_callback,
297
+ .data_writable_callback = NULL
298
+};
299
+
300
+int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
301
+{
302
+ int rc;
303
+
304
+ if(mqtt_over_websockets) {
305
+ // we will connect when WebSocket connection is up
306
+ // based on callback
307
+ if(!lws_wss_client_initialized) {
308
+ lws_engine_instance = aclk_lws_wss_client_init(&aclk_lws_engine_callbacks, aclk_hostname, aclk_port);
309
+ aclk_lws_wss_service_loop(lws_engine_instance);
310
+ lws_wss_client_initialized = 1;
311
+ }
312
+ }
313
+
314
+ rc = _mqtt_lib_init(aclk_hostname, aclk_port, on_connect, on_disconnect);
315
+ if(rc != MOSQ_ERR_SUCCESS)
316
+ return rc;
317
+
318
+ if(mqtt_over_websockets) {
319
+ mosquitto_external_callbacks_set(mosq, _mqtt_external_write_hook, _mqtt_external_read_hook);
320
+ return MOSQ_ERR_SUCCESS;
321
+ } else {
322
+ // if direct mqtt connection is used
323
+ // connect immediatelly
324
+ return _link_mqtt_connect(aclk_hostname, aclk_port);
325
+ }
326
+}
327
+
328
+static inline int _link_event_loop_wss()
329
+{
330
+ if(lws_engine_instance && lws_engine_instance->websocket_connection_up)
331
+ _link_mosquitto_write();
332
+
333
+ aclk_lws_wss_service_loop(lws_engine_instance);
334
+ // this is because if use LWS we don't want
335
+ // mqtt to reconnect by itself
336
+ return MOSQ_ERR_SUCCESS;
337
+}
338
+
339
+static inline int _link_event_loop_plain_mqtt(int timeout)
340
{
341
int rc;
342
@@ -245,6 +355,14 @@ int _link_event_loop(int timeout)
355
return rc;
356
}
357
358
+int _link_event_loop(int timeout)
359
+{
360
+ if(mqtt_over_websockets)
361
+ return _link_event_loop_wss();
362
+
363
+ return _link_event_loop_plain_mqtt(timeout);
364
+}
365
+
366
void _link_shutdown()
367
{
368
int rc;
@@ -261,6 +379,12 @@ void _link_shutdown()
379
380
mosquitto_destroy(mosq);
381
mosq = NULL;
382
+
383
+ if(lws_engine_instance) {
384
+ aclk_lws_wss_client_destroy(lws_engine_instance);
385
+ lws_engine_instance = NULL;
386
+ }
387
+
388
return;
389
}
390
@@ -281,6 +405,8 @@ int _link_subscribe(char *topic, int qos)
405
return 1;
406
}
407
408
+ _link_mosquitto_write();
409
+
410
return 0;
411
}
412
@@ -313,6 +439,8 @@ int _link_send_message(char *topic, char *message)
439
error("MQTT message failed : %s", mosquitto_strerror(rc));
440
}
441
442
+ _link_mosquitto_write();
443
+
444
return rc;
445
}
446
#endif
\ No newline at end of file
aclk/mqtt.h
+3
-1
@@ -15,7 +15,9 @@ int _link_subscribe(char *topic, int qos);
15
int _link_send_message(char *topic, char *message);
16
const char *_link_strerror(int rc);
17
18
-extern int aclk_connection_initialized;
18
int aclk_handle_cloud_request(char *);
19
20
+extern int aclk_connection_initialized;
21
+extern int aclk_mqtt_connected;
22
+
23
#endif //NETDATA_MQTT_H
libnetdata/libnetdata.h
+2
@@ -288,6 +288,8 @@ extern void recursive_config_double_dir_load(
288
289
#define BITS_IN_A_KILOBIT 1000
290
291
+/* misc. */
292
+#define UNUSED(x) (void)(x)
293
294
extern void netdata_cleanup_and_exit(int ret) NORETURN;
295
extern void send_statistics(const char *action, const char *action_result, const char *action_data);
libnetdata/log/log.h
+1
@@ -37,6 +37,7 @@
37
#define D_POLLFD 0x0000000020000000
38
#define D_STREAM 0x0000000040000000
39
#define D_RRDENGINE 0x0000000100000000
40
+#define D_ACLK 0x0000000200000000
41
#define D_SYSTEM 0x8000000000000000
42
43
//#define DEBUG (D_WEB_CLIENT_ACCESS|D_LISTENER|D_RRD_STATS)