MQTT loopback sample

mqtt_loop.c

/*
 * Linumiz <parthiban@linumiz.com>
 * mqtt_loop: subscribe to topic and loopbacks the message to publish topic
 *
 * gcc -I./lib -L./lib -DMQTT_TLS -o ./mqtt_loop ./mqtt_loop.c -lmosquitto
 */

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

#include <mosquitto.h>

#define SUB_TOPIC	"topic/to/subscribe"
#define PUB_TOPIC	"topic/to/publish"

#ifdef MQTT_TLS
#define MOSQ_CA_FILE	"./certs/ca.pem"
#define MOSQ_CERT_FILE	"./certs/client.crt"
#define MOSQ_KEY_FILE	"./certs/client.key"
#endif

static void connect_cb(struct mosquitto *mosq, void *obj, int result)
{
	fprintf(stderr, "%s\n", __func__);
}

static void disconnect_cb(struct mosquitto *mosq, void *obj, int result)
{
	fprintf(stderr, "%s\n", __func__);
}

static void subscribe_cb(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
	fprintf(stderr, "%s\n", __func__);
}

void message_cb(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
{
	int rc;
	struct mosquitto *pub = (struct mosquitto *)obj;

	rc = mosquitto_publish(pub, NULL, PUB_TOPIC, message->payloadlen, message->payload, message->qos, message->retain);
	if (rc < 0) {
		printf("Unable to publish\n");
	}
}

int main(int argc, char **argv)
{
	int rc;
	struct mosquitto *sub;
	struct mosquitto *pub;

	if (argc != 3) {
		printf("Usage %s <host> <port>\n", argv[0]);
		return 1;
	}

	mosquitto_lib_init();

	pub = mosquitto_new(NULL, true, NULL);
	sub = mosquitto_new(NULL, true, pub);
	if (!sub || !pub) {
		fprintf(stderr, "Unable to create mqtt: %s\n", strerror(errno));
		goto error;
	}

#ifdef MQTT_TLS
	rc = mosquitto_tls_set(pub, MOSQ_CA_FILE, NULL, MOSQ_CERT_FILE, MOSQ_KEY_FILE, NULL);
	if (rc != MOSQ_ERR_SUCCESS) {
		fprintf(stderr, "Publish: Unable to set TLS: %s\n", strerror(errno));
		goto error;
	}

	rc = mosquitto_tls_opts_set(pub, 1, NULL, NULL);
	if (rc != MOSQ_ERR_SUCCESS) {
		fprintf(stderr, "Publish: Unable to set TLS opts: %s\n", strerror(errno));
		goto error;
	}

	rc = mosquitto_tls_set(sub, MOSQ_CA_FILE, NULL, MOSQ_CERT_FILE, MOSQ_KEY_FILE, NULL);
	if (rc != MOSQ_ERR_SUCCESS) {
		fprintf(stderr, "Subscribe: Unable to set TLS: %s\n", strerror(errno));
		goto error;
	}

	rc = mosquitto_tls_opts_set(sub, 1, NULL, NULL);
	if (rc != MOSQ_ERR_SUCCESS) {
		fprintf(stderr, "Subscribe: Unable to set TLS opts: %s\n", strerror(errno));
		goto error;
	}
#endif

	mosquitto_connect_callback_set(sub, connect_cb);
	mosquitto_disconnect_callback_set(sub, disconnect_cb);
	mosquitto_subscribe_callback_set(sub, subscribe_cb);
	mosquitto_message_callback_set(sub, message_cb);

	rc = mosquitto_connect(pub, argv[1], atoi(argv[2]), 5);
	if (rc != MOSQ_ERR_SUCCESS) {
		fprintf(stderr, "Publish: Unable to connect: %s\n", strerror(rc));
		goto error;
	}

	rc = mosquitto_connect(sub, argv[1], atoi(argv[2]), 5);
	if (rc != MOSQ_ERR_SUCCESS) {
		fprintf(stderr, "Subscribe: Unable to connect: %s\n", strerror(rc));
		goto error;
	}

	rc = mosquitto_subscribe(sub, NULL, SUB_TOPIC, 0);
	if (rc != MOSQ_ERR_SUCCESS) {
		mosquitto_disconnect(pub);
		mosquitto_disconnect(sub);
		fprintf(stderr, "Unable to subscribe: %s\n", strerror(errno));
		goto error;
	}

	mosquitto_loop_start(sub);
	mosquitto_loop_start(pub);
	while(true)
		sleep(180);

error:
	if (sub)
		mosquitto_destroy(sub);
	if (pub)
		mosquitto_destroy(pub);

	mosquitto_lib_cleanup();

	return 0;
}