Add log invoke and logger

Summary: related to T15252, #24

Reviewers: ivica

Reviewed By: ivica

Subscribers: iljazovic, miljen

Maniphest Tasks: T15252

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

include/async_mqtt5/impl/autoconnect_stream.hpp

@@ -20,6 +20,7 @@
 
 #include <async_mqtt5/detail/async_mutex.hpp>
 #include <async_mqtt5/detail/async_traits.hpp>
+#include <async_mqtt5/detail/log_invoke.hpp>
 
 #include <async_mqtt5/impl/endpoints.hpp>
 #include <async_mqtt5/impl/read_op.hpp>
@@ -33,13 +34,15 @@ using error_code = boost::system::error_code;
 
 template <
 	typename StreamType,
-	typename StreamContext = std::monostate
+	typename StreamContext = std::monostate,
+	typename LoggerType = noop_logger
 >
 class autoconnect_stream {
 public:
-	using self_type = autoconnect_stream<StreamType, StreamContext>;
+	using self_type = autoconnect_stream<StreamType, StreamContext, LoggerType>;
 	using stream_type = StreamType;
 	using stream_context_type = StreamContext;
+	using logger_type = LoggerType;
 	using executor_type = typename stream_type::executor_type;
 private:
 	using stream_ptr = std::shared_ptr<stream_type>;
@@ -47,29 +50,33 @@ private:
 	executor_type _stream_executor;
 	async_mutex _conn_mtx;
 	asio::steady_timer _read_timer, _connect_timer;
-	endpoints _endpoints;
+	endpoints<logger_type> _endpoints;
 
 	stream_ptr _stream_ptr;
 	stream_context_type& _stream_context;
 
+	log_invoke<logger_type>& _log;
+
 	template <typename Owner, typename Handler>
 	friend class read_op;
 
 	template <typename Owner, typename Handler>
 	friend class write_op;
 
-	template <typename Stream>
+	template <typename Owner>
 	friend class reconnect_op;
 
 public:
 	autoconnect_stream(
-		const executor_type& ex, stream_context_type& context
+		const executor_type& ex, stream_context_type& context,
+		log_invoke<logger_type>& log
 	) :
 		_stream_executor(ex),
 		_conn_mtx(_stream_executor),
 		_read_timer(_stream_executor), _connect_timer(_stream_executor),
-		_endpoints(_stream_executor, _connect_timer),
-		_stream_context(context)
+		_endpoints(_stream_executor, _connect_timer, log),
+		_stream_context(context),
+		_log(log)
 	{
 		replace_next_layer(construct_next_layer());
 	}
@@ -166,6 +173,11 @@ public:
 	}
 
 private:
+
+	log_invoke<logger_type>& log() {
+		return _log;
+	}
+
 	static void open_lowest_layer(const stream_ptr& sptr, asio::ip::tcp protocol) {
 		error_code ec;
 		auto& layer = lowest_layer(*sptr);

include/async_mqtt5/impl/client_service.hpp

@@ -19,9 +19,8 @@
 #include <boost/asio/prepend.hpp>
 #include <boost/asio/experimental/basic_channel.hpp>
 
-#include <async_mqtt5/types.hpp>
-
 #include <async_mqtt5/detail/channel_traits.hpp>
+#include <async_mqtt5/detail/log_invoke.hpp>
 #include <async_mqtt5/detail/internal_types.hpp>
 
 #include <async_mqtt5/impl/assemble_op.hpp>
@@ -213,18 +212,20 @@ public:
 
 template <
 	typename StreamType,
-	typename TlsContext = std::monostate
+	typename TlsContext = std::monostate,
+	typename LoggerType = noop_logger
 >
 class client_service {
-	using self_type = client_service<StreamType, TlsContext>;
+	using self_type = client_service<StreamType, TlsContext, LoggerType>;
 	using stream_context_type = stream_context<StreamType, TlsContext>;
 	using stream_type = autoconnect_stream<
-		StreamType, stream_context_type
+		StreamType, stream_context_type, LoggerType
 	>;
 public:
 	using executor_type = typename stream_type::executor_type;
 private:
 	using tls_context_type = TlsContext;
+	using logger_type = LoggerType;
 	using receive_channel = asio::experimental::basic_channel<
 		executor_type,
 		channel_traits<>,
@@ -251,6 +252,8 @@ private:
 
 	executor_type _executor;
 
+	log_invoke<logger_type> _log;
+
 	stream_context_type _stream_context;
 	stream_type _stream;
 
@@ -267,8 +270,10 @@ private:
 	asio::steady_timer _sentry_timer;
 
 	client_service(const client_service& other) :
-		_executor(other._executor), _stream_context(other._stream_context),
-		_stream(_executor, _stream_context),
+		_executor(other._executor),
+		_log(other._log),
+		_stream_context(other._stream_context),
+		_stream(_executor, _stream_context, _log),
 		_replies(_executor),
 		_async_sender(*this),
 		_active_span(_read_buff.cend(), _read_buff.cend()),
@@ -283,11 +288,12 @@ public:
 
 	explicit client_service(
 		const executor_type& ex,
-		tls_context_type tls_context = {}
+		tls_context_type tls_context = {}, logger_type logger = {}
 	) :
 		_executor(ex),
+		_log(std::move(logger)),
 		_stream_context(std::move(tls_context)),
-		_stream(ex, _stream_context),
+		_stream(ex, _stream_context, _log),
 		_replies(ex),
 		_async_sender(*this),
 		_active_span(_read_buff.cend(), _read_buff.cend()),
@@ -411,6 +417,10 @@ public:
 		_stream.close();
 	}
 
+	log_invoke<LoggerType>& log() {
+		return _log;
+	}
+
 	uint16_t allocate_pid() {
 		return _pid_allocator.allocate();
 	}

include/async_mqtt5/impl/connect_op.hpp

@@ -37,13 +37,14 @@
 #include <async_mqtt5/detail/async_traits.hpp>
 #include <async_mqtt5/detail/control_packet.hpp>
 #include <async_mqtt5/detail/internal_types.hpp>
+#include <async_mqtt5/detail/log_invoke.hpp>
 
 #include <async_mqtt5/impl/codecs/message_decoders.hpp>
 #include <async_mqtt5/impl/codecs/message_encoders.hpp>
 
 namespace async_mqtt5::detail {
 
-template <typename Stream>
+template <typename Stream, typename LoggerType>
 class connect_op {
 	static constexpr size_t min_packet_sz = 5;
 
@@ -60,6 +61,7 @@ class connect_op {
 
 	Stream& _stream;
 	mqtt_ctx& _ctx;
+	log_invoke<LoggerType>& _log;
 
 	using handler_type = asio::any_completion_handler<void (error_code)>;
 	handler_type _handler;
@@ -72,9 +74,11 @@ class connect_op {
 public:
 	template <typename Handler>
 	connect_op(
-		Stream& stream, mqtt_ctx& ctx, Handler&& handler
+		Stream& stream, mqtt_ctx& ctx,
+		log_invoke<LoggerType>& log,
+		Handler&& handler
 	) :
-		_stream(stream), _ctx(ctx),
+		_stream(stream), _ctx(ctx), _log(log),
 		_handler(std::forward<Handler>(handler)),
 		_cancellation_state(
 			asio::get_associated_cancellation_slot(_handler),
@@ -121,6 +125,7 @@ public:
 		if (is_cancelled())
 			return complete(asio::error::operation_aborted);
 
+		_log.at_tcp_connect(ec, ep);
 		if (ec)
 			return complete(ec);
 
@@ -153,19 +158,19 @@ public:
 	}
 
 	void operator()(
-		on_tls_handshake, error_code ec,
-		endpoint ep, authority_path ap
+		on_tls_handshake, error_code ec, endpoint ep, authority_path ap
 	) {
 		if (is_cancelled())
 			return complete(asio::error::operation_aborted);
 
+		_log.at_tls_handshake(ec, ep);
 		if (ec)
 			return complete(ec);
 
 		do_ws_handshake(std::move(ep), std::move(ap));
 	}
 
-	void do_ws_handshake(endpoint, authority_path ap) {
+	void do_ws_handshake(endpoint ep, authority_path ap) {
 		if constexpr (has_ws_handshake<Stream>) {
 			using namespace boost::beast;
 
@@ -189,17 +194,23 @@ public:
 
 			_stream.async_handshake(
 				ap.host + ':' + ap.port, ap.path,
-				asio::prepend(std::move(*this), on_ws_handshake {})
+				asio::append(
+					asio::prepend(std::move(*this), on_ws_handshake {}),
+					ep
+				)
 			);
 		}
 		else
-			(*this)(on_ws_handshake {}, error_code {});
+			(*this)(on_ws_handshake {}, error_code {}, ep);
 	}
 
-	void operator()(on_ws_handshake, error_code ec) {
+	void operator()(on_ws_handshake, error_code ec, endpoint ep) {
 		if (is_cancelled())
 			return complete(asio::error::operation_aborted);
 
+		if constexpr (has_ws_handshake<Stream>)
+			_log.at_ws_handshake(ec, ep);
+
 		if (ec)
 			return complete(ec);
 
@@ -343,6 +354,7 @@ public:
 		if (!rc.has_value()) // reason code not allowed in CONNACK
 			return complete(client::error::malformed_packet);
 
+		_log.at_connack(*rc, session_present, ca_props);
 		if (*rc)
 			return complete(asio::error::try_again);
 
@@ -401,7 +413,7 @@ public:
 		detail::async_write(
 			_stream, asio::buffer(wire_data),
 			asio::consign(
-				asio::prepend(std::move(*this), on_send_auth{}),
+				asio::prepend(std::move(*this), on_send_auth {}),
 				std::move(packet)
 			)
 		);

include/async_mqtt5/impl/endpoints.hpp

@@ -27,6 +27,7 @@
 #include <boost/spirit/home/x3.hpp>
 
 #include <async_mqtt5/types.hpp>
+#include <async_mqtt5/detail/log_invoke.hpp>
 
 namespace async_mqtt5::detail {
 
@@ -107,11 +108,13 @@ public:
 		error_code timer_ec, authority_path ap
 	) {
 		if (
-			ord[0] == 0 && resolve_ec == asio::error::operation_aborted ||
-			ord[0] == 1 && timer_ec == asio::error::operation_aborted
+			(ord[0] == 0 && resolve_ec == asio::error::operation_aborted) ||
+			(ord[0] == 1 && timer_ec == asio::error::operation_aborted)
 		)
 			return complete(asio::error::operation_aborted, {}, {});
 
+		resolve_ec = timer_ec ? resolve_ec : asio::error::timed_out;
+		_owner._log.at_resolve(resolve_ec, ap.host, ap.port, epts);
 		if (!resolve_ec)
 			return complete(error_code {}, std::move(epts), std::move(ap));
 
@@ -136,7 +139,10 @@ private:
 };
 
 
+template <typename LoggerType>
 class endpoints {
+	using logger_type = LoggerType;
+
 	asio::ip::tcp::resolver _resolver;
 	asio::steady_timer& _connect_timer;
 
@@ -144,6 +150,8 @@ class endpoints {
 
 	int _current_host { -1 };
 
+	log_invoke<logger_type>& _log;
+
 	template <typename Owner, typename Handler>
 	friend class resolve_op;
 
@@ -159,8 +167,12 @@ class endpoints {
 
 public:
 	template <typename Executor>
-	endpoints(Executor ex, asio::steady_timer& timer) :
-		_resolver(ex), _connect_timer(timer)
+	endpoints(
+		Executor ex, asio::steady_timer& timer,
+		log_invoke<logger_type>& log
+	) :
+		_resolver(std::move(ex)), _connect_timer(timer),
+		_log(log)
 	{}
 
 	endpoints(const endpoints&) = delete;

include/async_mqtt5/impl/read_message_op.hpp

@@ -16,6 +16,7 @@
 #include <boost/asio/recycling_allocator.hpp>
 
 #include <async_mqtt5/types.hpp>
+#include <async_mqtt5/reason_codes.hpp>
 
 #include <async_mqtt5/detail/control_packet.hpp>
 
@@ -114,6 +115,20 @@ private:
 			}
 			break;
 			case control_code_e::disconnect: {
+				auto rv = decoders::decode_disconnect(
+					static_cast<uint32_t>(std::distance(first, last)), first
+				);
+				if (!rv.has_value())
+					return on_malformed_packet(
+						"Malformed DISCONNECT received: cannot decode"
+					);
+
+				const auto& [rc, props] = *rv;
+				_svc_ptr->log().at_disconnect(
+					to_reason_code<reason_codes::category::disconnect>(rc)
+						.value_or(reason_codes::unspecified_error),
+					props
+				);
 				_svc_ptr->close_stream();
 				_svc_ptr->open_stream();
 			}

include/async_mqtt5/impl/reconnect_op.hpp

@@ -182,9 +182,10 @@ public:
 
 		auto init_connect = [](
 			auto handler, typename Owner::stream_type& stream,
-			mqtt_ctx& context, endpoint ep, authority_path ap
+			mqtt_ctx& context, log_invoke<typename Owner::logger_type>& log,
+			endpoint ep, authority_path ap
 		) {
-			connect_op { stream, context, std::move(handler) }
+			connect_op { stream, context, log, std::move(handler) }
 				.perform(ep, std::move(ap));
 		};
 
@@ -192,6 +193,7 @@ public:
 			asio::async_initiate<const asio::deferred_t, void(error_code)>(
 				init_connect, asio::deferred, std::ref(*sptr),
 				std::ref(_owner._stream_context.mqtt_context()),
+				std::ref(_owner.log()),
 				ep, ap
 			),
 			_owner._connect_timer.async_wait(asio::deferred)