-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathmosquitto_mosq.c
1658 lines (1449 loc) · 60.5 KB
/
mosquitto_mosq.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
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
* FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
* Copyright (C) 2005-2012, Anthony Minessale II <[email protected]>
*
* Version: MPL 1.1
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
*
* The Initial Developer of the Original Code is
* Anthony Minessale II <[email protected]>
* Portions created by the Initial Developer are Copyright (C)
* the Initial Developer. All Rights Reserved.
*
* Contributor(s):
*
* Norm Brandinger <[email protected]>
*
* mod_mosquitto: Interface to an MQTT broker using Mosquitto
* Implements a Publish/Subscribe (pub/sub) messaging pattern using the Mosquitto API library
* Publishes FreeSWITCH events to one more more MQTT brokers
* Subscribes to topics located on one more more MQTT brokers
*
* MQTT http://mqtt.org/
* Mosquitto https://mosquitto.org/
*
*/
#include <switch.h>
#include "mod_mosquitto.h"
#include "mosquitto_cli.h"
#include "mosquitto_config.h"
#include "mosquitto_events.h"
#include "mosquitto_mosq.h"
#include "mosquitto_utils.h"
static switch_status_t process_originate_message(mosquitto_mosq_userdata_t *userdata, char *payload_string,
const struct mosquitto_message *message);
static switch_status_t process_bgapi_message(mosquitto_mosq_userdata_t *userdata, char *payload_string,
const struct mosquitto_message *message);
/**
* @brief This is a threaded bgapi execution routine.
*
* @details This routine is called 'in a new thread' to process a bgapi command.
*
* @param[in] *thread Pointer to a FreeSWITCH thread structure that this routine is called in
* @param[in] *obj Pointer to the bgapi command and arguments
*/
void *SWITCH_THREAD_FUNC bgapi_exec(switch_thread_t *thread, void *obj)
{
mosquitto_bgapi_job_t *job = NULL;
switch_stream_handle_t stream = {0};
char *reply, *freply = NULL;
switch_event_t *event;
char *arg = NULL;
switch_memory_pool_t *pool;
job = (mosquitto_bgapi_job_t *)obj;
if (!job) { return NULL; }
switch_thread_rwlock_rdlock(mosquitto_globals.bgapi_rwlock);
pool = job->pool;
SWITCH_STANDARD_STREAM(stream);
if ((arg = strchr(job->cmd, ' '))) { *arg++ = '\0'; }
if ((switch_api_execute(job->cmd, arg, NULL, &stream)) == SWITCH_STATUS_SUCCESS) {
reply = (char *)stream.data;
} else {
freply = switch_mprintf("%s: Command not found!\n", job->cmd);
reply = freply;
}
if (!reply) { reply = "Command returned no output!"; }
if (switch_event_create(&event, SWITCH_EVENT_BACKGROUND_JOB) == SWITCH_STATUS_SUCCESS) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Job-UUID", job->uuid_str);
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Job-Command", job->cmd);
if (arg) { switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Job-Command-Arg", arg); }
switch_event_add_body(event, "%s", reply);
switch_event_fire(&event);
}
switch_safe_free(stream.data);
switch_safe_free(freply);
job = NULL;
switch_core_destroy_memory_pool(&pool);
pool = NULL;
switch_thread_rwlock_unlock(mosquitto_globals.bgapi_rwlock);
return NULL;
}
/**
* @brief This routine is called when an 'originate' message is received by a subscription
*
* @details This routine will set up and execute an originate command
*
* @param[in] *userdata Pointer to a userdata structure set up when the connection associated with this subscription
* was performed.
* @param[in] *payload_string Pointer to a local copy of the subscribed (received) message
* @param[in] *message Pointer to the received message in a Mosquitto message data structure
*
* @retval SWITCH_STATUS_SUCCESS
*/
static switch_status_t process_originate_message(mosquitto_mosq_userdata_t *userdata, char *payload_string,
const struct mosquitto_message *message)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
char *argv[3] = {0};
switch_channel_t *channel = NULL;
switch_core_session_t *session = NULL;
switch_call_cause_t cause = SWITCH_CAUSE_NONE;
int timeout = 30;
switch_event_t *originate_vars = NULL;
// char *cid_name = NULL;
// char *cid_number = NULL;
char *aleg = NULL;
char *bleg = NULL;
mosquitto_profile_t *profile = userdata->profile;
mosquitto_connection_t *connection = userdata->connection;
mosquitto_topic_t *topic = NULL;
switch_event_create(&originate_vars, SWITCH_EVENT_REQUEST_PARAMS);
switch_assert(originate_vars);
switch_event_add_header(originate_vars, SWITCH_STACK_BOTTOM, "originate_timeout", "%d", 30);
switch_separate_string(payload_string, ' ', argv, 3);
topic = locate_connection_topic(profile, connection, message->topic);
if (!topic) {
log(SWITCH_LOG_ERROR, "Unknown topic: messsage topic %s within profile %s and connection %s\n", message->topic,
profile->name, connection->name);
return status;
} else {
log(SWITCH_LOG_DEBUG, "Matched topic topic %s\n", topic->name);
}
if (!topic->originate_authorized) {
log(SWITCH_LOG_ERROR, "Topic %s not authorized to originate calls within profile %s and connection %s\n",
message->topic, profile->name, connection->name);
return status;
}
if (zstr(argv[1])) {
log(SWITCH_LOG_ERROR, "Aleg passed in from the originate is empty\n");
return status;
}
if (zstr(argv[2])) {
log(SWITCH_LOG_ERROR, "Bleg passed in from the originate is empty\n");
return status;
}
aleg = argv[1];
bleg = argv[2];
#if SWITCH_API_VERSION < 5
status = switch_ivr_originate(session, NULL, &cause, NULL, timeout, NULL, cid_name, cid_number, NULL,
originate_vars, SOF_NONE, NULL);
#else
status = switch_ivr_originate(NULL, &session, &cause, aleg, timeout, NULL, NULL, NULL, NULL, originate_vars,
SOF_NONE, NULL, NULL);
#endif
if (status != SWITCH_STATUS_SUCCESS || !session) {
log(SWITCH_LOG_WARNING, "Originate to [%s] failed, cause: %s\n", aleg, switch_channel_cause2str(cause));
return status;
}
channel = switch_core_session_get_channel(session);
if (*bleg == '&' && *(bleg + 1)) {
switch_caller_extension_t *extension = NULL;
char *app_name = switch_core_session_strdup(session, (bleg + 1));
char *arg = NULL, *e;
if ((e = strchr(app_name, ')'))) { *e = '\0'; }
if ((arg = strchr(app_name, '('))) { *arg++ = '\0'; }
if ((extension = switch_caller_extension_new(session, app_name, arg)) == 0) {
log(SWITCH_LOG_CRIT, "Memory Error!\n");
abort();
}
switch_caller_extension_add_application(session, extension, app_name, arg);
switch_channel_set_caller_extension(channel, extension);
switch_channel_set_state(channel, CS_EXECUTE);
} else {
char *dp = "XML";
char *context = "default";
switch_ivr_session_transfer(session, bleg, dp, context);
}
return status;
}
/**
* @brief This routine is called when an 'bgapi' message is received by a subscription
*
* @details This routine will set up and execute a bgapi command
*
* @param[in] *userdata Pointer to a userdata structure set up when the connection associated with this subscription
* was performed.
* @param[in] *payload_string Pointer to a local copy of the subscribed (received) message
* @param[in] *message Pointer to the received message in a Mosquitto message data structure
*
* @retval SWITCH_STATUS_SUCCESS
*/
static switch_status_t process_bgapi_message(mosquitto_mosq_userdata_t *userdata, char *payload_string,
const struct mosquitto_message *message)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
mosquitto_bgapi_job_t *job = NULL;
switch_uuid_t uuid;
switch_memory_pool_t *pool;
switch_thread_t *thread;
switch_threadattr_t *thd_attr = NULL;
const char *arg = payload_string;
char my_uuid[SWITCH_UUID_FORMATTED_LENGTH + 1] = "";
if (!strncasecmp(payload_string, "uuid:", 5)) {
const char *p = payload_string + 5;
if ((arg = strchr(p, ' ')) && *arg++) { switch_copy_string(my_uuid, p, arg - p); }
}
if (zstr(arg)) {
log(SWITCH_LOG_ERROR, "-ERR Invalid syntax arg empty\n");
return status;
}
switch_core_new_memory_pool(&pool);
job = (mosquitto_bgapi_job_t *)switch_core_alloc(pool, sizeof(*job));
job->cmd = switch_core_strdup(pool, arg);
job->pool = pool;
if (*my_uuid) {
switch_copy_string(job->uuid_str, my_uuid, strlen(my_uuid) + 1);
} else {
switch_uuid_get(&uuid);
switch_uuid_format(job->uuid_str, &uuid);
}
switch_threadattr_create(&thd_attr, job->pool);
switch_threadattr_detach_set(thd_attr, 1);
switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
log(SWITCH_LOG_INFO, "+OK Job-UUID: %s\n", job->uuid_str);
switch_thread_create(&thread, thd_attr, bgapi_exec, job, job->pool);
return status;
}
/**
* @brief This routine sets up callbacks for associated with a connection to an MQTT broker
*
* @details Set the logging callback. This should be used if you want event logging information from the client
* library.
*
* @param[in] *connection Pointer to a connection hash that these callbacks will be associated with
*/
void mosq_callbacks_set(mosquitto_connection_t *connection)
{
mosquitto_log_callback_set(connection->mosq, mosq_log_callback);
mosquitto_connect_callback_set(connection->mosq, mosq_connect_callback);
mosquitto_message_callback_set(connection->mosq, mosq_message_callback);
mosquitto_subscribe_callback_set(connection->mosq, mosq_subscribe_callback);
mosquitto_publish_callback_set(connection->mosq, mosq_publish_callback);
mosquitto_disconnect_callback_set(connection->mosq, mosq_disconnect_callback);
}
/**
* @brief This routine is called when a disconnect request is processed
*
* @details This callback performs cleanup housekeeping
*
* @param[in] *mosq Pointer to the mosquitto structure associated with the request
* @param[in] *user_data Pointer to userdata that was set up when the connection was created
* @param[in] *rc Return code associated with the disconnect request
*/
void mosq_disconnect_callback(struct mosquitto *mosq, void *user_data, int rc)
{
mosquitto_profile_t *profile = NULL;
mosquitto_connection_t *connection = NULL;
mosquitto_mosq_userdata_t *userdata = NULL;
if (!user_data) {
log(SWITCH_LOG_ERROR, "disconnect userdata NULL rc:%d\n", rc);
return;
} else {
userdata = (mosquitto_mosq_userdata_t *)user_data;
}
if (!userdata->connection) {
log(SWITCH_LOG_ERROR, "disconnect connection NULL rc:%d\n", rc);
return;
} else {
connection = userdata->connection;
}
if (!userdata->profile) {
log(SWITCH_LOG_ERROR, "disconnect profile NULL rc:%d\n", rc);
return;
} else {
profile = userdata->profile;
}
log(SWITCH_LOG_DEBUG, "Profile %s connection %s rc %d disconnected", profile->name, connection->name, rc);
// connection->connected = SWITCH_FALSE;
// mosq_loop_stop(connection, SWITCH_TRUE);
log(SWITCH_LOG_ALERT, "Reconnect rc %d\n", mosquitto_reconnect(connection->mosq));
}
/**
* @brief This routine is called when a publish request is processed
*
* @details This callback currently only logs the published message
*
* @param[in] *mosq Pointer to the mosquitto structure associated with the request
* @param[in] *user_data Pointer to userdata that was set up when the connection was created
* @param[in] *message_id Message ID of the published message
*/
void mosq_publish_callback(struct mosquitto *mosq, void *user_data, int message_id)
{
mosquitto_mosq_userdata_t *userdata;
mosquitto_profile_t *profile;
mosquitto_connection_t *connection;
userdata = (mosquitto_mosq_userdata_t *)user_data;
profile = userdata->profile;
connection = userdata->connection;
log(SWITCH_LOG_DEBUG, "Profile %s connection %s message id %d published\n", profile->name, connection->name,
message_id);
mosquitto_log(SWITCH_LOG_INFO, "Profile %s connection %s message id %d published\n", profile->name,
connection->name, message_id);
profile_log(SWITCH_LOG_INFO, profile, "Profile %s connection %s message id %d published\n", profile->name,
connection->name, message_id);
}
/**
* @brief This routine is called when a message is received from an MQTT broker
*
* @details This callback processes and possbily takes action based on the content of the received message
*
* @param[in] *mosq Pointer to the mosquitto structure associated with the request
* @param[in] *user_data Pointer to userdata that was set up when the connection was created
* @param[in] *message Pointer to the mosquitto message structure
*/
void mosq_message_callback(struct mosquitto *mosq, void *user_data, const struct mosquitto_message *message)
{
char *payload_string = NULL;
mosquitto_mosq_userdata_t *userdata = NULL;
if (!message->payloadlen) {
log(SWITCH_LOG_DEBUG, "mosq_message_callback(): Received topic %s NULL message exiting.\n",
(char *)message->topic);
return;
}
if (!user_data) {
log(SWITCH_LOG_DEBUG, "mosq_message_callback(): Received topic %s user_data NULL exiting.\n",
(char *)message->topic);
return;
}
userdata = (mosquitto_mosq_userdata_t *)user_data;
if (!(payload_string = strndup((char *)message->payload, message->payloadlen))) {
log(SWITCH_LOG_ERROR, "mosq_message_callback(): Out of memory trying to duplicate %s\n",
(char *)message->payload);
return;
}
log(SWITCH_LOG_DEBUG, "Profile %s received topic %s payloadlen %d message %s\n", userdata->profile->name,
(char *)message->topic, message->payloadlen, payload_string);
if (mosquitto_globals.log.details) {
mosquitto_log(SWITCH_LOG_INFO, "Profile %s received topic %s payloadlen %d message %s\n",
userdata->profile->name, (char *)message->topic, message->payloadlen, payload_string);
} else {
mosquitto_log(SWITCH_LOG_INFO, "Profile %s received topic %s\n", userdata->profile->name,
(char *)message->topic);
}
if (userdata->profile->log->details) {
profile_log(SWITCH_LOG_INFO, userdata->profile, "Profile %s received topic %s payloadlen %d message %s\n",
userdata->profile->name, (char *)message->topic, message->payloadlen, payload_string);
} else {
profile_log(SWITCH_LOG_INFO, userdata->profile, "Profile %s received topic %s\n", userdata->profile->name,
(char *)message->topic);
}
if (!strncasecmp(payload_string, "bgapi", 5)) {
process_bgapi_message(userdata, payload_string, message);
} else if (!strncasecmp(payload_string, "originate", 9)) {
process_originate_message(userdata, payload_string, message);
}
switch_safe_free(payload_string);
}
/**
* @brief This routine is called when a subscribe has been processed
*
* @details This callback currently logs information related to the subscription
*
* @param[in] *mosq Pointer to the mosquitto structure associated with the request
* @param[in] *userdata Pointer to userdata that was set up when the connection was created
* @param[in] *mid Message ID
* @param[in] *qos_count Quality of Service
* @param[in] *granted_qos Granted Quality of Service
*/
void mosq_subscribe_callback(struct mosquitto *mosq, void *user_data, int mid, int qos_count, const int *granted_qos)
{
mosquitto_mosq_userdata_t *userdata;
mosquitto_profile_t *profile;
mosquitto_connection_t *connection;
userdata = (mosquitto_mosq_userdata_t *)user_data;
profile = userdata->profile;
connection = userdata->connection;
log(SWITCH_LOG_DEBUG, "Profile %s connection %s message id %d qos %d subscribed\n", profile->name, connection->name,
mid, granted_qos[0]);
mosquitto_log(SWITCH_LOG_INFO, "Profile %s connection %s message id %d qos %d subscribed\n", profile->name,
connection->name, mid, granted_qos[0]);
profile_log(SWITCH_LOG_INFO, profile, "Profile %s connection %s message id %d qos %d subscribed\n", profile->name,
connection->name, mid, granted_qos[0]);
}
/**
* @brief This routine is called for ALL messages, regardless of level
*
* @details This callback currently logs the message
*
* @param[in] mosq The mosquitto instance making the callback.
* @param[in] userdata The user data provided in mosquitto_new
* @param[in] level The log message level from the values: MOSQ_LOG_INFO MOSQ_LOG_NOTICE MOSQ_LOG_WARNING MOSQ_LOG_ERR
* MOSQ_LOG_SWITCH_LOG_DEBUG
* @param[in] str The message string.
*/
void mosq_log_callback(struct mosquitto *mosq, void *userdata, int level, const char *str)
{
switch_log_level_t log_level = SWITCH_LOG_DEBUG;
switch (level) {
case MOSQ_LOG_INFO:
log_level = SWITCH_LOG_INFO;
break;
case MOSQ_LOG_NOTICE:
log_level = SWITCH_LOG_NOTICE;
break;
case MOSQ_LOG_WARNING:
log_level = SWITCH_LOG_WARNING;
break;
case MOSQ_LOG_ERR:
log_level = SWITCH_LOG_ERROR;
break;
case MOSQ_LOG_DEBUG:
log_level = SWITCH_LOG_DEBUG;
break;
}
/* Print log messages after having converted the mosquitto levels into FreeSWITCH levels */
log(log_level, "mosq_log_callback() %s\n", str);
}
/**
* @brief This routine is called when a connect request has been processed
*
* @details This callback performs housekeeping related to the connect request
*
* @param[in] *mosq Pointer to the mosquitto structure associated with the request
* @param[in] *user_data Pointer to userdata that was set up when the connection was created
* @param[in] *result Result of the connection
*/
void mosq_connect_callback(struct mosquitto *mosq, void *user_data, int result)
{
mosquitto_mosq_userdata_t *userdata = NULL;
mosquitto_connection_t *connection = NULL;
switch_bool_t force_loop_stop = SWITCH_TRUE;
log(SWITCH_LOG_DEBUG, "mosq_connect_callback() result %d\n", result);
if (!user_data) {
return;
} else {
userdata = (mosquitto_mosq_userdata_t *)user_data;
}
if (!userdata->connection) {
return;
} else {
connection = userdata->connection;
}
if (!result) {
mosquitto_profile_t *profile = NULL;
log(SWITCH_LOG_CONSOLE, "mosq_connect_callback() Profile %s connection %s successful\n",
connection->profile_name, connection->name);
connection->retry_count = 0;
connection->connected = SWITCH_TRUE;
profile = locate_profile(connection->profile_name);
profile_activate(profile);
// mosquitto_subscribe(mosq, NULL, "FreeSWITCH/command", 2);
// mosquitto_subscribe(mosq, NULL, "$SYS/#", 2);
} else {
if (connection->retries && (connection->retry_count == connection->retries)) {
log(SWITCH_LOG_CONSOLE, "mosq_connect_callback() Profile %s connection to %s retried %d times, stopping\n",
connection->profile_name, connection->name, connection->retry_count);
mosq_disconnect(connection, force_loop_stop);
mosq_destroy(connection);
}
connection->retry_count++;
}
}
/**
* @brief This routine is called to set some initialization options before a connection is requested
*
* @details This routine sets up various options, such as the version of the MQTT protocol to used with the connection
*
* MOSQ_OPT_PROTOCOL_VERSION - Value must be set to either MQTT_PROTOCOL_V31, MQTT_PROTOCOL_V311, or
*MQTT_PROTOCOL_V5. Must be set before the client connects. Defaults to MQTT_PROTOCOL_V311.
*
* MOSQ_OPT_RECEIVE_MAXIMUM - Value can be set between 1 and 65535 inclusive, and represents the maximum number
*of incoming QoS 1 and QoS 2 messages that this client wants to process at once. Defaults to 20. This option is not
*valid for MQTT v3.1 or v3.1.1 clients. Note that if the MQTT_PROP_RECEIVE_MAXIMUM property is in the proplist passed
*to mosquitto_connect_v5(), then that property will override this option. Using this option is the recommended method
*however.
*
* MOSQ_OPT_SEND_MAXIMUM - Value can be set between 1 and 65535 inclusive, and represents the maximum number of
* outgoing QoS 1 and QoS 2 messages that this client will attempt to have “in flight” at once.
* Defaults to 20. This option is not valid for MQTT v3.1 or v3.1.1 clients. Note that if the broker being
*connected to sends a MQTT_PROP_RECEIVE_MAXIMUM property that has a lower value than this option, then the broker
*provided value will be used.
*
* MOSQ_OPT_SSL_CTX_WITH_DEFAULTS - If value is set to a non zero value, then the user specified SSL_CTX passed in
*using MOSQ_OPT_SSL_CTX will have the default options applied to it. This means that you only need to change the
*values that are relevant to you. If you use this option then you must configure the TLS options as normal, i.e. you
*should use mosquitto_tls_set to configure the cafile/capath as a minimum. This option is only available for
*openssl 1.1.0 and higher.
*
* MOSQ_OPT_TLS_OCSP_REQUIRED - Set whether OCSP checking on TLS connections is required. Set to 1 to enable
*checking, or 0 (the default) for no checking.
*
* @param[in] *connection Pointer to a connection structure
*
* @retval SWITCH_STATUS_SUCCESS
*/
switch_status_t mosq_int_option(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
int rc;
int protocol_version = MQTT_PROTOCOL_V31;
if (!strncasecmp(connection->protocol_version, "V311", 4)) { protocol_version = MQTT_PROTOCOL_V311; }
/*
* mosq A valid mosquitto instance.
* option The option to set.
* value The option specific value.
*/
#if LIBMOSQUITTO_VERSION_NUMBER < 1006008
rc = mosquitto_opts_set(connection->mosq, MOSQ_OPT_PROTOCOL_VERSION, &protocol_version);
#else
rc = mosquitto_int_option(connection->mosq, MOSQ_OPT_PROTOCOL_VERSION, protocol_version);
#endif
log(SWITCH_LOG_DEBUG, "Options set for Profile %s connection %s protocol version %s rc %d\n",
connection->profile_name, connection->name, connection->protocol_version, rc);
/*
rc = mosquitto_init_option(connection->mosq, MOSQ_OPT_RECEIVE_MAXIMUM, connection->receive_maximum);
rc = mosquitto_init_option(connection->mosq, MOSQ_OPT_SEND_MAXIMUM, connection->send_maximum);
rc = mosquitto_init_option(connection->mosq, MOSQ_OPT_SSL_CTX_WITH_DEFAULTS, connection->ssl_ctx_with_defaults);
rc = mosquitto_init_option(connection->mosq, MOSQ_OPT_TLS_OCSP_REQUIRED, connection->tls_ocsp_required);
*/
return status;
}
/**
* @brief This routine is called to set the reconnect options for the connection
*
* @details This routine sets the reconnect_delay, reconnect_delay_max and reconnect_exponential_backup values for the
* connection
*
* @param[in] *connection Pointer to a connection structure
*
* @retval SWITCH_STATUS_SUCCESS
*/
switch_status_t mosq_reconnect_delay_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
int rc;
rc = mosquitto_reconnect_delay_set(connection->mosq, connection->reconnect_delay, connection->reconnect_delay_max,
connection->reconnect_exponential_backoff);
switch (rc) {
case MOSQ_ERR_SUCCESS:
log(SWITCH_LOG_DEBUG,
"Succeeded setting reconnect delay for profile %s connection %s delay %d delay_max %d backoff %s\n",
connection->profile_name, connection->name, connection->reconnect_delay, connection->reconnect_delay_max,
connection->reconnect_exponential_backoff ? "enabled" : "disabled");
break;
case MOSQ_ERR_INVAL:
log(SWITCH_LOG_DEBUG,
"Failed setting reconnect delay for profile %s connection %s delay %d delay_max %d backoff %s invalid "
"parameters\n",
connection->profile_name, connection->name, connection->reconnect_delay, connection->reconnect_delay_max,
connection->reconnect_exponential_backoff ? "enabled" : "disabled");
return SWITCH_STATUS_GENERR;
default:
log(SWITCH_LOG_DEBUG,
"Failed setting reconnect delay for profile %s connection %s delay %d delay_max %d backoff %s unknown "
"return code %d\n",
connection->profile_name, connection->name, connection->reconnect_delay, connection->reconnect_delay_max,
connection->reconnect_exponential_backoff ? "enabled" : "disabled", rc);
return SWITCH_STATUS_GENERR;
}
return status;
}
/**
* @brief This routine sets the number of times that a message will be retried (if unsuccessful)
*
* @details This routine sets the message_retry count to the user specified value
*
* @param[in] *connection Pointer to a connection structure
*
* @retval SWITCH_STATUS_SUCCESS
*/
switch_status_t mosq_message_retry_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
if (connection->message_retry > 0) {
mosquitto_message_retry_set(connection->mosq, connection->message_retry);
log(SWITCH_LOG_DEBUG, "Message retry set to %d for profile %s connection %s\n", connection->message_retry,
connection->profile_name, connection->name);
}
return status;
}
/**
* @brief This routine sets the number of inflight messages to an MQTT broker
*
* @details This routine sets the mad_inflight_messages count to the user specified value
*
* @param[in] *connection Pointer to a connection structure
*
* @retval SWITCH_STATUS_SUCCESS
*/
switch_status_t mosq_max_inflight_messages_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
if (connection->max_inflight_messages > 0) {
int rc = mosquitto_max_inflight_messages_set(connection->mosq, connection->max_inflight_messages);
switch (rc) {
case MOSQ_ERR_SUCCESS:
log(SWITCH_LOG_DEBUG, "Max inflight messages set to %d for profile %s connection %s\n",
connection->max_inflight_messages, connection->profile_name, connection->name);
break;
case MOSQ_ERR_INVAL:
log(SWITCH_LOG_DEBUG,
"Max inflight messages set to %d for profile %s connection %s resulted in invalid parameter input\n",
connection->max_inflight_messages, connection->profile_name, connection->name);
return SWITCH_STATUS_GENERR;
}
}
return status;
}
/**
* @brief This routine sets the username and password used to connect to the MQTT broker
*
* @details Configure username and password for a mosquitto instance.
* By default, no username or password will be sent.
* For v3.1 and v3.1.1 clients, if username is NULL, the password argument is ignored.
* This is must be called before calling mosquitto_connect.
*
* @param[in] *connection Pointer to a connection structure
*
* @retval SWITCH_STATUS_SUCCESS
*/
switch_status_t mosq_username_pw_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
if (connection->username && connection->password) {
/*
* mosq A valid mosquitto instance.
* username The username to send as a string, or NULL to disable authentication.
* password The password to send as a string. Set to NULL when username is valid in order to send just a
* username.
*/
int rc = mosquitto_username_pw_set(connection->mosq, connection->username, connection->password);
switch (rc) {
case MOSQ_ERR_SUCCESS:
log(SWITCH_LOG_DEBUG, "Client username set to %s\n", connection->username);
break;
case MOSQ_ERR_INVAL:
log(SWITCH_LOG_ERROR, "Setting username/pw %s failed invalid parameters\n", connection->username);
return SWITCH_STATUS_GENERR;
case MOSQ_ERR_NOMEM:
log(SWITCH_LOG_ERROR, "Setting username/pw %s failed out of memory\n", connection->username);
return SWITCH_STATUS_GENERR;
default:
log(SWITCH_LOG_ERROR, "Setting username/pw %s unknown return code %d\n", connection->username, rc);
return SWITCH_STATUS_GENERR;
}
}
return status;
}
/*
* @brief This routine configures the client for certificate based SSL/TLS support.
*
* @details The TLS configuration options specified in the 'connection' section define various
* parameters related to setting up a TLS connection to a broker. This routine takes
* the settings and add them to the client mosq structure (prior to connecting to a broker).
*
* @param[in] *connection Pointer to a connections structure
*
* @retval status returned by the mosquitto library of the attempt to set the TLS parameters
*/
switch_status_t mosq_tls_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
int rc;
if (!connection) {
log(SWITCH_LOG_ERROR, "Cannot execute mosquitto_tls_set because connection name is NULL\n");
return SWITCH_STATUS_GENERR;
}
/*
* mosq a valid mosquitto instance.
* cafile path to a file containing the PEM encoded trusted CA certificate files. Either cafile or capath must
*not be NULL. capath path to a directory containing the PEM encoded trusted CA certificate files. See
*mosquitto.conf for more details on configuring this directory. Either cafile or capath must not be NULL. certfile
*path to a file containing the PEM encoded certificate file for this client. If NULL, keyfile must also be NULL and
*no client certificate will be used. keyfile path to a file containing the PEM encoded private key for this
*client. If NULL, certfile must also be NULL and no client certificate will be used. pw_callback if keyfile is
*encrypted, set pw_callback to allow your client to pass the correct password for decryption. If set to NULL, the
*password must be entered on the command line. Your callback must write the password into “buf”, which is “size”
*bytes long. The return value must be the length of the password. “userdata” will be set to the calling mosquitto
*instance. The mosquitto userdata member variable can be retrieved using mosquitto_userdata.
*/
rc = mosquitto_tls_set(connection->mosq, connection->tls.cafile, connection->tls.capath, connection->tls.certfile,
connection->tls.certfile, NULL);
switch (rc) {
case MOSQ_ERR_SUCCESS:
log(SWITCH_LOG_INFO, "mosquitto_tls_set TLS profile: %s connection: %s %s:%d\n", connection->profile_name,
connection->name, connection->host, connection->port);
status = SWITCH_STATUS_SUCCESS;
break;
case MOSQ_ERR_INVAL:
log(SWITCH_LOG_ERROR, "mosquitto_tls_set profile: %s connection: %s %s:%d input parameters were invalid\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
case MOSQ_ERR_NOMEM:
log(SWITCH_LOG_ERROR, "mosquitto_tls_set profile: %s connection: %s %s:%d out of memory condition occurred\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
}
if (connection->tls.advanced_options == SWITCH_TRUE) { mosq_tls_opts_set(connection); }
return status;
}
/*
* @brief This routine configures a client for pre-shared-key based TLS support
*
* @details Configure the client for pre-shared-key based TLS support.
* Must be called before mosquitto_connect.
* Cannot be used in conjunction with mosquitto_tls_set.
*
* @param[in] *connection Pointer to a connections structure
*
* @retval status returned by the mosquitto library of the attempt to set the pre-shared-key based TLS parameters
*/
switch_status_t mosq_tls_psk_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
int rc;
if (!connection) {
log(SWITCH_LOG_ERROR, "Cannot execute mosquitto_tls_psk_set because connection name is NULL\n");
return SWITCH_STATUS_GENERR;
}
/*
* mosq a valid mosquitto instance.
* psk the pre-shared-key in hex format with no leading “0x”.
* identity the identity of this client. May be used as the username depending on the server settings.
* ciphers a string describing the PSK ciphers available for use.
* See the “openssl ciphers” tool for more information. If NULL, the default ciphers will be used.
*/
rc = mosquitto_tls_psk_set(connection->mosq, connection->tls.psk, connection->tls.identity,
connection->tls.psk_ciphers);
switch (rc) {
case MOSQ_ERR_SUCCESS:
log(SWITCH_LOG_INFO, "mosquitto_tls_psk_set profile: %s connection %s %s:%d\n", connection->profile_name,
connection->name, connection->host, connection->port);
status = SWITCH_STATUS_SUCCESS;
break;
case MOSQ_ERR_INVAL:
log(SWITCH_LOG_ERROR, "mosquitto_tls_psk_set: profile: %s connection %s %s:%d input parameters were invalid\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
case MOSQ_ERR_NOMEM:
log(SWITCH_LOG_ERROR,
"mosquitto_tls_psk_set: profile: %s connection %s %s:%d out of memory condition occurred\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
}
if (connection->tls.advanced_options == SWITCH_TRUE) { mosq_tls_opts_set(connection); }
return status;
}
/*
* @brief This routine configures a client for advacned SSL/TLS options.
*
* @details Configure the client for advanced SSL/TLS options.
* Must be called before mosquitto_connect.
*
* @param[in] *connection Pointer to a connection structure
*
* @retval status returned by the mosquitto library of the attempt to set advanced SSL/TLS options.
*/
switch_status_t mosq_tls_opts_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
int rc;
if (!connection) {
log(SWITCH_LOG_ERROR, "Cannot execute mosquitto_tls_opts_set because connection name is NULL\n");
return SWITCH_STATUS_GENERR;
}
/*
* mosq a valid mosquitto instance.
* cert_reqs an integer defining the verification requirements the client will impose on the server. This can be
*one of: SSL_VERIFY_NONE (0): the server will not be verified in any way. SSL_VERIFY_PEER (1): the server
*certificate will be verified and the connection aborted if the verification fails. The default and recommended
*value is SSL_VERIFY_PEER. Using SSL_VERIFY_NONE provides no security. version the version of the SSL/TLS
*protocol to use as a string. If NULL, the default value is used. The default value and the available values
*depend on the version of openssl that the library was compiled against. For openssl >= 1.0.1, the available
*options are tlsv1.2, tlsv1.1 and tlsv1, with tlv1.2 as the default. For openssl < 1.0.1, only tlsv1 is available.
* ciphers a string describing the ciphers available for use. See the “openssl ciphers” tool for more
*information. If NULL, the default ciphers will be used.
*/
rc = mosquitto_tls_opts_set(connection->mosq, connection->tls.cert_reqs, connection->tls.version,
connection->tls.opts_ciphers);
switch (rc) {
case MOSQ_ERR_SUCCESS:
log(SWITCH_LOG_INFO, "mosquitto_tls_opts_set profile: %s connection: %s %s:%d\n", connection->profile_name,
connection->name, connection->host, connection->port);
status = SWITCH_STATUS_SUCCESS;
break;
case MOSQ_ERR_INVAL:
log(SWITCH_LOG_ERROR, "mosquitto_tls_opts_set: profile: %s connection %s %s:%d input parameters were invalid\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
case MOSQ_ERR_NOMEM:
log(SWITCH_LOG_ERROR,
"mosquitto_tls_opts_set: profile: %s connection %s %s:%d out of memory condition occurred\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
}
return status;
}
/*
* @brief This routine configures a Last Will and Testament (will) for a client.
*
* @details Configure will information for a mosquitto instance.
* By default, clients do not have a will.
* This must be called before calling mosquitto_connect.
*
* @param[in] *connection Pointer to a connection structure
*
* @retval status returned by the mosquitto library of the attempt to set a will.
*/
switch_status_t mosq_will_set(mosquitto_connection_t *connection)
{
switch_status_t status = SWITCH_STATUS_SUCCESS;
int rc;
int payloadlen = 0;
if (!connection) {
log(SWITCH_LOG_ERROR, "Cannot execute mosquitto_will_set() because connection name is NULL\n");
return SWITCH_STATUS_GENERR;
}
payloadlen = strlen(connection->will.payload);
/*
* mosq A valid mosquitto instance.
* topic The topic on which to publish the will.
* payloadlen The size of the payload (bytes).
* Valid values are between 0 and 268,435,455.
* payload Pointer to the data to send.
* If payloadlen > 0 this must be a valid memory location.
* qos Integer value 0, 1 or 2 indicating the Quality of Service to be used for the will.
* retain Set to true to make the will a retained message.
*/
rc = mosquitto_will_set(connection->mosq, connection->will.topic, payloadlen, connection->will.payload,
connection->will.qos, connection->will.retain);
switch (rc) {
case MOSQ_ERR_SUCCESS:
log(SWITCH_LOG_INFO, "mosquitto_will_set profile %s connection: %s %s:%d\n", connection->profile_name,
connection->name, connection->host, connection->port);
status = SWITCH_STATUS_SUCCESS;
break;
case MOSQ_ERR_INVAL:
log(SWITCH_LOG_ERROR, "mosquitto_will_set profile %s connection %s %s:%d input parameters were invalid\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
case MOSQ_ERR_NOMEM:
log(SWITCH_LOG_ERROR, "mosquitto_will_set profile %s connection %s %s:%d out of memory condition occurred\n",
connection->profile_name, connection->name, connection->host, connection->port);
status = SWITCH_STATUS_GENERR;
return status;
case MOSQ_ERR_PAYLOAD_SIZE:
log(SWITCH_LOG_ERROR, "mosquitto_will_set profile %s connection %s %s:%d payload size %d is too large\n",
connection->profile_name, connection->name, connection->host, connection->port, payloadlen);
status = SWITCH_STATUS_GENERR;
return status;
case MOSQ_ERR_MALFORMED_UTF8:
log(SWITCH_LOG_ERROR, "mosquitto_will_set profile %s connection %s %s:%d the topic %s is not valid UTF-8\n",
connection->profile_name, connection->name, connection->host, connection->port, connection->will.topic);
status = SWITCH_STATUS_GENERR;
return status;
}
return status;
}