// // Copyright (c) 2023-2025 Ivica Siladic, Bruno Iljazovic, Korina Simicevic // // Distributed under the Boost Software License, Version 1.0. // (See accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt) // #ifndef BOOST_MQTT5_TEST_DELAYED_OP_HPP #define BOOST_MQTT5_TEST_DELAYED_OP_HPP #include #include #include #include #include #include #include #include #include namespace boost::mqtt5::test { namespace asio = boost::asio; using error_code = boost::system::error_code; using time_stamp = std::chrono::time_point; using duration = time_stamp::duration; template class delayed_op { struct on_timer {}; std::unique_ptr _timer; time_stamp::duration _delay; asio::cancellation_slot _cancel_slot; std::tuple _args; public: template delayed_op( const Executor& ex, time_stamp::duration delay, Args&& ...args ) : _timer(new asio::steady_timer(ex)), _delay(delay), _args(std::move(args)...) {} delayed_op(delayed_op&&) = default; delayed_op(const delayed_op&) = delete; delayed_op& operator=(delayed_op&&) = default; delayed_op& operator=(const delayed_op&) = delete; using allocator_type = asio::recycling_allocator; allocator_type get_allocator() const noexcept { return allocator_type {}; } using cancellation_slot_type = asio::cancellation_slot; asio::cancellation_slot get_cancellation_slot() const noexcept { return _cancel_slot; } using executor_type = asio::steady_timer::executor_type; executor_type get_executor() const noexcept { return _timer->get_executor(); } template void perform(CompletionHandler&& handler) { _cancel_slot = asio::get_associated_cancellation_slot(handler); _timer->expires_after(_delay); _timer->async_wait( asio::prepend(std::move(*this), on_timer {}, std::move(handler)) ); } template void operator()(on_timer, CompletionHandler&& h, error_code ec) { // The timer places a handler into the cancellation slot // and does not clear it. Therefore, we need to clear it explicitly // to properly remove the corresponding cancellation signal // in the test_broker. get_cancellation_slot().clear(); auto bh = std::apply( [h = std::move(h)](auto&&... args) mutable { return asio::append(std::move(h), std::move(args)...); }, _args ); asio::dispatch(asio::prepend(std::move(bh), ec)); } }; template decltype(auto) async_delay( asio::cancellation_slot cancel_slot, delayed_op&& op, CompletionToken&& token ) { using Signature = void (error_code, boost::remove_cv_ref_t...); auto initiation = []( auto handler, asio::cancellation_slot cancel_slot, delayed_op op ) { op.perform( asio::bind_cancellation_slot(cancel_slot, std::move(handler)) ); }; return asio::async_initiate( std::move(initiation), token, cancel_slot, std::move(op) ); } } // end namespace boost::mqtt5::test #endif // BOOST_MQTT5_TEST_DELAYED_OP_HPP