MQTT sample code using libmosquitto

Sample code for publisher

#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <mosquitto.h>

// Server connection parameters
#define MQTT_HOSTNAME "localhost"
#define MQTT_PORT 1883
#define MQTT_TOPIC "/brook"

struct brook_obj {
 unsigned long long published;
 unsigned long long connected;
 unsigned long long disconnected;
 unsigned long long loged;
};

/*
 mosq the mosquitto instance making the callback.
 obj the user data provided in mosquitto_new
 rc the return code of the connection response, one of:
*/
static void connect_cb(struct mosquitto *mosq, void *obj, int rc)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->connected++;
        printf("%s(), connect:%llu, published:%llu, loged:%llu, rc:%d\n",
  __FUNCTION__, bobj->connected, bobj->published, bobj->loged, rc);
}

static void publish_cb(struct mosquitto *mosq, void *obj, int mid)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->published++;
        printf("%s(), connect:%llu, published:%llu, loged:%llu, mid:%d\n",
  __FUNCTION__, bobj->connected, bobj->published, bobj->loged, mid);
}

static void log_cb(struct mosquitto *mosq, void *obj, int level, const char *str)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->loged++;
        printf("%s(), connect:%llu, published:%llu, loged:%llu, level:%d,\n\tstr:%s\n",
  __FUNCTION__, bobj->connected, bobj->published, bobj->loged, level, str);
}

static void disconnect_cb(struct mosquitto *mosq, void *obj, int rc)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->disconnected++;
        printf("%s(), connect:%llu, published:%llu, loged:%llu, disconnected:%llu\n",
  __FUNCTION__, bobj->connected, bobj->published, bobj->loged, bobj->disconnected);
}

/*
 * Start here
 */
int main (int argc, char **argv)
{
 char clientid[24], text[20];
 struct mosquitto *mosq = NULL;
 unsigned long long i;
 int rc, major, minor, rev;
 struct brook_obj brook_obj = {};

 /* Must be called before any other mosquitto functions.
    This function is not thread safe. */
  mosquitto_lib_init();

  rc = mosquitto_lib_version(&major, &minor, &rev);
  fprintf(stdout, "rc:%d, major:%d, minor:%d, rev:%d\n",
  rc, major, minor, rev);

  memset(clientid, 0, sizeof(clientid));
  snprintf(clientid, sizeof(clientid) - 1, "cid-pub-%d", getpid());
  mosq = mosquitto_new(clientid, true, &brook_obj);
 if (!mosq) {
  fprintf (stderr, "new mosq failed \n");
  exit(-1);
 }

  rc = mosquitto_tls_opts_set(mosq, 1, "tlsv1", NULL);
 if (rc) {
  fprintf (stderr, "set tls failed %d\n", rc);
  exit (-1);
 }

  rc = mosquitto_tls_set(mosq, "ca/client/ca.crt", NULL, "ca/client/client.crt", "ca/client/client.key", NULL);
  if (rc) {
  fprintf (stderr, "set tls failed %d\n", rc);
  exit (-1);
  }

   rc = mosquitto_tls_insecure_set(mosq, true);
  if (rc) {
  fprintf (stderr, "set insecure failed %d\n", rc);
  exit (-1);
  }

 // Set callback function
  mosquitto_connect_callback_set(mosq, connect_cb);
  mosquitto_publish_callback_set(mosq, publish_cb);
  mosquitto_log_callback_set(mosq, log_cb);
  mosquitto_disconnect_callback_set(mosq, disconnect_cb);

  rc = mosquitto_connect(mosq, MQTT_HOSTNAME, MQTT_PORT, 60);
  if (rc) {
   fprintf (stderr, "Can't connect to Mosquitto server. %d\n", rc);
   exit (-1);
  }

  for (i = 0; i < 10 ;i++) {
   sprintf(text, "%llu", i);
   printf("send %d, %s\n", (int)strlen(text), text);
   // Publish the message to the topic
  rc = mosquitto_publish(mosq, NULL, MQTT_TOPIC, strlen(text), text, 2, false);
  if (rc) {
   fprintf (stderr, "Can't publish to Mosquitto server\n");
   exit (-1);
  }
  if (!(i % 3)) {
   do {
    rc = mosquitto_loop(mosq, 3, 1);
    fprintf (stdout, "mosquitto_loop %d, published:%llu, i:%llu\n",
      rc, brook_obj.published, i);
   } while(rc == 0 && brook_obj.published < i);
  }
 }

 do {
  rc = mosquitto_loop(mosq, 3, 1);
  fprintf (stdout, "mosquitto_loop %d, published:%llu, i:%llu\n",
    rc, brook_obj.published, i);
 } while(rc == 0 && brook_obj.published != i);

 // Tidy up
 mosquitto_disconnect(mosq);
 mosquitto_destroy(mosq);
 mosquitto_lib_cleanup();

 return 0;
}

Sample code subscriber

#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdint.h>
#include <string.h>

#include <sys/types.h>
#include <unistd.h>

#include <mosquitto.h>

#define MQTT_HOSTNAME "localhost" 
#define MQTT_PORT 1883 
#define MQTT_TOPIC "/brook/#"

struct brook_obj {
 unsigned long long msgs;
 unsigned long long connected;
 unsigned long long disconnected;
 unsigned long long loged;
 unsigned long long subscribed;
 unsigned long long unsubscribed;
};

/*
 mosq the mosquitto instance making the callback.
 obj the user data provided in mosquitto_new
 rc the return code of the connection response, one of:
*/
static void connect_cb(struct mosquitto *mosq, void *obj, int rc)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->connected++;
 printf("%s(), connect:%llu, msgs:%llu, loged:%llu, rc:%d\n",
  __FUNCTION__, bobj->connected, bobj->msgs, bobj->loged, rc);
}

static void log_cb(struct mosquitto *mosq, void *obj, int level, const char *str)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->loged++;
 printf("%s(), connect:%llu, msgs:%llu, loged:%llu, level:%d,\n\tstr:%s\n",
  __FUNCTION__, bobj->connected, bobj->msgs, bobj->loged, level, str);
}

static void disconnect_cb(struct mosquitto *mosq, void *obj, int rc)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->disconnected++;
 printf("%s(), connect:%llu, msgs:%llu, loged:%llu, disconnected:%llu\n",
  __FUNCTION__, bobj->connected, bobj->msgs, bobj->loged, bobj->disconnected);
}

static void message_cb(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->msgs++;
 bool match = 0;
 static unsigned long long i = 0, j = 0;

 printf("%s(), connect:%llu, msgs:%llu, loged:%llu, disconnected:%llu\n",
  __FUNCTION__, bobj->connected, bobj->msgs, bobj->loged, bobj->disconnected);
 printf("%s(), got message mid:%d, '%.*s' for topic '%s'\n",
  __FUNCTION__, message->mid, message->payloadlen, (char*) message->payload, message->topic);

 mosquitto_topic_matches_sub(MQTT_TOPIC, message->topic, &match);
 if (match) {
  printf("got matched topic to %s\n", MQTT_TOPIC);
 }
}

static void unsubscribe_cb(struct mosquitto *mosq, void *obj, int mid)
{
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->unsubscribed++;
 printf("%s(), connect:%llu, msgs:%llu, loged:%llu, disconnected:%llu, unsubscribed:%llu\n",
  __FUNCTION__, bobj->connected, bobj->msgs, bobj->loged, bobj->disconnected,
  bobj->unsubscribed);
}

static void subscribe_cb(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
 int i;
 struct brook_obj *bobj = (struct brook_obj*) obj;
 bobj->subscribed++;
 printf("%s(), mid:%d, qos_count:%d\n",
  __FUNCTION__, mid, qos_count);
 for (i = 0; i < qos_count; i++) {
  printf("\tQ[%d]:%d\n", i, granted_qos[i]);
 }
 printf("%s(), connect:%llu, msgs:%llu, loged:%llu, disconnected:%llu, subscribed:%llu\n",
  __FUNCTION__, bobj->connected, bobj->msgs, bobj->loged, bobj->disconnected,
  bobj->subscribed);
}

int main(int argc, char *argv[])
{
 char clientid[24];
 struct mosquitto *mosq;
 int rc = 0;
 struct brook_obj brook_obj = {};

 mosquitto_lib_init();

 memset(clientid, 0, sizeof(clientid));
 snprintf(clientid, sizeof(clientid) - 1, "cid-sub-%d", getpid());
 mosq = mosquitto_new(clientid, true, &brook_obj);
 if (!mosq) {
  fprintf (stderr, "new mosq failed\n");
  exit(-1);
 }

 rc = mosquitto_tls_opts_set(mosq, 1, "tlsv1", NULL);
 if (rc) {
  fprintf (stderr, "set tls opt failed %d\n", rc);
  exit(-1);
 }

 rc = mosquitto_tls_insecure_set(mosq, true);
 if (rc) {
  fprintf (stderr, "set insecure failed %d\n", rc);
  exit(-1);
 }

 rc = mosquitto_tls_set(mosq, "ca/client/ca.crt", NULL, "ca/client/client.crt", "ca/client/client.key", NULL);
 if (rc) {
  fprintf (stderr, "set tls failed %d\n", rc);
  exit(-1);
 }

 mosquitto_connect_callback_set(mosq, connect_cb);
 mosquitto_log_callback_set(mosq, log_cb);
 mosquitto_subscribe_callback_set(mosq, subscribe_cb);
 mosquitto_unsubscribe_callback_set(mosq, unsubscribe_cb);
 mosquitto_message_callback_set(mosq, message_cb);
 mosquitto_disconnect_callback_set(mosq, disconnect_cb);

 rc = mosquitto_connect(mosq, MQTT_HOSTNAME, MQTT_PORT, 60);
 if (rc) {
  fprintf(stderr, "mosquitto_connect failed. %d\n", rc);
  exit(-1);
 }

 mosquitto_subscribe(mosq, NULL, MQTT_TOPIC, 1);

 do {
  rc = mosquitto_loop(mosq, 3000, 1);
  fprintf(stdout, "mosquitto_loop %d, msgs%llu\n",
   rc, brook_obj.msgs);
 } while(rc == 0 && brook_obj.msgs < 5);

 mosquitto_disconnect(mosq);
 mosquitto_destroy(mosq);
 mosquitto_lib_cleanup();

 return rc;
}