thread-safe client

Reviewers: ivica

Reviewed By: ivica

Subscribers: korina

Differential Revision: https://repo.mireo.local/D26864

include/async_mqtt5/detail/control_packet.hpp

@@ -1,7 +1,6 @@
 #ifndef ASYNC_MQTT5_CONTROL_PACKET_HPP
 #define ASYNC_MQTT5_CONTROL_PACKET_HPP
 
-#include <mutex>
 #include <vector>
 
 #include <boost/smart_ptr/allocate_unique.hpp>
@@ -115,7 +114,6 @@ class packet_id_allocator {
 		{}
 	};
 
-	std::mutex _mtx;
 	std::vector<interval> _free_ids;
 	static constexpr uint16_t MAX_PACKET_ID = 65535;
 
@@ -125,7 +123,6 @@ public:
 	}
 
 	uint16_t allocate() {
-		std::lock_guard _(_mtx);
 		if (_free_ids.empty()) return 0;
 		auto& last = _free_ids.back();
 		if (last.start == ++last.end) {
@@ -137,7 +134,6 @@ public:
 	}
 
 	void free(uint16_t pid) {
-		std::lock_guard _(_mtx);
 		auto it = std::upper_bound(
 			_free_ids.begin(), _free_ids.end(), pid,
 			[](const uint16_t x, const interval& i) { return x > i.start; }

include/async_mqtt5/detail/internal_types.hpp

@@ -3,6 +3,7 @@
 
 #include <optional>
 #include <string>
+#include <shared_mutex>
 
 #include <async_mqtt5/detail/any_authenticator.hpp>
 
@@ -70,6 +71,7 @@ struct mqtt_ctx {
 	credentials creds;
 	std::optional<will> will_msg;
 	connect_props co_props;
+	std::shared_mutex ca_mtx;
 	connack_props ca_props;
 	session_state state;
 	any_authenticator authenticator;

include/async_mqtt5/impl/client_service.hpp

@@ -61,9 +61,18 @@ public:
 
 	template <typename Prop>
 	decltype(auto) connack_prop(Prop p) {
+		std::shared_lock reader_lock(_mqtt_context.ca_mtx);
 		return _mqtt_context.ca_props[p];
 	}
 
+	template <typename Prop0, typename ...Props>
+	decltype(auto) connack_props(Prop0 p0, Props ...props) {
+		std::shared_lock reader_lock(_mqtt_context.ca_mtx);
+		return std::make_tuple(
+			_mqtt_context.ca_props[p0], _mqtt_context.ca_props[props]...
+		);
+	}
+
 	void credentials(
 		std::string client_id,
 		std::string username = "", std::string password = ""
@@ -109,9 +118,18 @@ public:
 
 	template <typename Prop>
 	decltype(auto) connack_prop(Prop p) {
+		std::shared_lock reader_lock(_mqtt_context.ca_mtx);
 		return _mqtt_context.ca_props[p];
 	}
 
+	template <typename Prop0, typename ...Props>
+	decltype(auto) connack_props(Prop0 p0, Props ...props) {
+		std::shared_lock reader_lock(_mqtt_context.ca_mtx);
+		return std::make_tuple(
+			_mqtt_context.ca_props[p0], _mqtt_context.ca_props[props]...
+		);
+	}
+
 	void credentials(
 		std::string client_id,
 		std::string username = "", std::string password = ""
@@ -243,6 +261,11 @@ public:
 		return _stream_context.connack_prop(p);
 	}
 
+	template <typename Prop0, typename ...Props>
+	decltype(auto) connack_props(Prop0 p0, Props ...props) {
+		return _stream_context.connack_props(p0, props...);
+	}
+
 	void run() {
 		_stream.open();
 		_rec_channel.reset();

include/async_mqtt5/impl/connect_op.hpp

@@ -283,7 +283,11 @@ public:
 			return complete(client::error::malformed_packet);
 		const auto& [session_present, reason_code, ca_props] = *rv;
 
-		_ctx.ca_props = ca_props;
+		{
+			std::unique_lock writer_lock(_ctx.ca_mtx);
+			_ctx.ca_props = ca_props;
+		}
+
 		_ctx.state.session_present(session_present);
 
 		//  Unexpected result handling:

include/async_mqtt5/impl/disconnect_op.hpp

@@ -66,10 +66,10 @@ public:
 			static_cast<uint8_t>(_context.reason_code), _context.props
 		);
 
-		send_disconnect(std::move(disconnect));
+		asio::dispatch(asio::prepend(std::move(*this), std::move(disconnect)));
 	}
 
-	void send_disconnect(control_packet<allocator_type> disconnect) {
+	void operator()(control_packet<allocator_type> disconnect) {
 		const auto& wire_data = disconnect.wire_data();
 
 		_svc_ptr->async_send(

include/async_mqtt5/impl/publish_send_op.hpp

@@ -103,6 +103,18 @@ public:
 		if (ec)
 			return complete_post(ec);
 
+		asio::dispatch(
+			asio::prepend(
+				std::move(*this), std::move(topic),
+				std::move(payload), retain, props
+			)
+		);
+	}
+
+	void operator()(
+		std::string topic, std::string payload,
+		retain_e retain, const publish_props& props
+	) {
 		uint16_t packet_id = 0;
 		if constexpr (qos_type != qos_e::at_most_once) {
 			packet_id = _svc_ptr->allocate_pid();
@@ -334,15 +346,18 @@ private:
 		if (!is_valid_utf8_topic(topic))
 			return client::error::invalid_topic;
 
-		auto max_qos = _svc_ptr->connack_prop(prop::maximum_qos);
+		const auto& [max_qos, retain_avail, topic_alias_max] =
+			_svc_ptr->connack_props(
+				prop::maximum_qos, prop::retain_available,
+				prop::topic_alias_maximum
+			);
+
 		if (max_qos && uint8_t(qos_type) > *max_qos)
 			return client::error::qos_not_supported;
 
-		auto retain_avail = _svc_ptr->connack_prop(prop::retain_available);
 		if (retain_avail && *retain_avail == 0 && retain == retain_e::yes)
 			return client::error::retain_not_available;
 
-		auto topic_alias_max = _svc_ptr->connack_prop(prop::topic_alias_maximum);
 		auto topic_alias = props[prop::topic_alias];
 		if (
 			(!topic_alias_max || topic_alias_max && *topic_alias_max == 0) &&

include/async_mqtt5/impl/subscribe_op.hpp

@@ -66,6 +66,15 @@ public:
 		if (ec)
 			return complete_post(ec, topics.size());
 
+		asio::dispatch(
+			asio::prepend(std::move(*this), topics, props)
+		);
+	}
+
+	void operator()(
+		const std::vector<subscribe_topic>& topics,
+		const subscribe_props& props
+	) {
 		uint16_t packet_id = _svc_ptr->allocate_pid();
 		if (packet_id == 0)
 			return complete_post(client::error::pid_overrun, topics.size());

include/async_mqtt5/impl/unsubscribe_op.hpp

@@ -63,6 +63,15 @@ public:
 		if (ec)
 			return complete_post(ec, topics.size());
 
+		asio::dispatch(
+			asio::prepend(std::move(*this), topics, props)
+		);
+	}
+
+	void operator()(
+		const std::vector<std::string>& topics,
+		const unsubscribe_props& props
+	) {
 		uint16_t packet_id = _svc_ptr->allocate_pid();
 		if (packet_id == 0)
 			return complete_post(client::error::pid_overrun, topics.size());