Added mutex in DecodeRXMessage

Fixed event system variables

Author: gizmocuz

main/EventSystem.cpp

@@ -1789,7 +1789,11 @@ bool CEventSystem::parseBlocklyActions(const std::string &Actions, const std::st
 		}
 		size_t eQPos = csubstr.find_first_of("=") + 1;
 		std::string doWhat = csubstr.substr(eQPos);
-		doWhat = doWhat.substr(1, doWhat.size() - 2);
+		if (doWhat.find('"') == 0)
+		{
+			//Strip quotes
+			doWhat = doWhat.substr(1, doWhat.size() - 2);
+		}
 
 		size_t sPos = csubstr.find_first_of("[") + 1;
 		size_t ePos = csubstr.find_first_of("]");

main/mainworker.cpp

@@ -134,6 +134,7 @@ m_LastSunriseSet("")
 {
 	m_SecCountdown=-1;
 	m_stoprequested=false;
+	m_stopRxMessageThread = false;
 	m_verboselevel=EVBL_None;
 
 	m_bStartHardware=false;
@@ -160,6 +161,8 @@ m_LastSunriseSet("")
 	m_LastUpdateCheck = 0;
 	m_bHaveUpdate = false;
 	m_iRevision = 0;
+
+	m_rxMessageIdx = 1;
 }
 
 MainWorker::~MainWorker()
@@ -884,6 +887,13 @@ bool MainWorker::Start()
 
 bool MainWorker::Stop()
 {
+	if (m_rxMessageThread) {
+		// Stop RxMessage thread before hardware to avoid NULL pointer exception
+		m_stopRxMessageThread = true;
+		UnlockRxMessageQueue();
+		m_rxMessageThread->join();
+		m_rxMessageThread.reset();
+	}
 	if (m_thread)
 	{
 		m_webservers.StopServers();
@@ -945,7 +955,9 @@ bool MainWorker::StartThread()
 	}
 
 	m_thread = boost::shared_ptr<boost::thread>(new boost::thread(boost::bind(&MainWorker::Do_Work, this)));
-	return (m_thread!=NULL);
+	m_rxMessageThread = boost::shared_ptr<boost::thread>(new boost::thread(boost::bind(&MainWorker::Do_Work_On_Rx_Messages, this)));
+
+	return (m_thread!=NULL) && (m_rxMessageThread!=NULL);
 }
 
 #define HEX( x ) \
@@ -1638,9 +1650,215 @@ unsigned long long MainWorker::PerformRealActionFromDomoticzClient(const unsigne
 	return -1;
 }
 
-void MainWorker::DecodeRXMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand) {
+void MainWorker::DecodeRXMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand)
+{
+	boost::lock_guard<boost::mutex> l(m_decodeRXMessageMutex);
 	if ((pHardware == NULL) || (pRXCommand == NULL))
 		return;
+	if ((pHardware->HwdType == HTYPE_Domoticz) && (pHardware->m_HwdID == 8765))
+	{
+		//Directly process the command
+		ProcessRXMessage(pHardware, pRXCommand);
+	}
+	else
+	{
+		ProcessRXMessage(pHardware, pRXCommand);
+		// Submit command without waiting for the command to be processed
+		//PushRxMessage(pHardware, pRXCommand);
+	}
+}
+
+void MainWorker::PushRxMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand)
+{
+	// Check command, submit it without waiting for it to be processed
+	CheckAndPushRxMessage(pHardware, pRXCommand, false);
+}
+
+void MainWorker::PushAndWaitRxMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand)
+{
+	// Check command, submit it and wait for it to be processed
+	CheckAndPushRxMessage(pHardware, pRXCommand, true);
+}
+
+void MainWorker::CheckAndPushRxMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand, const bool wait)
+{
+	if ((pHardware == NULL) || (pRXCommand == NULL)) {
+		_log.Log(LOG_ERROR, "RxQueue: cannot push message with undefined hardware (%s) or command (%s)",
+				(pHardware == NULL) ? "null" : "not null",
+				(pRXCommand == NULL) ? "null" : "not null");
+		return;
+	}
+	if (pHardware->m_HwdID < 1) {
+		_log.Log(LOG_ERROR, "RxQueue: cannot push message with invalid hardware id (id=%d, type=%d, name=%s)",
+				pHardware->m_HwdID,
+				pHardware->HwdType,
+				pHardware->Name.c_str());
+		return;
+	}
+
+	// Build queue item
+	_tRxQueueItem rxMessage;
+	rxMessage.rxMessageIdx = m_rxMessageIdx++;
+	rxMessage.hardwareId = pHardware->m_HwdID;
+	// defensive copy of the command
+	rxMessage.vrxCommand.insert(rxMessage.vrxCommand.begin(), pRXCommand, pRXCommand + pRXCommand[0] + 1);
+	rxMessage.crc = 0x0;
+#ifdef DEBUG_RXQUEUE
+		// CRC
+		boost::crc_optimal<16, 0x1021, 0xFFFF, 0, false, false> crc_ccitt2;
+		crc_ccitt2 = std::for_each(pRXCommand, pRXCommand + pRXCommand[0] + 1, crc_ccitt2);
+		rxMessage.crc = crc_ccitt2();
+#endif
+
+	if (m_stopRxMessageThread) {
+		// Server is stopping
+		return;
+	}
+
+	// Trigger
+	rxMessage.trigger = NULL; // Should be initialized to NULL if trigger is no used
+	if (wait) { // add trigger to wait for the message to be processed
+		rxMessage.trigger = new queue_element_trigger();
+	}
+
+#ifdef DEBUG_RXQUEUE
+		_log.Log(LOG_STATUS, "RxQueue: push a rxMessage(%lu) (hrdwId=%d, hrdwType=%d, hrdwName=%s, type=%02X, subtype=%02X)",
+				rxMessage.rxMessageIdx,
+				pHardware->m_HwdID,
+				pHardware->HwdType,
+				pHardware->Name.c_str(),
+				pRXCommand[1],
+				pRXCommand[2]);
+#endif
+
+	// Push item to queue
+	m_rxMessageQueue.push(rxMessage);
+
+	if (rxMessage.trigger != NULL) {
+#ifdef DEBUG_RXQUEUE
+			_log.Log(LOG_STATUS, "RxQueue: wait for rxMessage(%lu) to be processed...", rxMessage.rxMessageIdx);
+#endif
+		bool moreThanTimeout = true;
+		while(!rxMessage.trigger->timed_wait(boost::posix_time::milliseconds(1000))) {
+#ifdef DEBUG_RXQUEUE
+				_log.Log(LOG_STATUS, "RxQueue: wait 1s for rxMessage(%lu) to be processed...", rxMessage.rxMessageIdx);
+#endif
+			moreThanTimeout = true;
+			if (m_stopRxMessageThread) {
+				// Server is stopping
+				break;
+			}
+		}
+#ifdef DEBUG_RXQUEUE
+		if (moreThanTimeout) {
+			_log.Log(LOG_STATUS, "RxQueue: rxMessage(%lu) processed", rxMessage.rxMessageIdx);
+		}
+#endif
+		delete rxMessage.trigger;
+	}
+}
+
+void MainWorker::UnlockRxMessageQueue()
+{
+#ifdef DEBUG_RXQUEUE
+		_log.Log(LOG_STATUS, "RxQueue: unlock queue using dummy message");
+#endif
+	// Push dummy message to unlock queue
+	_tRxQueueItem rxMessage;
+	rxMessage.rxMessageIdx = m_rxMessageIdx++;
+	rxMessage.hardwareId = -1;
+	rxMessage.trigger = NULL;
+	m_rxMessageQueue.push(rxMessage);
+}
+
+void MainWorker::Do_Work_On_Rx_Messages()
+{
+	_log.Log(LOG_STATUS, "RxQueue: queue worker started...");
+
+	m_stopRxMessageThread = false;
+	while (true) {
+		if (m_stopRxMessageThread) {
+			// Server is stopping
+			break;
+		}
+
+		// Wait and pop next message or timeout
+		_tRxQueueItem rxQItem;
+		bool hasPopped = m_rxMessageQueue.timed_wait_and_pop<boost::posix_time::milliseconds>(rxQItem,
+				boost::posix_time::milliseconds(5000));// (if no message for 2 seconds, returns anyway to check m_stopRxMessageThread)
+
+		if (!hasPopped) {
+			// Timeout occurred : queue is empty
+#ifdef DEBUG_RXQUEUE
+				//_log.Log(LOG_STATUS, "RxQueue: the queue has been empty for five seconds");
+#endif
+			continue;
+		}
+		if (rxQItem.hardwareId == -1) {
+			// dummy message
+#ifdef DEBUG_RXQUEUE
+				_log.Log(LOG_STATUS, "RxQueue: dummy message popped");
+#endif
+			continue;
+		}
+		if (rxQItem.hardwareId < 1) {
+			_log.Log(LOG_ERROR, "RxQueue: cannot process invalid hardware id: (%d)", rxQItem.hardwareId);
+			// cannot process message with invalid id or null message
+			if (rxQItem.trigger != NULL) rxQItem.trigger->popped();
+			continue;
+		}
+
+		const CDomoticzHardwareBase *pHardware = GetHardware(rxQItem.hardwareId);
+
+		// Check pointers
+		if (pHardware == NULL) {
+			_log.Log(LOG_ERROR, "RxQueue: cannot retrieve hardware with id: %d", rxQItem.hardwareId);
+			if (rxQItem.trigger != NULL) rxQItem.trigger->popped();
+			continue;
+		}
+		if (rxQItem.vrxCommand.empty()) {
+			_log.Log(LOG_ERROR, "RxQueue: cannot retrieve command with id: %d", rxQItem.hardwareId);
+			if (rxQItem.trigger != NULL) rxQItem.trigger->popped();
+			continue;
+		}
+		
+		const unsigned char *pRXCommand = &rxQItem.vrxCommand[0];
+
+#ifdef DEBUG_RXQUEUE
+			// CRC
+			boost::uint16_t crc = rxQItem.crc;
+			boost::crc_optimal<16, 0x1021, 0xFFFF, 0, false, false> crc_ccitt2;
+			crc_ccitt2 = std::for_each(pRXCommand, pRXCommand+rxQItem.vrxCommand.size(), crc_ccitt2);
+			if (crc != crc_ccitt2()) {
+				_log.Log(LOG_ERROR, "RxQueue: cannot process invalid rxMessage(%lu) from hardware with id=%d (type %d)",
+						rxQItem.rxMessageIdx,
+						rxQItem.hardwareId,
+						pHardware->HwdType);
+				if (rxQItem.trigger != NULL) rxQItem.trigger->popped();
+				continue;
+			}
+
+			_log.Log(LOG_STATUS, "RxQueue: process a rxMessage(%lu) (hrdwId=%d, hrdwType=%d, hrdwName=%s, type=%02X, subtype=%02X)",
+					rxQItem.rxMessageIdx,
+					pHardware->m_HwdID,
+					pHardware->HwdType,
+					pHardware->Name.c_str(),
+					pRXCommand[1],
+					pRXCommand[2]);
+#endif
+
+		ProcessRXMessage(pHardware, pRXCommand);
+		if (rxQItem.trigger != NULL)
+		{
+			rxQItem.trigger->popped();
+		}
+	}
+
+	_log.Log(LOG_STATUS, "RxQueue: queue worker stopped...");
+}
+
+void MainWorker::ProcessRXMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand)
+{
 	// current date/time based on current system
 	size_t Len = pRXCommand[0] + 1;
 

main/mainworker.h

@@ -141,6 +141,8 @@ class MainWorker
 	std::map<std::string, time_t > m_componentheartbeats;
 	boost::mutex m_heartbeatmutex;
 
+	boost::mutex m_decodeRXMessageMutex;
+
 	std::vector<int> m_devicestorestart;
 
 	int m_SecCountdown;
@@ -195,6 +197,25 @@ class MainWorker
 	void decode_BateryLevel(bool bIsInPercentage, unsigned char level);
 	unsigned char get_BateryLevel(const CDomoticzHardwareBase *pHardware, bool bIsInPercentage, unsigned char level);
 
+	// RxMessage queue resources
+	volatile bool m_stopRxMessageThread;
+	volatile unsigned long m_rxMessageIdx;
+	boost::shared_ptr<boost::thread> m_rxMessageThread;
+	void Do_Work_On_Rx_Messages();
+	struct _tRxQueueItem {
+		unsigned long rxMessageIdx;
+		int hardwareId;
+		std::vector<unsigned char> vrxCommand;
+		boost::uint16_t crc;
+		queue_element_trigger* trigger;
+	};
+	concurrent_queue<_tRxQueueItem> m_rxMessageQueue;
+	void UnlockRxMessageQueue();
+	void PushRxMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand);
+	void PushAndWaitRxMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand);
+	void CheckAndPushRxMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand, const bool wait);
+	void ProcessRXMessage(const CDomoticzHardwareBase *pHardware, const unsigned char *pRXCommand);
+
 	//(RFX) Message decoders
 	unsigned long long decode_InterfaceMessage(const CDomoticzHardwareBase *pHardware, const int HwdID, const tRBUF *pResponse);
 	unsigned long long decode_InterfaceControl(const CDomoticzHardwareBase *pHardware, const int HwdID, const tRBUF *pResponse);

www/eventsframe.html

@@ -157,7 +157,7 @@
 					return '"'+encodeURIComponent($(titleB).text())+'"'; 
 				}
 				else if ($(titleB).attr("name") == "NUM") {
-					return '"'+$(titleB).text()+'"'; 
+					return $(titleB).text();
 				}
 				return "unknown comparevariable "+variableType;
 			}

www/html5.appcache

@@ -1,5 +1,5 @@
 CACHE MANIFEST
-# ref 1125
+# ref 1126
 
 CACHE:
 # CSS