added subscription for /get and /set#security

src/ebusd/mqtthandler.cpp

@@ -28,13 +28,13 @@ using namespace std;
 static const struct argp_option g_mqtt_argp_options[] = {
 	{NULL,             0, NULL,    0, "MQTT options:", 1 },
 	{"mqttport",       1, "PORT",  0, "Connect to MQTT broker on PORT, 0 to disable [0]", 0 },
-	{"mqtttopic",      2, "TOPIC", 0, "Use MQTT TOPIC (prefix if without wildcards) [ebusd]", 0 },
+	{"mqtttopic",      2, "TOPIC", 0, "Use MQTT TOPIC (prefix before /%circuit/%name or complete format) [ebusd]", 0 },
 
 	{NULL,             0,        NULL,    0, NULL, 0 },
 };
 
 static uint16_t g_port = 0; //!< optional port of MQTT broker, 0 to disable [0]
-static const char* g_topic = PACKAGE; //!< MQTT topic to use (prefix if without wildcards) [ebusd/$circuit/$name]
+static const char* g_topic = PACKAGE; //!< MQTT topic to use (prefix if without wildcards) [ebusd]
 
 
 /**
@@ -77,9 +77,9 @@ const struct argp_child* mqtthandler_getargs()
 	return &g_mqtt_argp_child;
 }
 
-DataHandler* mqtthandler_register(BusHandler* busHandler)
+DataHandler* mqtthandler_register(BusHandler* busHandler, MessageMap* messages)
 {
-	return new MqttHandler(busHandler);
+	return new MqttHandler(busHandler, messages);
 }
 
 /** the known topic column names. */
@@ -111,7 +111,7 @@ bool parseTopic(const string topic, vector<string> &strs, vector<size_t> &cols)
 	size_t lastpos = 0;
 	size_t end = topic.length();
 	vector<string> columns;
-	for (size_t pos=topic.find('$', lastpos); pos!=string::npos; ) {
+	for (size_t pos=topic.find('%', lastpos); pos!=string::npos; ) {
 		size_t col = columnCount;
 		size_t len = 0;
 		for (size_t i = 0; i < columnCount; i++) {
@@ -124,77 +124,78 @@ bool parseTopic(const string topic, vector<string> &strs, vector<size_t> &cols)
 		if (col==columnCount) {
 			return false;
 		}
-		strs.push_back(topic.substr(lastpos, pos));
+		for (vector<size_t>::iterator it=cols.begin(); it!=cols.end(); it++) {
+			if (*it==col) {
+				return false; // duplicate column
+			}
+		}
+		strs.push_back(topic.substr(lastpos, pos-lastpos));
 		cols.push_back(col);
 		lastpos = pos+1+len;
+		pos = topic.find('%', lastpos);
 	}
 	if (lastpos<end) {
-		strs.push_back(topic.substr(lastpos, end));
+		strs.push_back(topic.substr(lastpos, end-lastpos));
-	}
-	if (strs.empty()) {
-		strs.push_back("");
 	}
 	return true;
 }
 
 
-MqttHandler::MqttHandler(BusHandler* busHandler)
+MqttHandler::MqttHandler(BusHandler* busHandler, MessageMap* messages)
-	: DataSink(), DataSource(busHandler), Thread()
+	: DataSink(), DataSource(busHandler), Thread(), m_messages(messages)
 {
 	bool enabled = g_port!=0;
 	m_publishByField = false;
 	m_mosquitto = NULL;
 	if (enabled && !parseTopic(g_topic, m_topicStrs, m_topicCols)) {
 		logOtherError("mqtt", "malformed topic %s", g_topic);
-		enabled = false;
+		return;
 	}
-	if (enabled) {
+	if (m_topicCols.empty()) {
-		if (m_topicCols.empty()) {
+		if (m_topicStrs.empty()) {
-			if (m_topicStrs.empty()) {
+			m_topicStrs.push_back("");
-				m_topicStrs.push_back("");
-			} else {
-				string str = m_topicStrs[0];
-				if (str.empty() || str[str.length()-1]!='/') {
-					m_topicStrs[0] = str+"/";
-				}
-			}
-			m_topicCols.push_back(COLUMN_CIRCUIT); // circuit
-			m_topicStrs.push_back("/");
-			m_topicCols.push_back(COLUMN_NAME); // name
 		} else {
-			for (size_t i=0; i<m_topicCols.size(); i++) {
+			string str = m_topicStrs[0];
-				if (m_topicCols[i]==COLUMN_FIELDS) { // fields
+			if (str.empty() || str[str.length()-1]!='/') {
-					m_publishByField = true;
+				m_topicStrs[0] = str+"/";
-					break;
-				}
 			}
 		}
-		m_globalTopic = getTopic(NULL)+"global/";
+		m_topicCols.push_back(COLUMN_CIRCUIT); // circuit
-		m_mosquitto = NULL;
+		m_topicStrs.push_back("/");
-		if (mosquitto_lib_init()!=MOSQ_ERR_SUCCESS) {
+		m_topicCols.push_back(COLUMN_NAME); // name
-			logOtherError("mqtt", "unable to initialize");
+	} else {
-		} else {
+		for (size_t i=0; i<m_topicCols.size(); i++) {
-			string clientId = PACKAGE_STRING;
+			if (m_topicCols[i]==COLUMN_FIELDS) { // fields
-			clientId += " "+static_cast<unsigned>(getpid());
+				m_publishByField = true;
-			m_mosquitto = mosquitto_new(clientId.c_str(), NULL);
+				break;
-			if (!m_mosquitto) {
-				logOtherError("mqtt", "unable to instantiate");
 			}
 		}
-		if (m_mosquitto) {
+	}
-			/*mosquitto_log_init(m_mosquitto, MOSQ_LOG_DEBUG | MOSQ_LOG_ERR | MOSQ_LOG_WARNING
+	m_globalTopic = getTopic(NULL)+"global/";
-								  | MOSQ_LOG_NOTICE | MOSQ_LOG_INFO, MOSQ_LOG_STDERR);*/
+	m_mosquitto = NULL;
-			string willTopic = m_globalTopic+"running";
+	if (mosquitto_lib_init()!=MOSQ_ERR_SUCCESS) {
-			string willData = "false";
+		logOtherError("mqtt", "unable to initialize");
-			size_t len = willData.length();
+	} else {
-			mosquitto_will_set(m_mosquitto, true, willTopic.c_str(), (uint32_t)len, (uint8_t*)(willData.c_str()), 0, true);
+		string clientId = PACKAGE_STRING;
-			if (mosquitto_connect(m_mosquitto, "localhost", g_port, 60, true)!=MOSQ_ERR_SUCCESS) {
+		clientId += " "+static_cast<unsigned>(getpid());
-				logOtherError("mqtt", "unable to connect");
+		m_mosquitto = mosquitto_new(clientId.c_str(), this);
-				mosquitto_destroy(m_mosquitto);
+		if (!m_mosquitto) {
-				m_mosquitto = NULL;
+			logOtherError("mqtt", "unable to instantiate");
-			} else {
+		}
-				logOtherNotice("mqtt", "connection established");
+	}
-			}
+	if (m_mosquitto) {
+		/*mosquitto_log_init(m_mosquitto, MOSQ_LOG_DEBUG | MOSQ_LOG_ERR | MOSQ_LOG_WARNING
+							  | MOSQ_LOG_NOTICE | MOSQ_LOG_INFO, MOSQ_LOG_STDERR);*/
+		string willTopic = m_globalTopic+"running";
+		string willData = "false";
+		size_t len = willData.length();
+		mosquitto_will_set(m_mosquitto, true, willTopic.c_str(), (uint32_t)len, (uint8_t*)(willData.c_str()), 0, true);
+		if (mosquitto_connect(m_mosquitto, "localhost", g_port, 60, true)!=MOSQ_ERR_SUCCESS) {
+			logOtherError("mqtt", "unable to connect");
+			mosquitto_destroy(m_mosquitto);
+			m_mosquitto = NULL;
+		} else {
+			logOtherNotice("mqtt", "connection established");
 		}
 	}
 }
@@ -216,6 +217,107 @@ void MqttHandler::start()
 	}
 }
 
+void on_message(void *obj, const struct mosquitto_message *message)
+{
+	MqttHandler* handler = (MqttHandler*)obj;
+	if (!handler || !message || !handler->isRunning()) {
+		return;
+	}
+	string topic(message->topic);
+	string data(message->payloadlen>0 ? (char*)message->payload : "");
+	handler->notifyTopic(topic, data);
+}
+
+void MqttHandler::notifyTopic(string topic, string data)
+{
+	size_t pos = topic.rfind('/');
+	if (pos==string::npos) {
+		return;
+	}
+	string suffix = topic.substr(pos+1);
+	bool isWrite = false;
+	if (suffix.empty()) {
+		return;
+	}
+	string direction = suffix.substr(0, 3);
+	isWrite = direction=="set";
+	if (!isWrite && direction!="get") {
+		return;
+	}
+	suffix = suffix.substr(3); // security level
+	logOtherDebug("mqtt", "received topic %s", topic.c_str(), data.c_str());
+	string remain = topic.substr(0, pos);
+	size_t last = 0;
+	string circuit, name;
+	size_t idx;
+	for (idx=0; idx<m_topicStrs.size()+1; idx++) {
+		string field;
+		string chk;
+		if (idx<m_topicStrs.size()) {
+			chk = m_topicStrs[idx];
+			pos = remain.find(chk, last);
+			if (pos==string::npos) {
+				return;
+			}
+		} else if (idx-1<m_topicCols.size()) {
+			pos = remain.size();
+		} else if (last<remain.size()) {
+			return;
+		} else {
+			break;
+		}
+		field = remain.substr(last, pos-last);
+		last = pos+chk.size();
+		if (idx==0) {
+			if (pos>0) {
+				return;
+			}
+		} else {
+			if (field.empty()) {
+				return;
+			}
+			switch (m_topicCols[idx-1]) {
+			case COLUMN_CIRCUIT:
+				circuit = field;
+				break;
+			case COLUMN_NAME:
+				name = field;
+				break;
+			case COLUMN_FIELDS:
+				//field = field; // TODO add support for writing a single field
+				break;
+			default:
+				return;
+			}
+		}
+	}
+	if (circuit.empty() || name.empty()) {
+		return;
+	}
+	logOtherInfo("mqtt", "received topic for %s %s", circuit.c_str(), name.c_str());
+	if (suffix.length()>0) {
+		circuit += "#"+suffix;
+	}
+	Message* message = m_messages->find(circuit, name, isWrite);
+	if (message==NULL) {
+		message = m_messages->find(circuit, name, isWrite, true);
+	}
+	if (message==NULL) {
+		logOtherError("mqtt", "%s message %s %s not found", isWrite?"write":"read", circuit.c_str(), name.c_str());
+		return;
+	}
+	if (!message->isPassive()) {
+		result_t result = m_busHandler->readFromBus(message, data);
+		if (result!=RESULT_OK) {
+			logOtherError("mqtt", "%s %s %s: %s", isWrite?"write":"read", circuit.c_str(), name.c_str(), getResultCode(result));
+			return;
+		}
+		logOtherNotice("mqtt", "%s %s %s: %s", isWrite?"write":"read", circuit.c_str(), name.c_str(), data.c_str());
+	}
+	ostringstream ostream;
+	publishMessage(message, ostream);
+}
+
 void MqttHandler::run()
 {
 	time_t lastTaskRun, now, start, lastSignal = 0;
@@ -229,6 +331,9 @@ void MqttHandler::run()
 	publishTopic(m_globalTopic+"version", PACKAGE_STRING "." REVISION);
 	publishTopic(m_globalTopic+"running", "true");
 	publishTopic(signalTopic, "false");
+	mosquitto_message_callback_set(m_mosquitto, on_message);
+	string subTopic = getTopic(NULL)+"#";
+	mosquitto_subscribe(m_mosquitto, NULL, subTopic.c_str(), 0);
 	while (isRunning()) {
 		handleTraffic();
 		time(&now);
@@ -264,15 +369,7 @@ void MqttHandler::run()
 				updates.str("");
 				updates.clear();
 				updates << dec;
-				if (message->decodeLastData(updates)==RESULT_OK) {
+				publishMessage(message, updates);
-					if (m_publishByField) {
-						//message->hasField()
-					} else {
-						string topic = getTopic(message);
-						string data = updates.str();
-						publishTopic(topic, data);
-					}
-				}
 			}
 			m_updatedMessages.clear();
 		}
@@ -286,7 +383,7 @@ void MqttHandler::handleTraffic()
 	}
 }
 
-string MqttHandler::getTopic(Message* message)
+string MqttHandler::getTopic(Message* message, signed char fieldIndex)
 {
 	ostringstream ret;
 	for (size_t i=0; i<m_topicStrs.size(); i++) {
@@ -295,12 +392,37 @@ string MqttHandler::getTopic(Message* message)
 			break;
 		}
 		if (i<m_topicCols.size()) {
-			 message->dumpColumn(ret, m_topicCols[i], false);
+			if (m_topicCols[i]==COLUMN_FIELDS && fieldIndex>=0) {
+				ret << message->getFieldName(fieldIndex);
+			} else {
+				message->dumpColumn(ret, m_topicCols[i]);
+			}
 		}
 	}
 	return ret.str();
 }
 
+void MqttHandler::publishMessage(Message* message, ostringstream& updates)
+{
+	result_t result = message->decodeLastData(updates);
+	if (result!=RESULT_OK) {
+		logOtherError("mqtt", "decode %s %s: %s", message->getCircuit().c_str(), message->getName().c_str(), getResultCode(result));
+		return;
+	}
+	if (m_publishByField) {
+		signed char index = 0;
+		istringstream input(updates.str());
+		string token;
+		while (getline(input, token, UI_FIELD_SEPARATOR)) {
+			string topic = getTopic(message, index);
+			publishTopic(topic, token);
+			index++;
+		}
+	} else {
+		publishTopic(getTopic(message), updates.str());
+	}
+}
+
 void MqttHandler::publishTopic(string topic, string data, bool retain)
 {
 	logOtherDebug("mqtt", "publish %s %s", topic.c_str(), data.c_str());

src/ebusd/mqtthandler.h

@@ -39,22 +39,24 @@ const struct argp_child* mqtthandler_getargs();
 /**
  * Registration function that is called once during initialization.
  * @param busHandler the @a BusHandler instance.
+ * @param messages the @a MessageMap instance.
  * @return the create @a DataHandler, or NULL on error.
  */
-DataHandler* mqtthandler_register(BusHandler* busHandler);
+DataHandler* mqtthandler_register(BusHandler* busHandler, MessageMap* messages);
 
 /**
  * The main class supporting MQTT data handling.
  */
-class MqttHandler : public DataSink, DataSource, Thread
+class MqttHandler : public DataSink, public DataSource, public Thread
 {
 public:
 
 	/**
 	 * Constructor.
 	 * @param busHandler the @a BusHandler instance.
+	 * @param messages the @a MessageMap instance.
 	 */
-	MqttHandler(BusHandler* busHandler);
+	MqttHandler(BusHandler* busHandler, MessageMap* messages);
 
 	/**
 	 * Destructor.
@@ -64,6 +66,13 @@ public:
 	// @copydoc
 	virtual void start();
 
+	/**
+	 * Notify the handler of a received MQTT message.
+	 * @param topic the topic string.
+	 * @param data the data string.
+	 */
+	void notifyTopic(string topic, string data);
+
 protected:
 
 	// @copydoc
@@ -79,9 +88,17 @@ private:
 	/**
 	 * Build the MQTT topic string for the @a Message.
 	 * @param message the @a Message to build the topic string for.
+	 * @param fieldIndex the optional field index for the field column, or -1.
 	 * @return the topic string.
 	 */
-	string getTopic(Message* message);
+	string getTopic(Message* message, signed char fieldIndex=-1);
+
+	/**
+	 * Prepare a @a Message and publish as topic.
+	 * @param message the @a Message to publish.
+	 * @param updates the @a ostringstream for preparation.
+	 */
+	void publishMessage(Message* message, ostringstream& updates);
 
 	/**
 	 * Publish a topic update to MQTT.
@@ -91,6 +108,9 @@ private:
 	 */
 	void publishTopic(string topic, string data, bool retain=true);
 
+	/** the @a MessageMap instance. */
+	MessageMap* m_messages;
+
 	/** the MQTT topic string parts. */
 	vector<string> m_topicStrs;