Initiation lambdas replaced with classes with executor_type/get_executor

Summary:
related to #13, T13767
`asio::cancel_at` and `asio::cancel_after` (coming with Boost 1.86) use associated executor
from the initiation object to construct the underlying timer.

Therefore, initiation lambdas are replaced with classes with executor_type/get_executor to allow that functionality
(like Asio/Beast).

Alternative solutions:
 1) asio::bind_executor(get_executor(), initiation)
 2) async_compose

Reviewers: ivica

Reviewed By: ivica

Subscribers: iljazovic, miljen

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

include/async_mqtt5/impl/client_service.hpp

@@ -20,9 +20,11 @@
 #include <async_mqtt5/impl/async_sender.hpp>
 #include <async_mqtt5/impl/autoconnect_stream.hpp>
 #include <async_mqtt5/impl/ping_op.hpp>
+#include <async_mqtt5/impl/read_message_op.hpp>
 #include <async_mqtt5/impl/replies.hpp>
 #include <async_mqtt5/impl/sentry_op.hpp>
 
+
 namespace async_mqtt5::detail {
 
 namespace asio = boost::asio;
@@ -438,6 +440,44 @@ public:
 		return _async_sender.next_serial_num();
 	}
 
+	bool subscriptions_present() const {
+		return _stream_context.session_state().subscriptions_present();
+	}
+
+	void subscriptions_present(bool present) {
+		_stream_context.session_state().subscriptions_present(present);
+	}
+
+	void update_session_state() {
+		auto& session_state = _stream_context.session_state();
+
+		if (!session_state.session_present()) {
+			_replies.clear_pending_pubrels();
+			session_state.session_present(true);
+
+			if (session_state.subscriptions_present()) {
+				channel_store_error(client::error::session_expired);
+				session_state.subscriptions_present(false);
+			}
+		}
+
+		_cancel_ping.emit(asio::cancellation_type::total);
+	}
+
+	bool channel_store(decoders::publish_message message) {
+		auto& [topic, packet_id, flags, props, payload] = message;
+		return _rec_channel.try_send(
+			error_code {}, std::move(topic),
+			std::move(payload), std::move(props)
+		);
+	}
+
+	bool channel_store_error(error_code ec) {
+		return _rec_channel.try_send(
+			ec, std::string {}, std::string {}, publish_props {}
+		);
+	}
+
 	template <typename BufferType, typename CompletionToken>
 	decltype(auto) async_send(
 		const BufferType& buffer,
@@ -477,65 +517,37 @@ public:
 		);
 	}
 
-	bool subscriptions_present() const {
-		return _stream_context.session_state().subscriptions_present();
-	}
-
-	void subscriptions_present(bool present) {
-		_stream_context.session_state().subscriptions_present(present);
-	}
-
-	void update_session_state() {
-		auto& session_state = _stream_context.session_state();
-
-		if (!session_state.session_present()) {
-			_replies.clear_pending_pubrels();
-			session_state.session_present(true);
-
-			if (session_state.subscriptions_present()) {
-				channel_store_error(client::error::session_expired);
-				session_state.subscriptions_present(false);
-			}
-		}
-
-		_cancel_ping.emit(asio::cancellation_type::total);
-	}
-
-	bool channel_store(decoders::publish_message message) {
-		auto& [topic, packet_id, flags, props, payload] = message;
-		return _rec_channel.try_send(
-			error_code {}, std::move(topic),
-			std::move(payload), std::move(props)
-		);
-	}
-
-	bool channel_store_error(error_code ec) {
-		return _rec_channel.try_send(
-			ec, std::string {}, std::string {}, publish_props {}
-		);
-	}
-
 	template <typename CompletionToken>
 	decltype(auto) async_channel_receive(CompletionToken&& token) {
-		using Signature =
+		return _rec_channel.async_receive(std::forward<CompletionToken>(token));
-			void(error_code, std::string, std::string, publish_props);
-
-		auto initiation = [] (auto handler, self_type& self) {
-			auto ex = asio::get_associated_executor(
-				handler, self.get_executor()
-			);
-			return self._rec_channel.async_receive(
-				asio::bind_executor(ex, std::move(handler))
-			);
-		};
-
-		return asio::async_initiate<CompletionToken, Signature> (
-			initiation, token, std::ref(*this)
-		);
 	}
 
 };
 
+template <typename ClientService>
+class initiate_async_run {
+	std::shared_ptr<ClientService> _svc_ptr;
+public:
+	explicit initiate_async_run(std::shared_ptr<ClientService> svc_ptr) :
+		_svc_ptr(std::move(svc_ptr))
+	{}
+
+	using executor_type = typename ClientService::executor_type;
+	executor_type get_executor() const noexcept {
+		return _svc_ptr->get_executor();
+	}
+
+	template <typename Handler>
+	void operator()(Handler&& handler) {
+		auto ex = asio::get_associated_executor(handler, get_executor());
+
+		_svc_ptr->run(std::move(handler));
+		detail::ping_op { _svc_ptr, ex }.perform();
+		detail::read_message_op { _svc_ptr, ex }.perform();
+		detail::sentry_op { _svc_ptr, ex }.perform();
+	}
+};
+
 
 } // namespace async_mqtt5::detail
 

include/async_mqtt5/impl/disconnect_op.hpp

@@ -52,10 +52,10 @@ class disconnect_op {
 public:
 	template <typename Handler>
 	disconnect_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr,
 		DisconnectContext&& context, Handler&& handler
 	) :
-		_svc_ptr(svc_ptr), _context(std::move(context)),
+		_svc_ptr(std::move(svc_ptr)), _context(std::move(context)),
 		_handler(std::move(handler), _svc_ptr->get_executor())
 	{
 		auto slot = asio::get_associated_cancellation_slot(_handler);
@@ -189,10 +189,10 @@ class terminal_disconnect_op {
 
 public:
 	terminal_disconnect_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr,
 		Handler&& handler
 	) :
-		_svc_ptr(svc_ptr),
+		_svc_ptr(std::move(svc_ptr)),
 		_timer(new asio::steady_timer(_svc_ptr->get_executor())),
 		_handler(std::move(handler), _svc_ptr->get_executor())
 	{}
@@ -254,6 +254,34 @@ public:
 	}
 };
 
+template <typename ClientService, bool terminal>
+class initiate_async_disconnect {
+	std::shared_ptr<ClientService> _svc_ptr;
+public:
+	explicit initiate_async_disconnect(std::shared_ptr<ClientService> svc_ptr) :
+		_svc_ptr(std::move(svc_ptr))
+	{}
+
+	using executor_type = typename ClientService::executor_type;
+	executor_type get_executor() const noexcept {
+		return _svc_ptr->get_executor();
+	}
+
+	template <typename Handler>
+	void operator()(
+		Handler&& handler,
+		disconnect_rc_e rc, const disconnect_props& props
+	) {
+		auto ctx = disconnect_ctx { rc, props, terminal };
+		if constexpr (terminal)
+			terminal_disconnect_op { _svc_ptr, std::move(handler) }
+				.perform(std::move(ctx));
+		else
+			disconnect_op { _svc_ptr, std::move(ctx), std::move(handler) }
+				.perform();
+	}
+};
+
 template <typename ClientService, typename CompletionToken>
 decltype(auto) async_disconnect(
 	disconnect_rc_e reason_code, const disconnect_props& props,
@@ -261,20 +289,9 @@ decltype(auto) async_disconnect(
 	CompletionToken&& token
 ) {
 	using Signature = void (error_code);
-
-	auto initiation = [](
-		auto handler, disconnect_ctx ctx,
-		const std::shared_ptr<ClientService>& svc_ptr
-	) {
-		disconnect_op {
-			svc_ptr, std::move(ctx), std::move(handler)
-		}.perform();
-	};
-
 	return asio::async_initiate<CompletionToken, Signature>(
-		initiation, token,
+		initiate_async_disconnect<ClientService, false>(svc_ptr), token,
-		disconnect_ctx { reason_code, props, false },
+		reason_code, props
-		svc_ptr
 	);
 }
 
@@ -285,20 +302,9 @@ decltype(auto) async_terminal_disconnect(
 	CompletionToken&& token
 ) {
 	using Signature = void (error_code);
-
-	auto initiation = [](
-		auto handler, disconnect_ctx ctx,
-		const std::shared_ptr<ClientService>& svc_ptr
-	) {
-		terminal_disconnect_op {
-			svc_ptr, std::move(handler)
-		}.perform(std::move(ctx));
-	};
-
 	return asio::async_initiate<CompletionToken, Signature>(
-		initiation, token,
+		initiate_async_disconnect<ClientService, true>(svc_ptr), token,
-		disconnect_ctx { reason_code, props, true },
+		reason_code, props
-		svc_ptr
 	);
 }
 

include/async_mqtt5/impl/ping_op.hpp

@@ -44,13 +44,13 @@ private:
 
 public:
 	ping_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr,
-		const executor_type& ex
+		executor_type ex
 	) :
-		_svc_ptr(svc_ptr), _executor(ex),
+		_svc_ptr(std::move(svc_ptr)), _executor(ex),
 		_ping_timer(new asio::steady_timer(_svc_ptr->get_executor())),
 		_cancellation_state(
-			svc_ptr->_cancel_ping.slot(),
+			_svc_ptr->_cancel_ping.slot(),
 			asio::enable_total_cancellation {},
 			asio::enable_total_cancellation {}
 		)

include/async_mqtt5/impl/publish_rec_op.hpp

@@ -43,8 +43,8 @@ class publish_rec_op {
 	decoders::publish_message _message;
 
 public:
-	explicit publish_rec_op(const std::shared_ptr<client_service>& svc_ptr) :
+	explicit publish_rec_op(std::shared_ptr<client_service> svc_ptr) :
-		_svc_ptr(svc_ptr)
+		_svc_ptr(std::move(svc_ptr))
 	{}
 
 	publish_rec_op(publish_rec_op&&) noexcept = default;

include/async_mqtt5/impl/publish_send_op.hpp

@@ -75,10 +75,10 @@ class publish_send_op {
 
 public:
 	publish_send_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr,
 		Handler&& handler
 	) :
-		_svc_ptr(svc_ptr),
+		_svc_ptr(std::move(svc_ptr)),
 		_handler(std::move(handler), _svc_ptr->get_executor())
 	{
 		auto slot = asio::get_associated_cancellation_slot(_handler);
@@ -469,6 +469,32 @@ private:
 	}
 };
 
+template <typename ClientService, qos_e qos_type>
+class initiate_async_publish {
+	std::shared_ptr<ClientService> _svc_ptr;
+public:
+	explicit initiate_async_publish(std::shared_ptr<ClientService> svc_ptr) :
+		_svc_ptr(std::move(svc_ptr))
+	{}
+
+	using executor_type = typename ClientService::executor_type;
+	executor_type get_executor() const noexcept {
+		return _svc_ptr->get_executor();
+	}
+
+	template <typename Handler>
+	void operator()(
+		Handler&& handler,
+		std::string topic, std::string payload,
+		retain_e retain, const publish_props& props
+	) {
+		detail::publish_send_op<ClientService, Handler, qos_type> {
+			_svc_ptr, std::move(handler) 
+		}.perform(
+			std::move(topic), std::move(payload), retain, props
+		);
+	}
+};
 
 } // end namespace async_mqtt5::detail
 

include/async_mqtt5/impl/re_auth_op.hpp

@@ -35,8 +35,8 @@ class re_auth_op {
 	any_authenticator& _auth;
 
 public:
-	explicit re_auth_op(const std::shared_ptr<client_service>& svc_ptr) :
+	explicit re_auth_op(std::shared_ptr<client_service> svc_ptr) :
-		_svc_ptr(svc_ptr),
+		_svc_ptr(std::move(svc_ptr)),
 		_auth(_svc_ptr->_stream_context.mqtt_context().authenticator)
 	{}
 

include/async_mqtt5/impl/read_message_op.hpp

@@ -41,10 +41,10 @@ private:
 	executor_type _executor;
 public:
 	read_message_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr,
-		const executor_type& ex
+		executor_type ex
 	) :
-		_svc_ptr(svc_ptr), _executor(ex)
+		_svc_ptr(std::move(svc_ptr)), _executor(ex)
 	{}
 
 	read_message_op(read_message_op&&) noexcept = default;

include/async_mqtt5/impl/read_op.hpp

@@ -35,8 +35,7 @@ class read_op {
 
 public:
 	read_op(Owner& owner, Handler&& handler) :
-		_owner(owner),
+		_owner(owner), _handler(std::move(handler))
-		_handler(std::move(handler))
 	{}
 
 	read_op(read_op&&) = default;

include/async_mqtt5/impl/reconnect_op.hpp

@@ -72,8 +72,7 @@ class reconnect_op {
 public:
 	template <typename Handler>
 	reconnect_op(Owner& owner, Handler&& handler) :
-		_owner(owner),
+		_owner(owner), _handler(std::move(handler))
-		_handler(std::forward<Handler>(handler))
 	{}
 
 	reconnect_op(reconnect_op&&) = default;

include/async_mqtt5/impl/replies.hpp

@@ -93,7 +93,7 @@ private:
 
 public:
 	template <typename Executor>
-	explicit replies(const Executor& ex) : _ex(ex) {}
+	explicit replies(Executor ex) : _ex(ex) {}
 
 	replies(replies&&) = default;
 	replies(const replies&) = delete;

include/async_mqtt5/impl/sentry_op.hpp

@@ -43,10 +43,9 @@ private:
 
 public:
 	sentry_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr, executor_type ex
-		const executor_type& ex
 	) :
-		_svc_ptr(svc_ptr), _executor(ex),
+		_svc_ptr(std::move(svc_ptr)), _executor(ex),
 		_sentry_timer(new asio::steady_timer(_svc_ptr->get_executor()))
 	{}
 

include/async_mqtt5/impl/subscribe_op.hpp

@@ -52,10 +52,10 @@ class subscribe_op {
 
 public:
 	subscribe_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr,
 		Handler&& handler
 	) :
-		_svc_ptr(svc_ptr),
+		_svc_ptr(std::move(svc_ptr)),
 		_handler(std::move(handler), _svc_ptr->get_executor())
 	{
 		auto slot = asio::get_associated_cancellation_slot(_handler);
@@ -302,6 +302,28 @@ private:
 	}
 };
 
+template <typename ClientService>
+class initiate_async_subscribe {
+	std::shared_ptr<ClientService> _svc_ptr;
+public:
+	explicit initiate_async_subscribe(std::shared_ptr<ClientService> svc_ptr) :
+		_svc_ptr(std::move(svc_ptr))
+	{}
+
+	using executor_type = typename ClientService::executor_type;
+	executor_type get_executor() const noexcept {
+		return _svc_ptr->get_executor();
+	}
+
+	template <typename Handler>
+	void operator()(
+		Handler&& handler,
+		const std::vector<subscribe_topic>& topics, const subscribe_props& props
+	) {
+		detail::subscribe_op { _svc_ptr, std::move(handler) }
+			.perform(topics, props);
+	}
+};
 
 } // end namespace async_mqtt5::detail
 

include/async_mqtt5/impl/unsubscribe_op.hpp

@@ -46,10 +46,10 @@ class unsubscribe_op {
 
 public:
 	unsubscribe_op(
-		const std::shared_ptr<client_service>& svc_ptr,
+		std::shared_ptr<client_service> svc_ptr,
 		Handler&& handler
 	) :
-		_svc_ptr(svc_ptr),
+		_svc_ptr(std::move(svc_ptr)),
 		_handler(std::move(handler), _svc_ptr->get_executor())
 	{
 		auto slot = asio::get_associated_cancellation_slot(_handler);
@@ -237,6 +237,29 @@ private:
 	}
 };
 
+template <typename ClientService>
+class initiate_async_unsubscribe {
+	std::shared_ptr<ClientService> _svc_ptr;
+public:
+	explicit initiate_async_unsubscribe(std::shared_ptr<ClientService> svc_ptr) :
+		_svc_ptr(std::move(svc_ptr))
+	{}
+
+	using executor_type = typename ClientService::executor_type;
+	executor_type get_executor() const noexcept {
+		return _svc_ptr->get_executor();
+	}
+
+	template <typename Handler>
+	void operator()(
+		Handler&& handler,
+		const std::vector<std::string>& topics, const unsubscribe_props& props
+	) {
+		detail::unsubscribe_op { _svc_ptr, std::move(handler) }
+			.perform(topics, props);
+	}
+};
+
 
 } // end namespace async_mqtt5::detail
 

include/async_mqtt5/mqtt_client.hpp

@@ -18,10 +18,9 @@
 
 #include <async_mqtt5/impl/client_service.hpp>
 #include <async_mqtt5/impl/publish_send_op.hpp>
-#include <async_mqtt5/impl/read_message_op.hpp>
+#include <async_mqtt5/impl/re_auth_op.hpp>
 #include <async_mqtt5/impl/subscribe_op.hpp>
 #include <async_mqtt5/impl/unsubscribe_op.hpp>
-#include <async_mqtt5/impl/re_auth_op.hpp>
 
 namespace async_mqtt5 {
 
@@ -202,18 +201,8 @@ public:
 	>
 	decltype(auto) async_run(CompletionToken&& token = {}) {
 		using Signature = void (error_code);
-
-		auto initiation = [] (auto handler, const impl_type& impl) {
-			auto ex = asio::get_associated_executor(handler, impl->get_executor());
-
-			impl->run(std::move(handler));
-			detail::ping_op { impl, ex }.perform();
-			detail::read_message_op { impl, ex }.perform();
-			detail::sentry_op { impl, ex }.perform();
-		};
-
 		return asio::async_initiate<CompletionToken, Signature>(
-			initiation, token, _impl
+			detail::initiate_async_run(_impl), token
 		);
 	}
 
@@ -511,24 +500,10 @@ public:
 		CompletionToken&& token = {}
 	) {
 		using Signature = detail::on_publish_signature<qos_type>;
-
-		auto initiation = [] (
-			auto handler, std::string topic, std::string payload,
-			retain_e retain, const publish_props& props,
-			const impl_type& impl
-		) {
-			detail::publish_send_op<
-				client_service_type, decltype(handler), qos_type
-			> { impl, std::move(handler) }
-				.perform(
-					std::move(topic), std::move(payload),
-					retain, props
-				);
-		};
-
 		return asio::async_initiate<CompletionToken, Signature>(
-			initiation, token,
+			detail::initiate_async_publish<client_service_type, qos_type>(_impl),
-			std::move(topic), std::move(payload), retain, props, _impl
+			token,
+			std::move(topic), std::move(payload), retain, props
 		);
 	}
 
@@ -599,17 +574,9 @@ public:
 		using Signature = void (
 			error_code, std::vector<reason_code>, suback_props
 		);
-
-		auto initiation = [] (
-			auto handler, const std::vector<subscribe_topic>& topics,
-			const subscribe_props& props, const impl_type& impl
-		) {
-			detail::subscribe_op { impl, std::move(handler) }
-				.perform(topics, props);
-		};
-
 		return asio::async_initiate<CompletionToken, Signature>(
-			initiation, token, topics, props, _impl
+			detail::initiate_async_subscribe(_impl), token,
+			topics, props
 		);
 	}
 
@@ -745,18 +712,9 @@ public:
 		using Signature = void (
 			error_code, std::vector<reason_code>, unsuback_props
 		);
-
-		auto initiation = [](
-			auto handler,
-			const std::vector<std::string>& topics,
-			const unsubscribe_props& props,	const impl_type& impl
-		) {
-			detail::unsubscribe_op { impl, std::move(handler) }
-				.perform(topics, props);
-		};
-
 		return asio::async_initiate<CompletionToken, Signature>(
-			initiation, token, topics, props, _impl
+			detail::initiate_async_unsubscribe(_impl), token,
+			topics, props
 		);
 	}
 
@@ -879,10 +837,7 @@ public:
 			typename asio::default_completion_token<executor_type>::type
 	>
 	decltype(auto) async_receive(CompletionToken&& token = {}) {
-		// Sig = void (error_code, std::string, std::string, publish_props)
+		return _impl->async_channel_receive(std::forward<CompletionToken>(token));
-		return _impl->async_channel_receive(
-			std::forward<CompletionToken>(token)
-		);
 	}
 
 	/**
@@ -1008,6 +963,7 @@ public:
 			disconnect_props {}, std::forward<CompletionToken>(token)
 		);
 	}
+
 };