libmosquitto test program

mosqsub.c

// Connect to local mosquitto server over tls with no client authentication, then subscribe to everything.
// $(CC) -o mosqsub -lmosquitto mosqsub.c
/* vim: set sw=2 ts=2 expandtab: */

//! @todo reconnect lost connections automatically



#include <mosquitto.h>
#include <poll.h>
#include <assert.h>
#include <errno.h>
#include <stdio.h>
#include <inttypes.h>

int toStop = 0;


void mq_msg_cb(struct mosquitto * mq, void * userdata, const struct mosquitto_message * msg) {
  printf(">> [%" PRIu16 ",%d,%d] %s: %.*s\n", msg->mid, msg->qos, msg->retain, msg->topic, msg->payloadlen, (const char *)msg->payload);
}

int main(int argc, const char * argv[]) {
  int userdata = 0;

  int rc = mosquitto_lib_init();
  assert(rc == MOSQ_ERR_SUCCESS);

  struct mosquitto * mq = mosquitto_new("loggen", 1, &userdata);
  assert(mq);

  rc = mosquitto_tls_set(mq, "/etc/mosquitto/certs/ca.crt", 0, 0, 0, 0);
  if (rc != MOSQ_ERR_SUCCESS) {
    printf("tls init failed [%d]\n", rc);
    return -1;
  }

  rc = mosquitto_tls_insecure_set(mq, 1);
  assert(rc == MOSQ_ERR_SUCCESS);

  mosquitto_message_callback_set(mq, &mq_msg_cb);

  rc = mosquitto_connect(mq, "localhost", 1883, 10); 
  assert(rc == MOSQ_ERR_SUCCESS);

  rc = mosquitto_subscribe(mq, 0, "#", 2);
  assert(rc == MOSQ_ERR_SUCCESS);

  while (!toStop) {
    // Run the socket first
    rc = mosquitto_loop_misc(mq);
    assert(rc == MOSQ_ERR_SUCCESS);

    // Drain all queued write packets
    while (mosquitto_want_write(mq)) {
      rc = mosquitto_loop_write(mq,1);
      assert(rc == MOSQ_ERR_SUCCESS);
    }

    struct pollfd pfds[1] = { { mosquitto_socket(mq), POLLIN, 0 } };
    int prc = poll(pfds, 1, 1000); // wait 1 sec for fd to have something to read.
    if (prc == -1 && errno != EINTR) {
      perror("poll");
      break;
    }

    if (prc == 1) {
      if (pfds[0].revents & (POLLERR | POLLNVAL)) {
        // Socket error occurred on the mosquitto socket.
        break;
      }

      if (pfds[0].revents & POLLIN) {
        rc = mosquitto_loop_read(mq, 1);
        assert(rc == MOSQ_ERR_SUCCESS);
      } 
    }
  }

  rc = mosquitto_disconnect(mq);
  assert(rc == MOSQ_ERR_SUCCESS);

  mosquitto_destroy(mq);

  rc = mosquitto_lib_cleanup();
  assert(rc == MOSQ_ERR_SUCCESS);
  return 0;
}