2023-10-05 13:59:32 +02:00
|
|
|
#ifndef ASYNC_MQTT5_UNSUBSCRIBE_OP_HPP
|
|
|
|
#define ASYNC_MQTT5_UNSUBSCRIBE_OP_HPP
|
|
|
|
|
2023-10-06 11:51:04 +02:00
|
|
|
#include <boost/asio/associated_allocator.hpp>
|
2023-10-05 13:59:32 +02:00
|
|
|
#include <boost/asio/detached.hpp>
|
|
|
|
|
|
|
|
#include <async_mqtt5/error.hpp>
|
|
|
|
|
2023-10-06 11:51:04 +02:00
|
|
|
#include <async_mqtt5/detail/cancellable_handler.hpp>
|
|
|
|
#include <async_mqtt5/detail/control_packet.hpp>
|
|
|
|
#include <async_mqtt5/detail/internal_types.hpp>
|
2023-12-13 15:13:07 +01:00
|
|
|
#include <async_mqtt5/detail/topic_validation.hpp>
|
2023-10-05 13:59:32 +02:00
|
|
|
|
|
|
|
#include <async_mqtt5/impl/disconnect_op.hpp>
|
2023-12-07 09:32:34 +01:00
|
|
|
#include <async_mqtt5/impl/codecs/message_decoders.hpp>
|
|
|
|
#include <async_mqtt5/impl/codecs/message_encoders.hpp>
|
2023-10-05 13:59:32 +02:00
|
|
|
|
|
|
|
namespace async_mqtt5::detail {
|
|
|
|
|
|
|
|
namespace asio = boost::asio;
|
|
|
|
|
|
|
|
template <typename ClientService, typename Handler>
|
|
|
|
class unsubscribe_op {
|
|
|
|
using client_service = ClientService;
|
|
|
|
struct on_unsubscribe {};
|
|
|
|
struct on_unsuback {};
|
|
|
|
|
|
|
|
std::shared_ptr<client_service> _svc_ptr;
|
|
|
|
|
|
|
|
cancellable_handler<
|
|
|
|
Handler,
|
|
|
|
typename client_service::executor_type,
|
|
|
|
std::tuple<std::vector<reason_code>, unsuback_props>
|
|
|
|
> _handler;
|
|
|
|
|
|
|
|
public:
|
|
|
|
unsubscribe_op(
|
2023-12-07 09:32:34 +01:00
|
|
|
const std::shared_ptr<client_service>& svc_ptr,
|
|
|
|
Handler&& handler
|
2023-10-05 13:59:32 +02:00
|
|
|
) :
|
|
|
|
_svc_ptr(svc_ptr),
|
|
|
|
_handler(std::move(handler), get_executor())
|
|
|
|
{}
|
|
|
|
|
|
|
|
unsubscribe_op(unsubscribe_op&&) noexcept = default;
|
|
|
|
unsubscribe_op(const unsubscribe_op&) noexcept = delete;
|
|
|
|
|
|
|
|
using executor_type = typename client_service::executor_type;
|
|
|
|
executor_type get_executor() const noexcept {
|
|
|
|
return _svc_ptr->get_executor();
|
|
|
|
}
|
|
|
|
|
|
|
|
using allocator_type = asio::associated_allocator_t<Handler>;
|
|
|
|
allocator_type get_allocator() const noexcept {
|
|
|
|
return asio::get_associated_allocator(_handler);
|
|
|
|
}
|
|
|
|
|
|
|
|
void perform(
|
|
|
|
const std::vector<std::string>& topics,
|
|
|
|
const unsubscribe_props& props
|
2023-12-07 12:18:28 +01:00
|
|
|
) {
|
2023-10-05 13:59:32 +02:00
|
|
|
uint16_t packet_id = _svc_ptr->allocate_pid();
|
|
|
|
if (packet_id == 0)
|
2023-12-22 09:17:45 +01:00
|
|
|
return complete_post(
|
|
|
|
client::error::pid_overrun, packet_id, topics.size()
|
|
|
|
);
|
|
|
|
|
|
|
|
auto ec = validate_unsubscribe(topics, props);
|
|
|
|
if (ec)
|
|
|
|
return complete_post(ec, packet_id, topics.size());
|
2023-10-05 13:59:32 +02:00
|
|
|
|
|
|
|
auto unsubscribe = control_packet<allocator_type>::of(
|
|
|
|
with_pid, get_allocator(),
|
|
|
|
encoders::encode_unsubscribe, packet_id,
|
|
|
|
topics, props
|
|
|
|
);
|
|
|
|
|
2024-01-02 16:01:56 +01:00
|
|
|
auto max_packet_size = static_cast<size_t>(
|
2024-01-04 15:31:06 +01:00
|
|
|
_svc_ptr->connack_property(prop::maximum_packet_size)
|
2024-01-02 16:01:56 +01:00
|
|
|
.value_or(default_max_send_size)
|
|
|
|
);
|
2023-12-22 09:17:45 +01:00
|
|
|
if (unsubscribe.size() > max_packet_size)
|
|
|
|
return complete_post(
|
2023-12-22 13:48:26 +01:00
|
|
|
client::error::packet_too_large, packet_id, topics.size()
|
2023-12-22 09:17:45 +01:00
|
|
|
);
|
|
|
|
|
2023-10-05 13:59:32 +02:00
|
|
|
send_unsubscribe(std::move(unsubscribe));
|
|
|
|
}
|
|
|
|
|
|
|
|
void send_unsubscribe(control_packet<allocator_type> unsubscribe) {
|
2023-11-28 11:04:00 +01:00
|
|
|
if (_handler.empty()) // already cancelled
|
|
|
|
return _svc_ptr->free_pid(unsubscribe.packet_id());
|
|
|
|
|
2023-12-22 10:07:20 +01:00
|
|
|
auto wire_data = unsubscribe.wire_data();
|
2023-10-05 13:59:32 +02:00
|
|
|
_svc_ptr->async_send(
|
|
|
|
wire_data,
|
|
|
|
no_serial, send_flag::none,
|
|
|
|
asio::prepend(
|
2023-12-22 09:17:45 +01:00
|
|
|
std::move(*this), on_unsubscribe {}, std::move(unsubscribe)
|
2023-10-05 13:59:32 +02:00
|
|
|
)
|
|
|
|
);
|
|
|
|
}
|
|
|
|
|
|
|
|
void operator()(
|
|
|
|
on_unsubscribe, control_packet<allocator_type> packet,
|
|
|
|
error_code ec
|
|
|
|
) {
|
|
|
|
if (ec == asio::error::try_again)
|
|
|
|
return send_unsubscribe(std::move(packet));
|
|
|
|
|
|
|
|
auto packet_id = packet.packet_id();
|
|
|
|
|
|
|
|
if (ec)
|
|
|
|
return complete(ec, packet_id, {}, {});
|
|
|
|
|
|
|
|
_svc_ptr->async_wait_reply(
|
|
|
|
control_code_e::unsuback, packet_id,
|
|
|
|
asio::prepend(std::move(*this), on_unsuback{}, std::move(packet))
|
|
|
|
);
|
|
|
|
}
|
|
|
|
|
|
|
|
void operator()(
|
|
|
|
on_unsuback, control_packet<allocator_type> packet,
|
|
|
|
error_code ec, byte_citer first, byte_citer last
|
|
|
|
) {
|
|
|
|
if (ec == asio::error::try_again) // "resend unanswered"
|
|
|
|
return send_unsubscribe(std::move(packet));
|
|
|
|
|
|
|
|
uint16_t packet_id = packet.packet_id();
|
|
|
|
|
|
|
|
if (ec)
|
|
|
|
return complete(ec, packet_id, {}, {});
|
|
|
|
|
|
|
|
auto unsuback = decoders::decode_unsuback(
|
2024-01-02 11:40:53 +01:00
|
|
|
static_cast<uint32_t>(std::distance(first, last)), first
|
2023-10-05 13:59:32 +02:00
|
|
|
);
|
|
|
|
if (!unsuback.has_value()) {
|
|
|
|
on_malformed_packet("Malformed UNSUBACK: cannot decode");
|
|
|
|
return send_unsubscribe(std::move(packet));
|
|
|
|
}
|
|
|
|
|
|
|
|
auto& [props, reason_codes] = *unsuback;
|
|
|
|
|
|
|
|
complete(
|
|
|
|
ec, packet_id,
|
|
|
|
to_reason_codes(std::move(reason_codes)), std::move(props)
|
|
|
|
);
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
|
2023-12-14 15:23:08 +01:00
|
|
|
static error_code validate_unsubscribe(
|
|
|
|
const std::vector<std::string>& topics,
|
|
|
|
const unsubscribe_props& props
|
|
|
|
) {
|
2023-11-29 11:50:07 +01:00
|
|
|
for (const auto& topic : topics)
|
2023-12-13 15:13:07 +01:00
|
|
|
if (validate_topic_filter(topic) != validation_result::valid)
|
2023-11-29 11:50:07 +01:00
|
|
|
return client::error::invalid_topic;
|
2023-12-14 15:23:08 +01:00
|
|
|
|
|
|
|
auto user_properties = props[prop::user_property];
|
|
|
|
for (const auto& user_prop: user_properties)
|
|
|
|
if (validate_mqtt_utf8(user_prop) != validation_result::valid)
|
|
|
|
return client::error::malformed_packet;
|
2023-11-29 11:50:07 +01:00
|
|
|
return error_code {};
|
|
|
|
}
|
|
|
|
|
2023-10-05 13:59:32 +02:00
|
|
|
static std::vector<reason_code> to_reason_codes(std::vector<uint8_t> codes) {
|
|
|
|
std::vector<reason_code> ret;
|
|
|
|
for (uint8_t code : codes) {
|
|
|
|
auto rc = to_reason_code<reason_codes::category::unsuback>(code);
|
|
|
|
if (rc)
|
|
|
|
ret.push_back(*rc);
|
|
|
|
}
|
|
|
|
return ret;
|
|
|
|
}
|
|
|
|
|
|
|
|
void on_malformed_packet(
|
|
|
|
const std::string& reason
|
|
|
|
) {
|
2023-10-06 11:51:04 +02:00
|
|
|
auto props = disconnect_props {};
|
2023-10-05 13:59:32 +02:00
|
|
|
props[prop::reason_string] = reason;
|
|
|
|
async_disconnect(
|
|
|
|
disconnect_rc_e::malformed_packet, props, false, _svc_ptr,
|
|
|
|
asio::detached
|
|
|
|
);
|
|
|
|
}
|
|
|
|
|
2023-12-22 09:17:45 +01:00
|
|
|
void complete_post(error_code ec, uint16_t packet_id, size_t num_topics) {
|
|
|
|
if (packet_id != 0)
|
|
|
|
_svc_ptr->free_pid(packet_id);
|
2023-10-05 13:59:32 +02:00
|
|
|
_handler.complete_post(
|
2023-12-07 09:32:34 +01:00
|
|
|
ec, std::vector<reason_code> { num_topics, reason_codes::empty },
|
|
|
|
unsuback_props {}
|
2023-10-05 13:59:32 +02:00
|
|
|
);
|
|
|
|
}
|
|
|
|
|
|
|
|
void complete(
|
|
|
|
error_code ec, uint16_t packet_id,
|
|
|
|
std::vector<reason_code> reason_codes, unsuback_props props
|
|
|
|
) {
|
|
|
|
_svc_ptr->free_pid(packet_id);
|
|
|
|
_handler.complete(ec, std::move(reason_codes), std::move(props));
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
|
|
} // end namespace async_mqtt5::detail
|
|
|
|
|
|
|
|
#endif // !ASYNC_MQTT5_UNSUBSCRIBE_OP_HPP
|