MQTT seems to work

1. subscribe now similar to HTTP server, with lambda 2. subscribe topic syntax support all HTTP features as path arguments and wildcards 3. event received() changed to receivedUnhandled() for unhandled messages (should never be called in proper work) 4. internal logic got more complicated, several endpoints may be serviced by single MQTT topic, so nested Map used
2026-05-28 20:26:49 +03:00
parent 236896fb0f
commit 38d09e272c
3 changed files with 261 additions and 56 deletions
@@ -20,6 +20,31 @@
 #include "pimqttclient.h"

 #include "MQTTClient.h"
+#include "piliterals_string.h"
+#include "piserverendpoint_p.h"
+
+
+struct PIMQTT::Client::Endpoint: public PIHTTP::ServerEndpoint {
+	PIMQTT::Client::MessageFunction function;
+};
+
+struct EndpointsStorage {
+	PIMap<uint, PIMap<PIString, PIVector<PIMQTT::Client::Endpoint>>> prepared; // [priority][topic] -> endpoints
+	PIMap<PIString, int> topics_binded;                                        // [topic] -> count
+};
+
+
+PIString topicFromEndpoint(const PIMQTT::Client::Endpoint & e) {
+	PIStringList ret;
+	for (const auto & i: e.prepared_path) {
+		switch (i.type) {
+		case PIHTTP::ServerEndpoint::PathElement::Type::Fixed: ret << i.source; break;
+		case PIHTTP::ServerEndpoint::PathElement::Type::AnyMany: ret << "#"_a; break;
+		default: ret << "+"_a; break;
+		}
+	}
+	return ret.join('/');
+}


 STATIC_INITIALIZER_BEGIN
@@ -32,6 +57,8 @@ PRIVATE_DEFINITION_START(PIMQTT::Client)
 	MQTTClient client = nullptr;
 	bool connected    = false;

+	PIProtectedVariable<EndpointsStorage> endpoints;
+
 	static void connectionLost_callback(void * context, char *) {
 		((PIMQTT::Client *)context)->mqtt_connectionLost();
 	}
@@ -88,9 +115,9 @@ void PIMQTT::Client::disconnect() {
 }


-void PIMQTT::Client::subscribe(const PIString & topic, QoS qos) {
+void PIMQTT::Client::subscribe(const PIString & topic, MessageFunction functor, QoS qos) {
 	if (is_destoying) return;
-	worker->enqueueTask([this, topic, qos] { subscribeInternal({topic, qos}); });
+	worker->enqueueTask([this, topic, functor, qos] { subscribeInternal({topic, functor, qos}); });
 }


@@ -115,8 +142,15 @@ void PIMQTT::Client::publish(const MessageConst & msg) {
 }


+void PIMQTT::Client::unsubscribeAll() {
+	unsubscribe("#");
+	unregisterAll();
+}
+
+
 void PIMQTT::Client::mqtt_connectionLost() {
 	piCoutObj << "mqtt_connectionLost";
+	unregisterAll();
 	PRIVATE->connected = false;
 	changeStatus(Idle);
 	disconnected(Error::ServerUnavailable);
@@ -126,9 +160,110 @@ void PIMQTT::Client::mqtt_connectionLost() {
 void PIMQTT::Client::mqtt_deliveryComplete(int token) {}


-void PIMQTT::Client::mqtt_messageArrived(const MessageConst & msg) {
+void PIMQTT::Client::mqtt_messageArrived(MessageMutable & msg) {
+	PIStringList in_path = msg.topicList();
+	PIMQTT::Client::MessageFunction function;
 	piCoutObj << "mqtt_messageArrived";
-	received(msg);
+	{
+		bool found = false;
+		auto ref   = PRIVATE->endpoints.getRef();
+		auto pit   = ref->prepared.makeReverseIterator();
+		while (pit.next()) { // by priority
+
+			auto tit = pit.value().makeIterator();
+			while (tit.next()) { // by MQTT topic
+
+				for (const auto & ep: tit.value()) {
+					PIMap<PIString, PIString> ext_args;
+					if (ep.match(in_path, ext_args)) {
+						msg.pathArguments() = ext_args;
+						function            = ep.function;
+						found               = true;
+						break;
+					}
+				}
+				if (found) break;
+			}
+			if (found) break;
+		}
+	}
+	if (function)
+		function(msg);
+	else
+		receivedUnhandled(msg);
+}
+
+
+PIString PIMQTT::Client::registerSubscribe(const Subscribe & sub) {
+	Endpoint ep;
+	ep.create(convertTopic2HTTP(sub.topic));
+	ep.function    = sub.functor;
+	PIString topic = topicFromEndpoint(ep);
+	if (topic.isEmpty()) {
+		piCoutObj << "Warning: subscribe to empty topic, ignore";
+		return {};
+	}
+	piCout << sub.topic << "->" << topic << ep.priority;
+	bool is_new_topic = false;
+	auto ref          = PRIVATE->endpoints.getRef();
+	auto & eps_by_topic(ref->prepared[ep.priority][topic]);
+	for (const auto & i: eps_by_topic) {
+		if (i.path == ep.path) {
+			piCoutObj << "Warning: subscribe duplicate path, ignore";
+			return {};
+		}
+	}
+	eps_by_topic << ep;
+	auto & counter(ref->topics_binded[topic]);
+	is_new_topic = counter == 0;
+	++counter;
+	if (!is_new_topic) return {};
+	return topic;
+}
+
+
+PIString PIMQTT::Client::unregisterSubscribe(const PIString & mqtt_topic) {
+	Endpoint ep;
+	ep.create(convertTopic2HTTP(mqtt_topic));
+	PIString topic = topicFromEndpoint(ep);
+	if (topic.isEmpty()) {
+		piCoutObj << "Warning: unsubscribe from empty topic, ignore";
+		return {};
+	}
+	piCout << mqtt_topic << "->" << topic << ep.priority;
+	auto ref = PRIVATE->endpoints.getRef();
+	auto pit = ref->prepared.makeIterator();
+	while (pit.next()) { // by priority
+
+		auto tit = pit.value().makeIterator();
+		while (tit.next()) { // by MQTT topic
+
+			auto & eps(tit.value());
+			for (int i = 0; i < eps.size_s(); ++i) {
+				if (eps[i].path == ep.path) {
+					eps.remove(i);
+					auto & counter(ref->topics_binded[tit.key()]);
+					--counter;
+					PIString ret;
+					if (counter <= 0) ret = tit.key();
+					if (eps.isEmpty()) {
+						piCout << "remove topics" << tit.key();
+						pit.value().remove(tit.key());
+					}
+					return ret;
+				}
+			}
+		}
+	}
+	piCoutObj << "Warning: unsubscribe from" << mqtt_topic << ", topic not found";
+	return {};
+}
+
+
+void PIMQTT::Client::unregisterAll() {
+	auto ref = PRIVATE->endpoints.getRef();
+	ref->prepared.clear();
+	ref->topics_binded.clear();
 }


@@ -189,15 +324,27 @@ void PIMQTT::Client::publishInternal(const MessageConst & m) {

 void PIMQTT::Client::subscribeInternal(const Subscribe & sub) {
 	if (!PRIVATE->client) return;
-	int ret = MQTTClient_subscribe(PRIVATE->client, sub.topic.dataUTF8(), static_cast<int>(sub.qos));
+	piCout << "";
+	piCout << "subscribeInternal" << sub.topic;
+	PIString topic = registerSubscribe(sub);
+	if (topic.isEmpty()) return;
+	piCout << "NEW" << topic; // << PRIVATE->endpoints.getRef()->size();
+	int ret = MQTTClient_subscribe(PRIVATE->client, topic.dataUTF8(), static_cast<int>(sub.qos));
 	if (ret != MQTTCLIENT_SUCCESS) {
-		piCoutObj << "Failed to subscribe" << sub.topic << ", code" << ret;
+		piCoutObj << "Failed to subscribe" << topic << ", code" << ret;
+
+		return;
 	}
 }


-void PIMQTT::Client::unsubscribeInternal(const PIString & topic) {
+void PIMQTT::Client::unsubscribeInternal(const PIString & mqtt_topic) {
 	if (!PRIVATE->client) return;
+	piCout << "";
+	piCout << "unsubscribeInternal" << mqtt_topic;
+	PIString topic = unregisterSubscribe(mqtt_topic);
+	if (topic.isEmpty()) return;
+	piCout << "DEL" << topic; // << PRIVATE->endpoints.getRef()->size();
 	int ret = MQTTClient_unsubscribe(PRIVATE->client, topic.dataUTF8());
 	if (ret != MQTTCLIENT_SUCCESS) {
 		piCoutObj << "Failed to unsubscribe" << topic << ", code" << ret;
@@ -206,6 +353,7 @@ void PIMQTT::Client::unsubscribeInternal(const PIString & topic) {


 void PIMQTT::Client::destroy() {
+	unregisterAll();
 	if (!PRIVATE->client) return;
 	if (PRIVATE->connected) MQTTClient_disconnect(PRIVATE->client, 1000);
 	MQTTClient_destroy(&PRIVATE->client);
@@ -217,3 +365,12 @@ void PIMQTT::Client::destroy() {
 void PIMQTT::Client::changeStatus(Status s) {
 	m_status = s;
 }
+
+PIString PIMQTT::Client::convertTopic2MQTT(const PIString & topic) {
+	return topic.replacedAll("**", '#').replacedAll('*', '+');
+}
+
+
+PIString PIMQTT::Client::convertTopic2HTTP(const PIString & topic) {
+	return topic.replacedAll('#', "**").replacedAll('+', '*');
+}