Re-Visiting an Old Friend - ESP32-WROOM-32

There are three sections :

1. ESP32 Code
2. GPIO message translator
3. Nagios Interface
   

I've had an ESP32-WROOM-32 sitting around for a while, and I thought I'd throw it back into action as an MQTT data ingress point for my home system.  It's been a LONG time (about 8 years, I think) since I was implementing one of these things.  I needed the following :

• Base it on MQTT (message brokering) to reduce complexity of the tools required
• DHCP compatible on-network presence for a drop-in, rapid security network
• Ease of configuring or adapting a new sensor into the system
• Robust enough that it could actually run and not encounter a memory leak
  

Add into the mix that I don't use the Mac any longer, and suddenly, I'm starting from scratch.  I started out following Espressif's guide, but ran into a snag straight out of the gate, and that was I couldn't get the stupid thing to compile in the Espressif environment.  Bugger.  I have to go back to the Arduino IDE.

So, back into the Arduino IDE, slap some code together (usual copy-and-paste from things like Stack Overflow, the forums, etc).

ESP32 Code

I did have an issue where the ESP32 dropped off the network after a few hours.  That almost killed my DHCP requirement out of the gate, and then one night, I had a dream about lines of code.  A quick modification in the code to actually check if the ethernet had disconnected, and I had something that stayed with me a bit longer.  The code turned out to be :

/**
 * A simple firmware for sending MQTT queue messages based on GPIO changes
 * Written (well, hack and slashed from examples) by Joe Lewis, December 18, 2024
 */

/*
 * Some defaults
 */
//#define DEBUG_WIFI
#define USE_EXTERNAL_RES_PINS_34_39   1
#define DEFAULT_MQTT_CLIENT_PREFIX    "mqttgpio-"
#define DEFAULT_MQTT_BROKER           "mqttserverhostname"
#define DEFAULT_MQTT_TOPIC            "GPIO"
#define DEFAULT_INIT_MESSAGE          "init-"
#define DEFAULT_MQTT_MSG_PREFIX       "pin-"
#define DEFAULT_MQTT_PIN_SEPARATOR    "-"
#define DEFAULT_MQTT_PORT             1883
#define DEFAULT_WIRELESS_NETWORK_NAME ""
#define DEFAULT_WIRELESS_NETWORK_KEY  ""
#define HALF_SECONDS_BEFORE_ETHDOWN   60

typedef struct {
  int             p;
  int             v;
  int             lastSteadyState;
  int             lastFlickerableState;
  unsigned long   lastDebounceTime;
} pin_t;

#include "driver/gpio.h"
#include <soc/gpio_struct.h>

// the trailing 0 is for the end of the list
#ifdef USE_EXTERNAL_RES_PINS_34_39
#define NUMBER_OF_PINS  16
pin_t pins[NUMBER_OF_PINS] = {
   {  GPIO_NUM_0, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_1, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_2, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_3, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_4, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_5, 0, LOW, LOW, 0 }, // possible boot time issues
   { GPIO_NUM_13, 0, LOW, LOW, 0 },
   { GPIO_NUM_14, 0, LOW, LOW, 0 },
   { GPIO_NUM_15, 0, LOW, LOW, 0 },
   { GPIO_NUM_16, 0, LOW, LOW, 0 },
   { GPIO_NUM_32, 0, LOW, LOW, 0 },
   { GPIO_NUM_33, 0, LOW, LOW, 0 },
   { GPIO_NUM_34, 0, LOW, LOW, 0 }, // has a pre-built pull up resistor
   { GPIO_NUM_35, 0, LOW, LOW, 0 }, // has to have external pull up resistor soldered in
   { GPIO_NUM_36, 0, LOW, LOW, 0 }, // has to have external pull up resistor soldered in
   { GPIO_NUM_39, 0, LOW, LOW, 0 }  // has to have external pull up resistor soldered in  //USED
};
#else
#define NUMBER_OF_PINS  13
pin_t pins[NUMBER_OF_PINS] = {
   {  GPIO_NUM_0, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_1, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_2, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_3, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_4, 0, LOW, LOW, 0 }, // possible boot time issues
   {  GPIO_NUM_5, 0, LOW, LOW, 0 }, // possible boot time issues
   { GPIO_NUM_13, 0, LOW, LOW, 0 },
   { GPIO_NUM_14, 0, LOW, LOW, 0 },
   { GPIO_NUM_15, 0, LOW, LOW, 0 },
   { GPIO_NUM_16, 0, LOW, LOW, 0 },
   { GPIO_NUM_32, 0, LOW, LOW, 0 },
   { GPIO_NUM_33, 0, LOW, LOW, 0 },
   { GPIO_NUM_34, 0, LOW, LOW, 0 } // has a pre-built pull up resistor
};
#endif

// Important to be defined BEFORE including ETH.h for ETH.begin() to work.
// Example RMII LAN8720 (Olimex, etc.)
#ifndef ETH_PHY_TYPE
#define ETH_PHY_TYPE  ETH_PHY_LAN8720
#define ETH_PHY_ADDR  0
#define ETH_PHY_MDC   23
#define ETH_PHY_MDIO  18
#define ETH_PHY_POWER -1
#define ETH_CLK_MODE  ETH_CLOCK_GPIO0_IN
#endif

#define ESP_INTR_FLAG_DEFAULT 0
#define DEBOUNCE_TIME         50

#include <ETH.h>
#include <WiFi.h>
#include <PubSubClient.h>
#include "esp_mac.h"


char  mqtt_topic[255];
char  mqtt_broker[255];
char  mqtt_msg_prefix[32];
char  mqtt_client_id[32];
int   mqtt_port;
int   delayCountSinceLastSend;
int   sendCount;
char  wifi_ssid[128];
char  wifi_password[128];
bool  use_wifi;
bool  eth_connected;

WiFiClient espClient;
PubSubClient mqttClient(espClient);

typedef struct GPIOEvent_t {
  pin_t       *pin;
  GPIOEvent_t *next;
} GPIOEvent_t;
GPIOEvent_t *events;

/*
 * GPIO Functions
 */

void sendMessage(char *msg) {
  if (!mqttClient.connected()) {
    reconnect();
  }
  mqttClient.publish(mqtt_topic,msg);
}

void processEvent(GPIOEvent_t *e) {
  char  msg[255];
  /*
  int   new_value;
  new_value = gpio_get_level((gpio_num_t)e->pin->p);
  if (new_value != e->pin->v) {
    e->pin->v = new_value;
    */
  e->pin->v = gpio_get_level((gpio_num_t)e->pin->p);;  
    sprintf(msg,"%s%s%d=%d",mqtt_msg_prefix,DEFAULT_MQTT_PIN_SEPARATOR,e->pin->p,e->pin->v);
    sendMessage(msg);
//  }
}

static void IRAM_ATTR gpio_isr_handler(void* arg) {
  pin_t *p = (pin_t*)arg;
  GPIOEvent_t *e = (GPIOEvent_t*)malloc(sizeof(GPIOEvent_t));
  e->next = events;
  e->pin = p;
  e->pin->v = gpio_get_level((gpio_num_t)e->pin->p);
  events = e;
}

void reconnect() {
  // Loop until we're reconnected
  int retries;
  retries = 3;
  while ((retries > 0) && (!mqttClient.connected())) {
//  while (!mqttClient.connected()) {
    retries--;
    Serial.print("Attempting MQTT connection...");
    // Attempt to connect
    if (mqttClient.connect(mqtt_client_id)) {
      Serial.println("connected");
    } else {
      Serial.print("failed, rc=");
      Serial.print(mqttClient.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}

/*
 * Ethernet-specific stuff
 */
// WARNING: onEvent is called from a separate FreeRTOS task (thread)!
void ethernetOnEvent(arduino_event_id_t event) {
  switch (event) {
    case ARDUINO_EVENT_ETH_START:
      Serial.println("ETH Started");
      // The hostname must be set after the interface is started, but needs
      // to be set before DHCP, so set it from the event handler thread.
      ETH.setHostname(mqtt_client_id);
      break;
    case ARDUINO_EVENT_ETH_CONNECTED: Serial.println("ETH Connected"); break;
    case ARDUINO_EVENT_ETH_GOT_IP:
      Serial.println("ETH Got IP");
      Serial.println(ETH);
      eth_connected = true;
      break;
    case ARDUINO_EVENT_ETH_LOST_IP:
      Serial.println("ETH Lost IP");
      eth_connected = false;
      break;
    case ARDUINO_EVENT_ETH_DISCONNECTED:
      Serial.println("ETH Disconnected");
      eth_connected = false;
      break;
    case ARDUINO_EVENT_ETH_STOP:
      Serial.println("ETH Stopped");
      eth_connected = false;
      break;
    default: break;
  }
}

void readConfiguration() {
  Serial.println("readConfiguration() not implemented, skipping (we use defaults)");
}

void setup() {
  pin_t   *pin;
  uint8_t mac_addr[6];
  char  *mqtt_pin_message,*tmp_message;
  Serial.begin(115200);
  events = NULL;
  strcpy(mqtt_topic,DEFAULT_MQTT_TOPIC);
  strcpy(mqtt_broker,DEFAULT_MQTT_BROKER);
  strcpy(mqtt_msg_prefix,DEFAULT_MQTT_MSG_PREFIX);
  delayCountSinceLastSend = 0;
  sendCount = 0;
  mqtt_port = DEFAULT_MQTT_PORT;
  use_wifi = false;
  // read the configuration file if we can
  readConfiguration();
  if (strlen(wifi_ssid) > 0) {
    // wifi network is defined, use it
    use_wifi = true;
    strcpy(wifi_ssid,DEFAULT_WIRELESS_NETWORK_NAME);
    strcpy(wifi_password,DEFAULT_WIRELESS_NETWORK_KEY);
    Serial.print("Starting connecting WiFi, connecting to ");
    Serial.print(wifi_ssid);
#ifdef DEBUG_WIFI
    Serial.print(" / ");
    Serial.print(wifi_password);
#endif
    Serial.println("...");
    delay(10);
    WiFi.begin(wifi_ssid, wifi_password);
    while (WiFi.status() != WL_CONNECTED) {
      delay(500);
      Serial.print(".");
    }
    Serial.println("WiFi connected");
    Serial.println("IP address: ");
    Serial.println(WiFi.localIP());
    esp_read_mac(mac_addr,ESP_MAC_WIFI_STA);
    sprintf(mqtt_client_id,"%02X%02X%02X%02X%02X%02X",
            mac_addr[0],mac_addr[1],mac_addr[2],
            mac_addr[3],mac_addr[4],mac_addr[5]);
    strcpy(mqtt_msg_prefix,mqtt_client_id);
  } else {
    use_wifi = false;
    Network.onEvent(ethernetOnEvent);
    ETH.begin();
    while (eth_connected == false) {
      delay(50);
    }
    Serial.print("Using Ethernet, IP address : ");
    Serial.println(ETH);
    esp_read_mac(mac_addr,ESP_MAC_ETH);
    sprintf(mqtt_client_id,"%02X%02X%02X%02X%02X%02X",
            mac_addr[0],mac_addr[1],mac_addr[2],
            mac_addr[3],mac_addr[4],mac_addr[5]);
    strcpy(mqtt_msg_prefix,mqtt_client_id);
  }
  mqttClient.setServer(mqtt_broker,mqtt_port);

  gpio_install_isr_service(ESP_INTR_FLAG_DEFAULT);
  for (int x = 0;x < NUMBER_OF_PINS;x++) {
    pin = &(pins[x]);

    // set up the pin
    gpio_reset_pin((gpio_num_t)pin->p);
    
    //set the direction
    gpio_set_direction((gpio_num_t)pin->p,GPIO_MODE_INPUT);
    
    if ((pin->p != GPIO_NUM_34) &&
        (pin->p != GPIO_NUM_35) &&
        (pin->p != GPIO_NUM_36) &&
        (pin->p != GPIO_NUM_39)) {
      gpio_set_pull_mode((gpio_num_t)pin->p,GPIO_PULLUP_ONLY);
    }
    
    // run on any edge
    gpio_set_intr_type((gpio_num_t)pin->p,GPIO_INTR_ANYEDGE);
    
    // add the handler
    gpio_isr_handler_add((gpio_num_t)pin->p,gpio_isr_handler,pin);
  }
}

void loop() {
  Serial.println("start sending events.");

  if (events == NULL) {
//    if (sendCount > 0) {
//      delayCountSinceLastSend = 0;
//      sendCount = 0;
//    }
//    if (delayCountSinceLastSend > HALF_SECONDS_BEFORE_ETHDOWN) {
//      if ((use_wifi == false) && (eth_connected == true)) {
//        ETH.end();
//      }
//    } else {
//      delayCountSinceLastSend++;
//    }
    delay(500);         // no events, let's just pause
  } else {
    if ((use_wifi == false) && (eth_connected == false)) {
      ETH.begin();
      while (eth_connected == false) {
        delay(50);
      }
    }
    while (events != NULL) {
      GPIOEvent_t *event = events;
      events = events->next;
      processEvent(event);
      free(event);
//      sendCount++;
    }
  }
  mqttClient.loop();
  // at some point, perhaps an esp_restart() might be needed.
}

I have NOT modified the code from the network.  I did not add WIRELESS information into those, but if I wanted to use wireless, I could.  I'm running this on a POE network, so I might as well use that.

Essentially, the basic code starts up the ethernet (it would use wireless if I defined those fields), and then uses the MAC address (I'm refusing to override it) as it's client name.

Additionally, there is no configuration for a GPIO pin other than "listening on this pin".  With these Olimex POE devices on the ESP32-WROOM-POE, pins 34, 36, and 39 are all needing to use an external pin grounding (e.g. add a resistor to the 3v pin) if you are going to use them.  Aside from that, it's plug-and-play.  Change the hostname, compile the code, and upload it to your devices.

Then, deploy a Raspberry Pi with mosquitto (the Linux MQTT server, and install the development libraries, too) with the same hostname you threw into your ESP32 code.  That likely needs a static IP address (either configure the device with a static, or configure your DHCP server to always assign the same IP address).

With that, you simply need the two following services.

• GPIO translator - this is a tool that subscribes to a channel of your choosing, and if something matches, it converts that into a message for another channel.  For example, A3C1FFD901-PIN33=1 would get translated to "gate=OPEN".  There was a very deep fear that I had to keep track of states for more complex things like a garage door where you had an OPENING, OPEN, CLOSING, and CLOSED states with two sensors, before I realized I didn't care, mapping the states would already take care of that.  I DID need to swallow duplicate events because the ESP32 would send the same message a few times (you COULD bypass that using a debounce function, but why when I could potentially swallow those duplicates here?).  Turns out to function pretty great.  Code is below.
  
• Nagios-to-CMD file - this was fairly simple in comparison to the last one.  All it did was listen on an MQTT channel, and parse a message, then dump it into the Nagios CMD file used for passive or external checks.

So, you have to make sure Nagios is set up to process external checks, define checks with no actual check  (set passive_checks_enabled to 1 in the service definition for your sensor), set command_file (it might already be set) and check_external_commands to a "1" in your global Nagios configuration file, and restart Nagios. Take note of your host name inside of your Nagios config for these services - you'll need it in one of the configs later.  Also, make sure the directory and file specified in command_file is writeable by the user you will be running these things as.

GPIO message translator

First, let's look at our MQTT GPIO message translator.  This is yet another hack-and-slash, I mean, copy-and-paste, chunk of code.  The code :

/**
  * @file sentinel_translator.c
  *
  * @brief Mqtt client used to translate ESP32 MQTT messages to other messages
  * 
  * @date 19-Dec-2024
  * @copyright GNU General Public License v3
  *
  */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdarg.h>
#include <string.h>
#include <semaphore.h>
#include <stdint.h>
#include <time.h>
#include <pthread.h>
#include <errno.h>
#include <stdbool.h>
#include <syslog.h>

#include <mosquitto.h>

#define	MAXIMUM_WORKING_QUEUE_LENGTH	128

#define	DEFAULT_MQTT_PIN_SEPARATOR	"-"
#define	DEFAULT_MQTT_INPUT_TOPIC	"GPIO"
#define	DEFAULT_MQTT_OUTPUT_TOPIC	"SENSORS"
#define	DEFAULT_MQTT_SERVER		"localhost"
#define	DEFAULT_MQTT_PORT		1883
#define	DEFAULT_CONFIG_FILE		"/etc/sentinel/translator.conf"
#define	DEFAULT_SYSLOG_IDENTITY		""
#define	DEFAULT_SYSLOG_FACILITY		1
#define	DEFAULT_SYSLOG_OPTIONS		LOG_NOWAIT|LOG_PID
#define	MAX_LENGTH			255
#define	__SHOW_MOSQUITTO_INFO__		1
#define	DEFAULT_MQTT_CLIENT_NAME	"sentinel-translator"

char	config_file[MAX_LENGTH];
char	mqtt_input_topic[MAX_LENGTH];
char	mqtt_output_topic[MAX_LENGTH];
char	mqtt_server[MAX_LENGTH];
int	mqtt_port;
char	mqtt_pin_separator[MAX_LENGTH];
int	log_level;
char	*program_name;
char	syslog_identity[MAX_LENGTH];
int	syslog_facility;
bool	syslog_started;
int	syslog_options;
bool	daemonize;
char	*last_message_sent;
struct mosquitto	*mosq = NULL;

typedef struct gpio_rule_t gpio_rule_t;
struct gpio_rule_t {
	char			*mac;
	char			*state;
	char			*translation;
	gpio_rule_t		*next;
};
typedef struct working_queue working_queue_t;
struct working_queue {
	// Maximum length of the queue in number of entries
	unsigned int max_queue_size;
	// Pthread mutex for controlling the queue access
	pthread_mutex_t access;
	// Conditional variable queue is empty. Informs the waiting thread
	// that the queue is empty.
	pthread_cond_t empty;
	// Conditional variable queue is not full. Informs the waiting thread
	// that new entries can be written in the queue
	pthread_cond_t not_full;
	// Conditional variable queue is not empty. Informs the waiting
	// thread that there are some entries in the queue
	pthread_cond_t not_empty;
	// Head of the FIFO queue. Worker thread process the head entry first
	int head;
	// Tail of the FIFO queue. New entry is appended at the tail of the
	// queue.
	int tail;
	// Current number of entries in the queue
	int number_of_entries;
	// Pointer to memory reserved for the queue
	void *entry;
	// Size of one queue entry in bites
	int entry_size;
};
typedef int (*do_work_f)(void *work_entry);
typedef struct worker worker_t;
struct worker {
	// FIFO queue for storing the received payloads from MQTT messages
	working_queue_t working_queue;
	// Worker will run in this thread
	pthread_t working_thread;
	// Stop the processing of the working queue items
	bool stop_working;
	// Function that will be called for each queue entry. MQTT payload
	// processor.
	do_work_f do_work;
};
void add_work_entry(working_queue_t *working_queue, void *working_entry);
void logMessage(int logLevel,char *msg,...);
static void *worker_thread(void *worker_thread_arguments);
void worker_clean_up(worker_t **worker);
working_queue_t *mqtt_message_queue;

gpio_rule_t		*rules;

/**
 * @brief This function will be called by the worker thread for
 * processing each entry from the queue of MQTT message payloads.
 * 
 * @param[in] process this entry from the queue
 */

void send_translation(char *msg) {
	logMessage(LOG_INFO,"sending translation to MQTT topic %s: %s",mqtt_output_topic,msg);
	mosquitto_publish(mosq, NULL, mqtt_output_topic, strlen(msg), msg, 2, false);
};

int isLastMessage(char*msg) {
	// if it's the same last message, do not send
	if ((last_message_sent != NULL) && (strcmp(last_message_sent,msg) == 0)) {
		logMessage(LOG_INFO,"swallowing a duplicate message (%s) as we've had multiple",msg);
		return 0;
	}

	// if we've already done a check, clean up
	if (last_message_sent != NULL) free(last_message_sent);

	// store this one
	last_message_sent = (char*)malloc(strlen(msg)+1);
	strcpy(last_message_sent,msg);
	return -1;
};

int process_message(void *message) {
	time_t		local_time;
	int		rules_found;
	gpio_rule_t	*rule;
//	struct tm	tm_result;
//	char		time_stamp[32];

	//Build the timestamp header
//	local_time = time(NULL);
//	localtime_r(&local_time, &tm_result);
//	strftime(time_stamp, sizeof(time_stamp), "%d.%h.%Y %H:%M:%S", &tm_result);

	char *e = (char *) message;
	rule = rules;
	rules_found = 0;
	while (rule != NULL) {
		if ((strcmp(e,rule->state) == 0) && (isLastMessage(message) != 0)) {
			logMessage(LOG_DEBUG,"process_message(): %s matched to '%s'",message,rule->translation);
			send_translation(rule->translation);
			rules_found++;
		};
		rule = rule->next;
	};
	if (rules_found == 0) {
		logMessage(LOG_ERR,"ERROR: Unknown message payload (%s), missing configuration item?",message);
	};

	fflush(stdout);
	
	return 0;
}


/**
 * @brief Call back function for received MQTT message.
 * 
 * Libmosquitto thread will call this function for every received MQTT message.
 * It extracts the payload info from the MQTT message and writes it into the
 * working FIFO queue.
 * 
 * @param[in] pointer to libmoquitto MQTT client instance
 * @param[in,out] pointer to the data defined by the Libmosquitto user/caller
 * @param[in] points to the received MQTT message 
 */
void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
	char		*e;
	char		msg[255];
	e = (char*)malloc(strlen((char*)message->payload)+1);
	strcpy(e,message->payload);
	add_work_entry(mqtt_message_queue, message->payload);
}

static void clean_up_libmosquitto(struct mosquitto *mosq) {
	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();
}

int create_worker(worker_t **worker, unsigned int working_queue_size, unsigned int working_queue_entry_size, do_work_f do_work) {
	if (*worker) {
		//Worker already created/exists.
		return -1;
	}
	pthread_attr_t attr_thread;
	int rc;
	*worker = malloc(sizeof(worker_t));
	(*worker)->working_queue.head = 0;
	(*worker)->working_queue.number_of_entries = 0;
	(*worker)->working_queue.tail = 0;
	(*worker)->working_queue.entry_size = working_queue_entry_size;
	(*worker)->working_queue.max_queue_size = working_queue_size > MAXIMUM_WORKING_QUEUE_LENGTH ? MAXIMUM_WORKING_QUEUE_LENGTH : working_queue_size;
	(*worker)->working_queue.entry = calloc((*worker)->working_queue.max_queue_size, working_queue_entry_size);
	(*worker)->stop_working = false;
	(*worker)->do_work = do_work;
	//Init the objects for synchronisation of the threds
	pthread_mutex_init(&((*worker)->working_queue.access), NULL);
	pthread_cond_init(&((*worker)->working_queue.not_full), NULL);
	pthread_cond_init(&((*worker)->working_queue.not_empty), NULL);
	pthread_cond_init(&((*worker)->working_queue.empty), NULL);
	pthread_attr_init(&attr_thread);
	pthread_attr_setdetachstate(&attr_thread, PTHREAD_CREATE_JOINABLE);
	//Inherit all of the thread scheduling properties from the calling thread.
	pthread_attr_setinheritsched(&attr_thread, PTHREAD_EXPLICIT_SCHED);
	//Start the worker thread
	rc = pthread_create(&((*worker)->working_thread), &attr_thread, worker_thread, *worker);
	if (rc != 0) {
		strerror(errno);
		worker_clean_up(worker);
		rc = -1;
	}
	return 0;
}


void add_work_entry(working_queue_t *working_queue, void *working_entry) {
	pthread_mutex_lock(&(working_queue->access));
	while((working_queue->number_of_entries) == working_queue->max_queue_size) {
		pthread_cond_wait(&(working_queue->not_full), &(working_queue->access));
	}
	// Place the new work entry on the working queue
	memcpy(working_queue->entry + working_queue->entry_size * working_queue->tail, working_entry, working_queue->entry_size);
	working_queue->tail = (working_queue->tail + 1) % working_queue->max_queue_size;
	working_queue->number_of_entries++;
	pthread_cond_signal(&(working_queue->not_empty));
	pthread_mutex_unlock(&working_queue->access);
}


void *worker_thread(void *worker_thread_arguments) {
	worker_t *worker = (worker_t *) worker_thread_arguments;
	void *working_entry = NULL;
	while(1) {
		pthread_mutex_lock(&(worker->working_queue.access));
		/* Check if the queue is empty. */
		while ((worker->working_queue.number_of_entries == 0) && (!worker->stop_working)) {
			/* Wait till the other side signals that there a new item on the queue. */
			pthread_cond_wait(&(worker->working_queue.not_empty), &(worker->working_queue.access));
		}
		if (worker->stop_working) {
			pthread_mutex_unlock(&(worker->working_queue.access));
			if (working_entry) {
				free(working_entry);
			}
			pthread_exit(NULL);
		}
		if (worker->working_queue.number_of_entries != 0) {
			working_entry = malloc(worker->working_queue.entry_size);
			memcpy(working_entry, worker->working_queue.entry + (worker->working_queue.head * worker->working_queue.entry_size), worker->working_queue.entry_size);
			worker->working_queue.head = (worker->working_queue.head + 1) % worker->working_queue.max_queue_size;
			worker->working_queue.number_of_entries--;
		}
		if (worker->working_queue.number_of_entries < worker->working_queue.max_queue_size) {
			pthread_cond_signal(&(worker->working_queue.not_full));
		}
		if (worker->working_queue.number_of_entries == 0) {
			pthread_cond_signal(&(worker->working_queue.empty));
		}
		pthread_mutex_unlock(&(worker->working_queue.access));
		//Process the entry from the working queue
		if (working_entry) {
			worker->do_work(working_entry);
			free(working_entry);
			working_entry = NULL;
		}
	}
	pthread_exit((void *) 0);
}


int stop_worker(worker_t *worker) {
	pthread_mutex_lock(&worker->working_queue.access);
	if (worker->stop_working) {
		pthread_mutex_unlock(&worker->working_queue.access);
		return 0;
	}
	while(worker->working_queue.number_of_entries != 0) {
		pthread_cond_wait(&worker->working_queue.empty, &worker->working_queue.access);
	}
	//The queue is empty. Set the stop_working flag and release the mutex.
	worker->stop_working = true;
	pthread_mutex_unlock(&worker->working_queue.access);
	//Wake up the worker thread that waits for non-empty queue so it can check the stop_working parameter.
	pthread_cond_broadcast(&worker->working_queue.not_empty);
	//Wait for the worker thread to make an exit
	pthread_join(worker->working_thread, NULL);
	return 0;
}


void worker_clean_up(worker_t **worker) {
	if (*worker) {
		pthread_cond_destroy(&((*worker)->working_queue.not_full));
		pthread_cond_destroy(&((*worker)->working_queue.not_empty));
		pthread_mutex_destroy(&((*worker)->working_queue.access));
		free((*worker)->working_queue.entry);
		free(*worker);
		*worker = NULL;
	}
}

void startLog() {
	syslog_started = true;
	openlog(syslog_identity,syslog_options,syslog_facility);
};
void logMessage(int logLevel,char *msg,...) {
	va_list		ap;
	char		complete_message[MAX_LENGTH];

	if (logLevel > log_level) return;
	va_start(ap, msg);
	vsnprintf(complete_message, sizeof(complete_message)-2, msg, ap);
	va_end(ap);

	if ((daemonize == false) || (syslog_started == false)) {
		fprintf(stdout,"%s: %s\n",program_name,complete_message);
	} else {
		syslog(logLevel,complete_message);
	}
};
void endLog() {
	syslog_started = false;
	closelog();
};

void splitStringInTwo(char *input,char **output_one,char **output_two) {
	char	*tmp,*secondInput;
	*output_one = input;
	*output_two = NULL;
	if ((input[0] == '\'') || (input[0] == '\"')) (*output_one)++;
	// find the space (unless quoted)
	tmp = *output_one;
	if (*output_one == input) {
		// go until we have a space;
		while ((tmp[0] != ' ') && (tmp[0] != '\t') && (tmp[0] != '\0')) tmp++;
		*output_two = tmp;
	} else {
		while ((tmp[0] != input[0]) && (tmp[0] != '\0'))  tmp++;
		*output_two = tmp;
		// okay, we did find a closing quote of some sort, process it
		if (*output_two[0] != '\0') {
			if ((*output_two)[0] == input[0]) (*output_two)++;
		};
	}
	if (*output_two[0] != '\0') (*output_two)++;
	tmp[0] = '\0';
	// eat the space in between
	while ((*output_two[0] == '\t') || (*output_two[0] == ' ')) (*output_two)++;
	secondInput = *output_two;
	if ((secondInput[0] == '\'') || (secondInput[0] == '\"')) {
		(*output_two)++;
		tmp = *output_two;
		while (tmp[0] != secondInput[0]) tmp++;
		if (tmp[0] != '\0') tmp[0] = '\0'; // null terminate this thing
	} else {
		tmp = *output_two;
		while ((tmp[0] != '\t') && (tmp[0] != '\0') && (tmp[0] != ' ')) {
			tmp++;
		}
		if (tmp[0] != '\0') tmp[0] = '\0';
	};
};
void parseRule(char *txt_rule) {
	gpio_rule_t	*rule;
	char		*state;
	char		*translated_state;

	splitStringInTwo(txt_rule,&state,&translated_state);
	rule = (gpio_rule_t *)malloc(sizeof(gpio_rule_t));
	rule->next = rules;
	rule->state = (char*)malloc(strlen(state)+1);
	rule->translation = (char*)malloc(strlen(translated_state)+1);

	strcpy(rule->state,state);
	strcpy(rule->translation,translated_state);

	rules = rule;
};

void print_help(char *program_name,int die_when_done) {
	if (die_when_done > 0) {
		endLog();
		exit(-1);
	}
};

void makeLowerCase(char *str) {
	for (int x = 0;x < strlen(str);x++) {
                if(str[x] >= 'A') && (str[x] <= 'Z')) {
			str[x] = str[x] - 'A' + 'a';
		}
	}
}
void setConfigItem(char *option,char *value) {
	if (strlen(option) < 1) return;
	// let's make this easier on ourselves and lowercase this string
	makeLowerCase(option);
	if (strcmp(option,"log_level") == 0) {
		makeLowerCase(value);
		if (strcmp(value,"critical")) {
			log_level = LOG_CRIT;
		} else if (strcmp(value,"error")) {
			log_level = LOG_ERR;
		} else if (strcmp(value,"warning")) {
			log_level = LOG_WARNING;
		} else if (strcmp(value,"info")) {
			log_level = LOG_INFO;
		} else if (strcmp(value,"debug")) {
			log_level = LOG_DEBUG;
		}
	} else if (strcmp(option,"input_topic") == 0) {
		strcpy(mqtt_input_topic,value);
	} else if (strcmp(option,"#") == 0) {
		// this is a comment, don't do anything
	} else if (strcmp(option,"output_topic") == 0) {
		strcpy(mqtt_output_topic,value);
	} else if (strcmp(option,"server") == 0) {
		strcpy(mqtt_server,value);
	} else if (strcmp(option,"hostname") == 0) {
		strcpy(mqtt_server,value);
	} else if (strcmp(option,"port") == 0) {
		mqtt_port = atoi(value);
	} else if (strcmp(option,"rule") == 0) {
		parseRule(value);
	} else if (strcmp(option,"mqtt_pin_separator") == 0) {
		strcpy(mqtt_pin_separator,optarg);
	} else if (strcmp(option,"daemonize") == 0) {
		makeLowerCase(value);
		if ((strcmp(value,"1") == 0) || (strcmp(value,"true") == 0)) {
			daemonize = true;
		} else {
			logMessage(LOG_INFO,"daemonize set to %s, not equal to \"1\" or \"true\", assuming false",value);
			daemonize = false;
		};
	} else if (strcmp(option,"syslog_facility") == 0) {
		syslog_options = atoi(value);
	} else if (strcmp(option,"syslog_identity") == 0) {
		strcpy(syslog_identity,value);
	} else if (strcmp(option,"syslog_options") == 0) {
		syslog_options = atoi(value);
	} else {
		logMessage(LOG_CRIT,"ERROR: unknown configuration option \"%s\" encountered (%s)",option,value);
		endLog();
		exit(-1);
	}
};
void parseConfigLine(char *line) {
	char	*cmd;
	char	*linecopy;
	char	*param,*t1,*t2;
	int	pos;
	linecopy = (char*)malloc(strlen(line)+1);
	strcpy(linecopy,line);
	cmd = linecopy;
	while ((cmd[0] == ' ') || (cmd[0] == '\t')) cmd++;
	param = cmd;
	pos = 0;
	while ((param[0] != ' ') && (param[0] != '\t') && (param[0] != 0)) param++;
	if (param[0] == 0) {
		setConfigItem(cmd,param);
		free(linecopy);
		return;
	}
	param[0] = 0;
	param++;
	while ((param[0] == ' ') || (param[0] == '\t')) param++;
	if ((param != NULL) && (strlen(param) > 0)) {
		setConfigItem(cmd,param);
	} else {
		setConfigItem(cmd,NULL);
	};
	free(linecopy);
};
void read_configuration(char *filename) {
	FILE	*fp;
	char	*line;
	size_t	len;
	ssize_t	read;

	line = NULL;
	len = 0;
	logMessage(LOG_DEBUG,"reading config file %s",filename);
	fp = fopen(filename,"r");
	if (fp == NULL) {
		logMessage(LOG_CRIT,"Failed to read configuration file : %s (%s)",filename,strerror(errno));
		endLog();
		exit(-1);
	};
	while ((read = getline(&line,&len,fp)) != -1) {
		// length of line = read, actual line = line
		while ((line[strlen(line)-1] == '\n') || (line[strlen(line)-1] == '\r')) line[strlen(line)-1] = 0;
		parseConfigLine(line);
	}
	fclose(fp);
	logMessage(LOG_DEBUG,"config file %s successful",filename);
	if (line) free(line);
};

int main(int argc, char *argv[]) {
	char		ch;
	char		subscription_string[MAX_LENGTH];
	bool		config_read,config_error;

	program_name = argv[0];
	config_read = false;
	config_error = false;

	// set defaults
	strcpy(mqtt_pin_separator,DEFAULT_MQTT_PIN_SEPARATOR);
	strcpy(mqtt_input_topic,DEFAULT_MQTT_INPUT_TOPIC);
	strcpy(mqtt_output_topic,DEFAULT_MQTT_OUTPUT_TOPIC);
	strcpy(mqtt_server,DEFAULT_MQTT_SERVER);
	strcpy(config_file,DEFAULT_CONFIG_FILE);
	strcpy(syslog_identity,DEFAULT_SYSLOG_IDENTITY);
	syslog_options = DEFAULT_SYSLOG_OPTIONS;
	syslog_facility = DEFAULT_SYSLOG_FACILITY;
	mqtt_port = DEFAULT_MQTT_PORT;
	log_level = LOG_WARNING;
	daemonize = false;
	syslog_started = false;
	last_message_sent = NULL;

	logMessage(LOG_DEBUG,"starting to parse command line");
	while ((ch = getopt(argc,argv,"a:c:df:h:i:l:o:p:qr:s:t:v")) != 255) {
	switch (ch) {
	case 'a':
		setConfigItem("output_topic",optarg);
		break;
	case 'c':
		read_configuration(optarg);
		config_read = true;
		break;
	case 'd':
		setConfigItem("daemonize","1");
		break;
	case 'f':
		setConfigItem("syslog_facility",optarg);
		break;
	case 'h':
		setConfigItem("server",optarg);
		break;
	case 'i':
		setConfigItem("syslog_identity",optarg);
		break;
	case 'l':
		setConfigItem("log_level",optarg);
		break;
	case 'o':
		setConfigItem("syslog_options",optarg);
		break;
	case 'p':
		setConfigItem("port",optarg);
		break;
	case 'q':
		log_level--;
		break;
	case 'r':
		setConfigItem("rule",optarg);
		break;
	case 's':
		setConfigItem("mqtt_pin_separator",optarg);
		break;
	case 't':
		setConfigItem("input_topic",optarg);
		break;
	case 'v':
		log_level++;
		break;
	case '?':
		logMessage(LOG_CRIT,"unknown option: %c",optopt);
		print_help(argv[0],1);
	};
	};
	for (; optind < argc; optind++) {
		logMessage(LOG_WARNING,"extra arguments on the command line that were ignored: %s",argv[optind]);
		config_error = true;
	};
	logMessage(LOG_DEBUG,"finished command line parsing");
	if (config_error == true) {
		exit(-1);
	};

	if (config_read == false) {
		read_configuration(config_file);
	}
	if (daemonize == true) {
		pid_t pID = fork();
		if (pID < 0) {
			logMessage(LOG_CRIT,"ERROR: Failed to fork()");
			exit(-1);
		} else if (pID != 0) {
			// parent
			exit(0);
		}
		if (setsid() < 0) exit(-1);
		pID = fork();
		if (pID < 0) {
			logMessage(LOG_CRIT,"ERROR: Failed to fork()");
			exit(-1);
		} else if (pID != 0) {
			// parent
			exit(0);
		}

	};
	startLog();

	// libmosquitto MQTT client instance pointer
	mosq = NULL;
	// Semaphore for blocking the main thread execution.
    	sem_t blocking_sem;

	// thread for processing received payload from all publishers
	worker_t *mqtt_message_processor = NULL;
	// FIFO queue for payloads
	mqtt_message_queue = NULL;

#ifdef __SHOW_MOSQUITTO_INFO__
	int major, minor, revision;
    
	// Get libmosquitto version info
	mosquitto_lib_version(&major, &minor, &revision);
	logMessage(LOG_DEBUG,"using libmosquitto version: %d.%d.%d\n", major, minor, revision);
#endif

	// Create a working thread used by the subscriber for processing
	// the received MQTT messages. 
	if (create_worker(&mqtt_message_processor, 32, MAX_LENGTH, process_message)) {
	//if (create_worker(&mqtt_message_processor, 32, sizeof(char*), process_message)) {
		logMessage(LOG_CRIT,"failed to create mosquitto message thread");
		endLog();
		exit(-1);
	}

	// FIFO queue for the payload from MQTT messages received by the
	// subscriber. Worker thread will process the items as they land
	// in the queue.
	mqtt_message_queue = &mqtt_message_processor->working_queue;

	// libmosquitto initialization
	mosquitto_lib_init();

	// Init the semaphore
	sem_init(&blocking_sem, 0, 0);
	
	// Create new libmosquitto client instance
	mosq = mosquitto_new(DEFAULT_MQTT_CLIENT_NAME, true, mqtt_message_queue);

	if (!mosq) {
		logMessage(LOG_CRIT,"failed to create mosquitto client");
		endLog();
		exit(-1);
	}

	// Define a function which will be called by libmosquitto client
	// every time when there is a new MQTT message
	mosquitto_message_callback_set(mosq, my_message_callback);
	
	// Connect to MQTT broker
	if (mosquitto_connect(mosq, mqtt_server,
			mqtt_port, 60) != MOSQ_ERR_SUCCESS) {
		logMessage(LOG_CRIT,"Error: connecting to MQTT broker (%s:%d) failed",mqtt_server,mqtt_port);
		endLog();
		exit(-1);

		stop_worker(mqtt_message_processor);
		worker_clean_up(&mqtt_message_processor);

		clean_up_libmosquitto(mosq);

		exit(-1);
	}

	// Subscribe to any GPIO channel
	sprintf(subscription_string,"%s/+",mqtt_input_topic);
	logMessage(LOG_INFO,"subscribing to %s",mqtt_input_topic);
	mosquitto_subscribe(mosq, NULL, mqtt_input_topic, 0);

	// Run libmosquitto client in a separate thread
	mosquitto_loop_start(mosq);
	
	while(1) {
		// Block the execution of the main thread
		sem_wait(&blocking_sem);
		break;
	}

	// Stop the worker thread
	stop_worker(mqtt_message_processor);

	// Stop libmosquitto cliet thread
	mosquitto_loop_stop(mosq, true);

	// Worker clean up
	worker_clean_up(&mqtt_message_processor);

	// Clean up/destroy objects created by libmosquitto
	clean_up_libmosquitto(mosq);
	endLog();
}

The code can be compiled using :

gcc  -o sentinel_translator.o -c sentinel_translator.c
gcc -lmosquitto -lpthread -o sentinel_translator sentinel_translator.o

uses a configuration file called /etc/sentinel/translator.conf, and you will need to make sure there are "rules" defined inside.  My configuration :

input_topic	GPIO
output_topic	SENSORS
hostname	localhost
port		1883

daemonize	true

log_level	INFO

Rule A3C1FFD901-34=1 'Attic Door=OPEN'
Rule A3C1FFD901-34=0 "Attic Door=CLOSED"
Rule A3C1FFD901-33=1 'Bedroom Door=OPEN'
Rule "A3C1FFD901-33=0" "Bedroom Door=CLOSED"

# I always thought I needed something special, but the
# more I thought about it, the less we actually needed
# anything special for a four-state-sensor like garage
# doors
Rule A3C1FFD901-13=1 'Garage Door=OPENING'
Rule A3C1FFD901-13=0 'Garage Door=CLOSED'
Rule A3C1FFD901-14=1 'Garage Door=OPEN'
Rule A3C1FFD901-14=0 'Garage Door=CLOSING'

The rules will be unknown when you start.  Simply run "mosquitto_sub -t \# -v" and then physically work the sensors tied to your GPIO pins (e.g. open the door).  It will print the topic, and the "A3C1FFD901-13=1" message.  Since you know it was slapped in there when your garage door started opening,  you can map the rule appropriately.  Run through each part until you know how each one behaves, and then you can launch the program (no parameters required).

This will see the pin message from the MQTT on the GPIO channel, and write it back as the second parameter, e.g. "Garage Door=OPENING".  Note that the left side of the equal sign ("Garage Door") MUST match the service check name in Nagios for each one you configure.

Now, all we need to do is get that second parameter into Nagios.

Nagios Interface

Since Nagios can be configured to notify via e-mails, etc, I'll not worry about that here.  You can play with that on your own.  But, we DO need an MQTT subscription tool that writes these sensors into the external command file for Nagios.  The code (yes, another hack-and-slash) :

/**
  * @file sentinel_nagios_fifo.c
  *
  * @brief Mqtt client used to take MQTT messages and write them to Nagios
  * 
  * @date 19-Dec-2024
  * @copyright GNU General Public License v3
  *
  */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdarg.h>
#include <string.h>
#include <semaphore.h>
#include <stdint.h>
#include <time.h>
#include <pthread.h>
#include <errno.h>
#include <stdbool.h>
#include <syslog.h>

#include <mosquitto.h>

#define	MAXIMUM_WORKING_QUEUE_LENGTH	128

#define	DEFAULT_MQTT_PIN_SEPARATOR	"-"
#define	DEFAULT_MQTT_TOPIC		"SENSORS"
#define	DEFAULT_MQTT_SERVER		"localhost"
#define	DEFAULT_MQTT_PORT		1883
#define	DEFAULT_NAGIOS_FIFO_FILE	"/var/lib/nagios3/rw/nagios.cmd"
#define	DEFAULT_NAGIOS_CHECK_HOST	"localhost"
#define	DEFAULT_CONFIG_FILE		"/etc/sentinel/nagiosfifo.conf"
#define	DEFAULT_NAGIOS_STATE		"UNKNOWN"
#define	DEFAULT_SYSLOG_IDENTITY		""
#define	DEFAULT_SYSLOG_FACILITY		1
#define	DEFAULT_SYSLOG_OPTIONS		LOG_NOWAIT|LOG_PID
#define	MAX_LENGTH			255
//#define	__SHOW_MOSQUITTO_INFO__		1
#define	DEFAULT_MQTT_CLIENT_NAME	"sentinel-nagiosprocessor"

char	config_file[MAX_LENGTH];
char	mqtt_topic[MAX_LENGTH];
char	mqtt_server[MAX_LENGTH];
int	mqtt_port;
int	log_level;
char	*program_name;
char	nagios_service_host[MAX_LENGTH];
char	nagios_fifo_file[MAX_LENGTH];
char	syslog_identity[MAX_LENGTH];
int	syslog_facility;
bool	syslog_started;
int	syslog_options;
bool	daemonize;
char	*last_message_sent;
struct mosquitto	*mosq = NULL;

typedef struct state_t state_t;
struct state_t {
	char		*name;
	char		*nagios_state;
	state_t		*next;
};
typedef struct working_queue working_queue_t;
struct working_queue {
	// Maximum length of the queue in number of entries
	unsigned int max_queue_size;
	// Pthread mutex for controlling the queue access
	pthread_mutex_t access;
	// Conditional variable queue is empty. Informs the waiting thread
	// that the queue is empty.
	pthread_cond_t empty;
	// Conditional variable queue is not full. Informs the waiting thread
	// that new entries can be written in the queue
	pthread_cond_t not_full;
	// Conditional variable queue is not empty. Informs the waiting
	// thread that there are some entries in the queue
	pthread_cond_t not_empty;
	// Head of the FIFO queue. Worker thread process the head entry first
	int head;
	// Tail of the FIFO queue. New entry is appended at the tail of the
	// queue.
	int tail;
	// Current number of entries in the queue
	int number_of_entries;
	// Pointer to memory reserved for the queue
	void *entry;
	// Size of one queue entry in bites
	int entry_size;
};
typedef int (*do_work_f)(void *work_entry);
typedef struct worker worker_t;
struct worker {
	// FIFO queue for storing the received payloads from MQTT messages
	working_queue_t working_queue;
	// Worker will run in this thread
	pthread_t working_thread;
	// Stop the processing of the working queue items
	bool stop_working;
	// Function that will be called for each queue entry. MQTT payload
	// processor.
	do_work_f do_work;
};
void add_work_entry(working_queue_t *working_queue, void *working_entry);
void logMessage(int logLevel,char *msg,...);
static void *worker_thread(void *worker_thread_arguments);
void worker_clean_up(worker_t **worker);
working_queue_t *mqtt_message_queue;

state_t		*states;

/**
 * @brief This function will be called by the worker thread for
 * processing each entry from the queue of MQTT message payloads.
 * 
 * @param[in] process this entry from the queue
 */

void sendUpdateToNagios(char *sensor,char *stateTxt,char *nagiosState) {
	time_t		local_time;
	struct tm	tm_result;
	char		time_stamp[32];
	char		nagios_cmd[MAX_LENGTH*2];
	int		stateNum;
	FILE		*nagiosCmdFile;

	logMessage(LOG_INFO,"sensor '%s' state set to %s (Nagios: %s)",sensor,stateTxt,nagiosState);
	if (strcmp(nagiosState,"OK") == 0) {
		stateNum = 0;
	} else if (strcmp(nagiosState,"WARNING") == 0) {
		stateNum = 1;
	} else if (strcmp(nagiosState,"CRITICAL") == 0) {
		stateNum = 2;
	} else {
		stateNum = 3;
	};

	local_time = time(NULL);
	localtime_r(&local_time, &tm_result);
	strftime(time_stamp, sizeof(time_stamp), "%s", &tm_result);
	sprintf(nagios_cmd,"[%s] PROCESS_SERVICE_CHECK_RESULT;%s;%s;%d;%s - %s %s",time_stamp,nagios_service_host,sensor,stateNum,nagiosState,sensor,stateTxt);
	logMessage(LOG_DEBUG,"sending NAGIOS message \"%s\" to cmd file %s",nagios_cmd,nagios_fifo_file);
	nagiosCmdFile = fopen(nagios_fifo_file,"a");
	if (nagiosCmdFile == NULL) {
		logMessage(LOG_ERR,"ERROR: Cannot open %s: %s",nagios_fifo_file,strerror(errno));
	} else {
		fprintf(nagiosCmdFile,"%s\n",nagios_cmd);
		fclose(nagiosCmdFile);
	}
};
int process_message(void *message) {
	time_t		local_time;
	char		*msgState;
	char		*sensor;
	char		*nagiosState;
	state_t		*state;

	sensor = (char*)malloc(strlen(message)+1);
	strcpy(sensor,message);
	logMessage(LOG_DEBUG,"process_message(): got message: %s",sensor);
	msgState = sensor;
	nagiosState = NULL;
	while ((msgState[0] != '=') && (msgState[0] != '\0')) msgState++;
	if (msgState[0] == '=') {
		msgState[0] = '\0';
		msgState++;
		// now find the corresponding state string
		state = states;
		while (state != NULL) {
			if (strcmp(msgState,state->name) == 0) {
				logMessage(LOG_DEBUG,"process_message(): mapping state %s to '%s'",message,state->name);
				nagiosState = state->nagios_state;
			};
			state = state->next;
		};
		if (nagiosState == NULL) {
			logMessage(LOG_WARNING,"ERROR: state %s is NOT mapped anywhere, defaulting to UNKNOWN",msgState);
			nagiosState = DEFAULT_NAGIOS_STATE;
		};
	} else {
		nagiosState = DEFAULT_NAGIOS_STATE;
	}
	sendUpdateToNagios(sensor,msgState,nagiosState);
	free(sensor);

	fflush(stdout);
	
	return 0;
}


/**
 * @brief Call back function for received MQTT message.
 * 
 * Libmosquitto thread will call this function for every received MQTT message.
 * It extracts the payload info from the MQTT message and writes it into the
 * working FIFO queue.
 * 
 * @param[in] pointer to libmoquitto MQTT client instance
 * @param[in,out] pointer to the data defined by the Libmosquitto user/caller
 * @param[in] points to the received MQTT message 
 */
void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
	char		*e;
	char		msg[255];
	e = (char*)malloc(strlen((char*)message->payload)+1);
	strcpy(e,message->payload);
	add_work_entry(mqtt_message_queue, message->payload);
}

static void clean_up_libmosquitto(struct mosquitto *mosq) {
	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();
}

int create_worker(worker_t **worker, unsigned int working_queue_size, unsigned int working_queue_entry_size, do_work_f do_work) {
	if (*worker) {
		//Worker already created/exists.
		return -1;
	}
	pthread_attr_t attr_thread;
	int rc;
	*worker = malloc(sizeof(worker_t));
	(*worker)->working_queue.head = 0;
	(*worker)->working_queue.number_of_entries = 0;
	(*worker)->working_queue.tail = 0;
	(*worker)->working_queue.entry_size = working_queue_entry_size;
	(*worker)->working_queue.max_queue_size = working_queue_size > MAXIMUM_WORKING_QUEUE_LENGTH ? MAXIMUM_WORKING_QUEUE_LENGTH : working_queue_size;
	(*worker)->working_queue.entry = calloc((*worker)->working_queue.max_queue_size, working_queue_entry_size);
	(*worker)->stop_working = false;
	(*worker)->do_work = do_work;
	//Init the objects for synchronisation of the threds
	pthread_mutex_init(&((*worker)->working_queue.access), NULL);
	pthread_cond_init(&((*worker)->working_queue.not_full), NULL);
	pthread_cond_init(&((*worker)->working_queue.not_empty), NULL);
	pthread_cond_init(&((*worker)->working_queue.empty), NULL);
	pthread_attr_init(&attr_thread);
	pthread_attr_setdetachstate(&attr_thread, PTHREAD_CREATE_JOINABLE);
	//Inherit all of the thread scheduling properties from the calling thread.
	pthread_attr_setinheritsched(&attr_thread, PTHREAD_EXPLICIT_SCHED);
	//Start the worker thread
	rc = pthread_create(&((*worker)->working_thread), &attr_thread, worker_thread, *worker);
	if (rc != 0) {
		strerror(errno);
		worker_clean_up(worker);
		rc = -1;
	}
	return 0;
}


void add_work_entry(working_queue_t *working_queue, void *working_entry) {
	pthread_mutex_lock(&(working_queue->access));
	while((working_queue->number_of_entries) == working_queue->max_queue_size) {
		pthread_cond_wait(&(working_queue->not_full), &(working_queue->access));
	}
	// Place the new work entry on the working queue
	memcpy(working_queue->entry + working_queue->entry_size * working_queue->tail, working_entry, working_queue->entry_size);
	working_queue->tail = (working_queue->tail + 1) % working_queue->max_queue_size;
	working_queue->number_of_entries++;
	pthread_cond_signal(&(working_queue->not_empty));
	pthread_mutex_unlock(&working_queue->access);
}


void *worker_thread(void *worker_thread_arguments) {
	worker_t *worker = (worker_t *) worker_thread_arguments;
	void *working_entry = NULL;
	while(1) {
		pthread_mutex_lock(&(worker->working_queue.access));
		/* Check if the queue is empty. */
		while ((worker->working_queue.number_of_entries == 0) && (!worker->stop_working)) {
			/* Wait till the other side signals that there a new item on the queue. */
			pthread_cond_wait(&(worker->working_queue.not_empty), &(worker->working_queue.access));
		}
		if (worker->stop_working) {
			pthread_mutex_unlock(&(worker->working_queue.access));
			if (working_entry) {
				free(working_entry);
			}
			pthread_exit(NULL);
		}
		if (worker->working_queue.number_of_entries != 0) {
			working_entry = malloc(worker->working_queue.entry_size);
			memcpy(working_entry, worker->working_queue.entry + (worker->working_queue.head * worker->working_queue.entry_size), worker->working_queue.entry_size);
			worker->working_queue.head = (worker->working_queue.head + 1) % worker->working_queue.max_queue_size;
			worker->working_queue.number_of_entries--;
		}
		if (worker->working_queue.number_of_entries < worker->working_queue.max_queue_size) {
			pthread_cond_signal(&(worker->working_queue.not_full));
		}
		if (worker->working_queue.number_of_entries == 0) {
			pthread_cond_signal(&(worker->working_queue.empty));
		}
		pthread_mutex_unlock(&(worker->working_queue.access));
		//Process the entry from the working queue
		if (working_entry) {
			worker->do_work(working_entry);
			free(working_entry);
			working_entry = NULL;
		}
	}
	pthread_exit((void *) 0);
}


int stop_worker(worker_t *worker) {
	pthread_mutex_lock(&worker->working_queue.access);
	if (worker->stop_working) {
		pthread_mutex_unlock(&worker->working_queue.access);
		return 0;
	}
	while(worker->working_queue.number_of_entries != 0) {
		pthread_cond_wait(&worker->working_queue.empty, &worker->working_queue.access);
	}
	//The queue is empty. Set the stop_working flag and release the mutex.
	worker->stop_working = true;
	pthread_mutex_unlock(&worker->working_queue.access);
	//Wake up the worker thread that waits for non-empty queue so it can check the stop_working parameter.
	pthread_cond_broadcast(&worker->working_queue.not_empty);
	//Wait for the worker thread to make an exit
	pthread_join(worker->working_thread, NULL);
	return 0;
}


void worker_clean_up(worker_t **worker) {
	if (*worker) {
		pthread_cond_destroy(&((*worker)->working_queue.not_full));
		pthread_cond_destroy(&((*worker)->working_queue.not_empty));
		pthread_mutex_destroy(&((*worker)->working_queue.access));
		free((*worker)->working_queue.entry);
		free(*worker);
		*worker = NULL;
	}
}

void startLog() {
	syslog_started = true;
	openlog(syslog_identity,syslog_options,syslog_facility);
};
void logMessage(int logLevel,char *msg,...) {
	va_list		ap;
	char		complete_message[MAX_LENGTH];

	if (logLevel > log_level) return;
	va_start(ap, msg);
	vsnprintf(complete_message, sizeof(complete_message)-2, msg, ap);
	va_end(ap);

	if ((daemonize == false) || (syslog_started == false)) {
		fprintf(stdout,"%s: %s\n",program_name,complete_message);
	} else {
		syslog(logLevel,complete_message);
	}
};
void endLog() {
	syslog_started = false;
	closelog();
};

void splitStringInTwo(char *input,char **output_one,char **output_two) {
	char	*tmp,*secondInput;
	*output_one = input;
	*output_two = NULL;
	if ((input[0] == '\'') || (input[0] == '\"')) (*output_one)++;
	// find the space (unless quoted)
	tmp = *output_one;
	if (*output_one == input) {
		// go until we have a space;
		while ((tmp[0] != ' ') && (tmp[0] != '\t') && (tmp[0] != '\0')) tmp++;
		*output_two = tmp;
	} else {
		while ((tmp[0] != input[0]) && (tmp[0] != '\0'))  tmp++;
		*output_two = tmp;
		// okay, we did find a closing quote of some sort, process it
		if (*output_two[0] != '\0') {
			if ((*output_two)[0] == input[0]) (*output_two)++;
		};
	}
	if (*output_two[0] != '\0') (*output_two)++;
	tmp[0] = '\0';
	// eat the space in between
	while ((*output_two[0] == '\t') || (*output_two[0] == ' ')) (*output_two)++;
	secondInput = *output_two;
	if ((secondInput[0] == '\'') || (secondInput[0] == '\"')) {
		(*output_two)++;
		tmp = *output_two;
		while (tmp[0] != secondInput[0]) tmp++;
		if (tmp[0] != '\0') tmp[0] = '\0'; // null terminate this thing
	} else {
		tmp = *output_two;
		while ((tmp[0] != '\t') && (tmp[0] != '\0') && (tmp[0] != ' ')) {
			tmp++;
		}
		if (tmp[0] != '\0') tmp[0] = '\0';
	};
};
void parseState(char *txt_state) {
	state_t		*state;
	char		*input_state;
	char		*output_state;

	splitStringInTwo(txt_state,&input_state,&output_state);
	state = (state_t *)malloc(sizeof(state_t));
	state->next = states;
	state->name = (char*)malloc(strlen(input_state)+1);
	state->nagios_state = (char*)malloc(strlen(output_state)+1);

	strcpy(state->name,input_state);
	strcpy(state->nagios_state,output_state);

	states = state;
};

void print_help(char *program_name,int die_when_done) {
	if (die_when_done > 0) {
		endLog();
		exit(-1);
	}
};

void makeLowerCase(char *str) {
	for (int x = 0;x<strlen(str)lx++) P
                if ((str[x] >= 'A') && (str[x] <= 'Z')) {
			str[x] = str[x] - 'A' + 'a';
		}
	}
}
void setConfigItem(char *option,char *value) {
	if (strlen(option) < 1) return;
	// let's make this easier on ourselves and lowercase this string
	makeLowerCase(option);
	if (strcmp(option,"log_level") == 0) {
		makeLowerCase(value);
		if (strcmp(value,"critical")) {
			log_level = LOG_CRIT;
		} else if (strcmp(value,"error")) {
			log_level = LOG_ERR;
		} else if (strcmp(value,"warning")) {
			log_level = LOG_WARNING;
		} else if (strcmp(value,"info")) {
			log_level = LOG_INFO;
		} else if (strcmp(value,"debug")) {
			log_level = LOG_DEBUG;
		}
	} else if (strcmp(option,"topic") == 0) {
		strcpy(mqtt_topic,value);
	} else if (strcmp(option,"#") == 0) {
		// this is a comment, don't do anything
	} else if (strcmp(option,"server") == 0) {
		strcpy(mqtt_server,value);
	} else if (strcmp(option,"hostname") == 0) {
		strcpy(mqtt_server,value);
	} else if (strcmp(option,"port") == 0) {
		mqtt_port = atoi(value);
	} else if (strcmp(option,"state") == 0) {
		parseState(value);
	} else if (strcmp(option,"daemonize") == 0) {
		makeLowerCase(value);
		if ((strcmp(value,"1") == 0) || (strcmp(value,"true") == 0)) {
			daemonize = true;
		} else {
			logMessage(LOG_INFO,"daemonize set to %s, not equal to \"1\" or \"true\", assuming false",value);
			daemonize = false;
		};
	} else if (strcmp(option,"syslog_facility") == 0) {
		syslog_options = atoi(value);
	} else if (strcmp(option,"syslog_identity") == 0) {
		strcpy(syslog_identity,value);
	} else if (strcmp(option,"syslog_options") == 0) {
		syslog_options = atoi(value);
	} else if (strcmp(option,"nagios_service_host") == 0) {
		strcpy(nagios_service_host,value);
	} else if (strcmp(option,"nagios_fifo_file") == 0) {
		strcpy(nagios_fifo_file,value);
	} else {
		logMessage(LOG_CRIT,"ERROR: unknown configuration option \"%s\" encountered (%s)",option,value);
		endLog();
		exit(-1);
	}
};
void parseConfigLine(char *line) {
	char	*cmd;
	char	*linecopy;
	char	*param,*t1,*t2;
	int	pos;
	linecopy = (char*)malloc(strlen(line)+1);
	strcpy(linecopy,line);
	cmd = linecopy;
	while ((cmd[0] == ' ') || (cmd[0] == '\t')) cmd++;
	param = cmd;
	pos = 0;
	while ((param[0] != ' ') && (param[0] != '\t') && (param[0] != 0)) param++;
	if (param[0] == 0) {
		setConfigItem(cmd,param);
		free(linecopy);
		return;
	}
	param[0] = 0;
	param++;
	while ((param[0] == ' ') || (param[0] == '\t')) param++;
	if ((param != NULL) && (strlen(param) > 0)) {
		setConfigItem(cmd,param);
	} else {
		setConfigItem(cmd,NULL);
	};
	free(linecopy);
};
void read_configuration(char *filename) {
	FILE	*fp;
	char	*line;
	size_t	len;
	ssize_t	read;

	line = NULL;
	len = 0;
	logMessage(LOG_DEBUG,"reading config file %s",filename);
	fp = fopen(filename,"r");
	if (fp == NULL) {
		logMessage(LOG_CRIT,"Failed to read configuration file : %s (%s)",filename,strerror(errno));
		endLog();
		exit(-1);
	};
	while ((read = getline(&line,&len,fp)) != -1) {
		// length of line = read, actual line = line
		while ((line[strlen(line)-1] == '\n') || (line[strlen(line)-1] == '\r')) line[strlen(line)-1] = 0;
		parseConfigLine(line);
	}
	fclose(fp);
	logMessage(LOG_DEBUG,"config file %s successful",filename);
	if (line) free(line);
};

int main(int argc, char *argv[]) {
	char		ch;
	bool		config_read,config_error;

	program_name = argv[0];
	config_read = false;
	config_error = false;

	// set defaults
	strcpy(nagios_service_host,DEFAULT_NAGIOS_CHECK_HOST);
	strcpy(nagios_fifo_file,DEFAULT_NAGIOS_FIFO_FILE);
	strcpy(mqtt_topic,DEFAULT_MQTT_TOPIC);
	strcpy(mqtt_server,DEFAULT_MQTT_SERVER);
	strcpy(config_file,DEFAULT_CONFIG_FILE);
	strcpy(syslog_identity,DEFAULT_SYSLOG_IDENTITY);
	syslog_options = DEFAULT_SYSLOG_OPTIONS;
	syslog_facility = DEFAULT_SYSLOG_FACILITY;
	mqtt_port = DEFAULT_MQTT_PORT;
	log_level = LOG_WARNING;
	daemonize = false;
	syslog_started = false;
	last_message_sent = NULL;

	logMessage(LOG_DEBUG,"starting to parse command line");
	while ((ch = getopt(argc,argv,"c:df:h:i:l:o:p:qs:t:v")) != 255) {
	switch (ch) {
	case 'c':
		read_configuration(optarg);
		config_read = true;
		break;
	case 'd':
		setConfigItem("daemonize","1");
		break;
	case 'f':
		setConfigItem("syslog_facility",optarg);
		break;
	case 'h':
		setConfigItem("server",optarg);
		break;
	case 'i':
		setConfigItem("syslog_identity",optarg);
		break;
	case 'l':
		setConfigItem("log_level",optarg);
		break;
	case 'n':
		setConfigItem("log_level",optarg);
		break;
	case 'o':
		setConfigItem("syslog_options",optarg);
		break;
	case 'p':
		setConfigItem("port",optarg);
		break;
	case 'q':
		log_level--;
		break;
	case 's':
		setConfigItem("state",optarg);
		break;
	case 't':
		setConfigItem("topic",optarg);
		break;
	case 'v':
		log_level++;
		break;
	case '?':
		logMessage(LOG_CRIT,"unknown option: %c",optopt);
		print_help(argv[0],1);
	};
	};
	for (; optind < argc; optind++) {
		logMessage(LOG_WARNING,"extra arguments on the command line that were ignored: %s",argv[optind]);
		config_error = true;
	};
	logMessage(LOG_DEBUG,"finished command line parsing");
	if (config_error == true) {
		exit(-1);
	};

	if (config_read == false) {
		read_configuration(config_file);
	}
	if (daemonize == true) {
		pid_t pID = fork();
		if (pID < 0) {
			logMessage(LOG_CRIT,"ERROR: Failed to fork()");
			exit(-1);
		} else if (pID != 0) {
			// parent
			exit(0);
		}
		if (setsid() < 0) exit(-1);
		pID = fork();
		if (pID < 0) {
			logMessage(LOG_CRIT,"ERROR: Failed to fork()");
			exit(-1);
		} else if (pID != 0) {
			// parent
			exit(0);
		}

	};
	startLog();

	// libmosquitto MQTT client instance pointer
	mosq = NULL;
	// Semaphore for blocking the main thread execution.
    	sem_t blocking_sem;

	// thread for processing received payload from all publishers
	worker_t *mqtt_message_processor = NULL;
	// FIFO queue for payloads
	mqtt_message_queue = NULL;

#ifdef __SHOW_MOSQUITTO_INFO__
	int major, minor, revision;
    
	// Get libmosquitto version info
	mosquitto_lib_version(&major, &minor, &revision);
	logMessage(LOG_DEBUG,"using libmosquitto version: %d.%d.%d\n", major, minor, revision);
#endif

	// Create a working thread used by the subscriber for processing
	// the received MQTT messages. 
	if (create_worker(&mqtt_message_processor, 32, MAX_LENGTH, process_message)) {
	//if (create_worker(&mqtt_message_processor, 32, sizeof(char*), process_message)) {
		logMessage(LOG_CRIT,"failed to create mosquitto message thread");
		endLog();
		exit(-1);
	}

	// FIFO queue for the payload from MQTT messages received by the
	// subscriber. Worker thread will process the items as they land
	// in the queue.
	mqtt_message_queue = &mqtt_message_processor->working_queue;

	// libmosquitto initialization
	mosquitto_lib_init();

	// Init the semaphore
	sem_init(&blocking_sem, 0, 0);
	
	// Create new libmosquitto client instance
	mosq = mosquitto_new(DEFAULT_MQTT_CLIENT_NAME, true, mqtt_message_queue);

	if (!mosq) {
		logMessage(LOG_CRIT,"failed to create mosquitto client");
		endLog();
		exit(-1);
	}

	// Define a function which will be called by libmosquitto client
	// every time when there is a new MQTT message
	mosquitto_message_callback_set(mosq, my_message_callback);
	
	// Connect to MQTT broker
	if (mosquitto_connect(mosq, mqtt_server,
			mqtt_port, 60) != MOSQ_ERR_SUCCESS) {
		logMessage(LOG_CRIT,"Error: connecting to MQTT broker (%s:%d) failed",mqtt_server,mqtt_port);
		endLog();
		exit(-1);

		stop_worker(mqtt_message_processor);
		worker_clean_up(&mqtt_message_processor);

		clean_up_libmosquitto(mosq);

		exit(-1);
	}

	logMessage(LOG_INFO,"subscribing to %s",mqtt_topic);
	mosquitto_subscribe(mosq, NULL, mqtt_topic, 0);

	// Run libmosquitto client in a separate thread
	mosquitto_loop_start(mosq);
	
	while(1) {
		// Block the execution of the main thread
		sem_wait(&blocking_sem);
		break;
	}

	// Stop the worker thread
	stop_worker(mqtt_message_processor);

	// Stop libmosquitto cliet thread
	mosquitto_loop_stop(mosq, true);

	// Worker clean up
	worker_clean_up(&mqtt_message_processor);

	// Clean up/destroy objects created by libmosquitto
	clean_up_libmosquitto(mosq);
	endLog();
}

Compile the code using :

gcc  -o sentinel_nagios_fifo.o -c sentinel_nagios_fifo.c
gcc -lmosquitto -lpthread -o sentinel_nagios_fifo sentinel_nagios_fifo.o

Now, remember that command_file and other stuff from Nagios?  We're going to need that.  Edit your configuration file (/etc/sentinel/nagiosfifo.conf) and make the appropriate edits.

Configuration file :

topic			security
daemonize		true
nagios_fifo_file	/var/lib/nagios3/rw/nagios.cmd
log_level		WARNING

# nagios_service_host	localhost
nagios_service_host	house

state			OPEN	CRITICAL
state			OPENING	WARNING
state			CLOSING	WARNING
state			CLOSED	OK

Then, start the processes up and you have your home door sensors tied into Nagios.  Congrats!