messageQueue.c 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. // includes
  2. #include <stdio.h>
  3. #include <string.h>
  4. #include <unistd.h>
  5. #include "mosquitto.h"
  6. #include "runtime.h"
  7. #include "messageQueue.h"
  8. #define MQTT_HOST "localhost"
  9. #define MQTT_PORT 1883
  10. // TODO: add ifndef to these ones
  11. #define MQTT_QOS 0
  12. #define MQTT_KEEPALIVE 60
  13. #define MQTT_MAX_CONNECT_TRIES 10
  14. #define MQTT_CONNECT_WAIT 5
  15. // global vars
  16. struct topicList {
  17. char** list;
  18. int numElements;
  19. } ;
  20. struct mosquitto *mosqPtr;
  21. int bConnected;
  22. // local functions
  23. void mqtt_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
  24. {
  25. int i;
  26. char* payloadptr;
  27. char payload[1024];
  28. duk_context *dukCtx = (duk_context*)userdata;
  29. printf("Message arrived to topic: %s, message is %d chars long\n", (char*)message->topic, message->payloadlen);
  30. printf(" message: '%s'\n", (char*)message->payload);
  31. handleMessage((char*)message->topic, (char*)message->payload);
  32. }
  33. void mqtt_connect_callback(struct mosquitto *mosq, void *userdata, int result)
  34. {
  35. int i;
  36. if(!result){
  37. /* Subscribe to required topics on successful connect. */
  38. mosquitto_subscribe(mosq, NULL, MQ_CARD_TOPIC, MQTT_QOS);
  39. mosquitto_subscribe(mosq, NULL, MQ_STATUS_TOPIC, MQTT_QOS);
  40. mosquitto_subscribe(mosq, NULL, MQ_NOTIFICATION_TOPIC, MQTT_QOS);
  41. }else{
  42. fprintf(stderr, "Connect failed\n");
  43. }
  44. }
  45. void mqtt_subscribe_callback(struct mosquitto *mosq, void *userdata, int mid, int qos_count, const int *granted_qos)
  46. {
  47. int i;
  48. printf("Subscribed (mid: %d): %d", mid, granted_qos[0]);
  49. for(i=1; i<qos_count; i++){
  50. printf(", %d", granted_qos[i]);
  51. }
  52. printf("\n");
  53. }
  54. void mqtt_disconnect_callback(struct mosquitto *mosq, void *userdata, int rc) {
  55. printf("Message Queue disconnected! (%d)\n", rc);
  56. // TODO: potentially attempt to reconnect here?
  57. }
  58. int messageQueueConnect(char* host, int port) {
  59. printf("initializing message queue\n");
  60. bConnected = 0;
  61. // instantiate new client (with a unique client id)
  62. mosqPtr = mosquitto_new(NULL, true, NULL);
  63. // onion.io: note that userdata is not passed along for now
  64. if(!mosqPtr){
  65. fprintf(stderr, "Error: Out of memory.\n");
  66. return 1;
  67. }
  68. // set callback functions
  69. mosquitto_connect_callback_set(mosqPtr, mqtt_connect_callback);
  70. mosquitto_message_callback_set(mosqPtr, mqtt_message_callback);
  71. mosquitto_subscribe_callback_set(mosqPtr, mqtt_subscribe_callback);
  72. mosquitto_disconnect_callback_set(mosqPtr, mqtt_disconnect_callback);
  73. // start the connection
  74. if(mosquitto_connect(mosqPtr, host, port, MQTT_KEEPALIVE)){
  75. fprintf(stderr, "MQTT unable to connect.\n");
  76. mosquitto_destroy(mosqPtr);
  77. return 1;
  78. }
  79. // start the loop
  80. int loop = mosquitto_loop_start(mosqPtr);
  81. if(loop != MOSQ_ERR_SUCCESS){
  82. fprintf(stderr, "Unable to start loop: %i\n", loop);
  83. mosquitto_destroy(mosqPtr);
  84. return 1;
  85. }
  86. return 0;
  87. }
  88. // API functions
  89. int initMessageQueue(char* host, int port) {
  90. int count = 0;
  91. int status = 1;
  92. // initialize the library
  93. mosquitto_lib_init();
  94. while (count < MQTT_MAX_CONNECT_TRIES) {
  95. if (messageQueueConnect(host, port) == 0) {
  96. printf("message queue initialized (%d)\n", count);
  97. status = 0;
  98. break;
  99. }
  100. count++;
  101. sleep(MQTT_CONNECT_WAIT);
  102. }
  103. return status;
  104. }
  105. void destroyMessageQueue () {
  106. /* mqtt release */
  107. mosquitto_destroy(mosqPtr);
  108. mosquitto_lib_cleanup();
  109. }
  110. int sendMessage (char* topic, char* payload) {
  111. int status = 0;
  112. status = mosquitto_publish( mosqPtr,
  113. NULL,
  114. topic,
  115. strlen(payload),
  116. payload,
  117. MQTT_QOS,
  118. false
  119. );
  120. return status;
  121. }
  122. int messageQueueSubscribe (char* topic) {
  123. int status = -1;
  124. printf("Subscribe to topic '%s' ", topic);
  125. status = mosquitto_subscribe( mosqPtr,
  126. NULL,
  127. topic,
  128. MQTT_QOS
  129. );
  130. if (status == MOSQ_ERR_SUCCESS) {
  131. printf("successful\n");
  132. } else if (status == MOSQ_ERR_NO_CONN) {
  133. printf("not successful, not connected to a broker\n");
  134. }
  135. return status;
  136. }