Skip to content

Fixed qos2_publish_handled_ logic. #824

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Aug 8, 2021
Merged

Fixed qos2_publish_handled_ logic. #824

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion include/mqtt/async_client.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,7 +267,7 @@ class async_client : public client<Socket, PacketIdBytes> {
) {
set_auto_pub_response();
base::set_async_pingreq(true);
base::set_async_error_notify(true);
base::set_async_notify(true);
}
};

Expand Down
167 changes: 101 additions & 66 deletions include/mqtt/endpoint.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -858,14 +858,16 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
}

/**
* @brief Set async error notify flag
* @param async send connack/disconnect if error happens
* @brief Set async notify flag
* @param async send packet
*
* MQTT protocol requests sending connack/disconnect packet with error reason code if some error happens.<BR>
* This function choose sync/async connack/disconnect.<BR>
* MQTT protocol requests sending pubrec even if the corresponding publish has already been handled.<BR>
* This function choose sync/async pubrec.<BR>
*/
void set_async_error_notify(bool async = true) {
async_error_notify_ = async;
void set_async_notify(bool async = true) {
async_notify_ = async;
}

/**
Expand Down Expand Up @@ -5023,7 +5025,7 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
}

void send_error_disconnect(v5::disconnect_reason_code rc) {
if (async_error_notify_) {
if (async_notify_) {
async_disconnect(rc);
}
else {
Expand All @@ -5032,7 +5034,7 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
}

void send_error_connack(v5::connect_reason_code rc) {
if (async_error_notify_) {
if (async_notify_) {
async_connack(false, rc);
}
else {
Expand Down Expand Up @@ -7342,23 +7344,12 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
[&] {
switch (ep_.version_) {
case protocol_version::v3_1_1:
if (ep_.on_publish(
packet_id_,
publish_options(ep_.fixed_header_),
force_move(topic_name_),
force_move(force_move(variant_get<buffer>(var))))) {
ep_.on_mqtt_message_processed(
force_move(
std::get<0>(
any_cast<
std::tuple<any, process_type_sp>
>(session_life_keeper)
)
)
);
return true;
}
break;
return ep_.on_publish(
packet_id_,
publish_options(ep_.fixed_header_),
force_move(topic_name_),
force_move(variant_get<buffer>(var))
);
case protocol_version::v5:
if (topic_name_.empty()) {
if (auto topic_alias = get_topic_alias_from_props(props_)) {
Expand Down Expand Up @@ -7390,37 +7381,43 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
}
}
}
if (ep_.on_v5_publish(
packet_id_,
publish_options(ep_.fixed_header_),
force_move(topic_name_),
force_move(variant_get<buffer>(var)),
force_move(props_)
)
) {
ep_.on_mqtt_message_processed(
force_move(
std::get<0>(
any_cast<
std::tuple<any, process_type_sp>
>(session_life_keeper)
)
)
);
return true;
}
break;
return ep_.on_v5_publish(
packet_id_,
publish_options(ep_.fixed_header_),
force_move(topic_name_),
force_move(variant_get<buffer>(var)),
force_move(props_)
);
default:
BOOST_ASSERT(false);
}
return false;
};
switch (qos_value_) {
case qos::at_most_once:
handler_call();
if (handler_call()) {
ep_.on_mqtt_message_processed(
force_move(
std::get<0>(
any_cast<
std::tuple<any, process_type_sp>
>(session_life_keeper)
)
)
);
}
break;
case qos::at_least_once:
if (handler_call()) {
ep_.on_mqtt_message_processed(
force_move(
std::get<0>(
any_cast<
std::tuple<any, process_type_sp>
>(session_life_keeper)
)
)
);
ep_.auto_pub_response(
[this] {
if (ep_.connected_) {
Expand All @@ -7445,29 +7442,67 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
}
break;
case qos::exactly_once:
if (handler_call()) {
ep_.qos2_publish_handled_.emplace(*packet_id_);
ep_.auto_pub_response(
[this] {
if (ep_.connected_) {
ep_.send_pubrec(
*packet_id_,
v5::pubrec_reason_code::success,
v5::properties{}
);
}
},
[this] {
if (ep_.connected_) {
ep_.async_send_pubrec(
*packet_id_,
v5::pubrec_reason_code::success,
v5::properties{},
[](auto){}
);
if (ep_.qos2_publish_handled_.find(*packet_id_) == ep_.qos2_publish_handled_.end()) {
if (handler_call()) {
ep_.on_mqtt_message_processed(
force_move(
std::get<0>(
any_cast<
std::tuple<any, process_type_sp>
>(session_life_keeper)
)
)
);
ep_.qos2_publish_handled_.emplace(*packet_id_);
ep_.auto_pub_response(
[this] {
if (ep_.connected_) {
ep_.send_pubrec(
*packet_id_,
v5::pubrec_reason_code::success,
v5::properties{}
);
}
},
[this] {
if (ep_.connected_) {
ep_.async_send_pubrec(
*packet_id_,
v5::pubrec_reason_code::success,
v5::properties{},
[](auto){}
);
}
}
}
);
}
}
else {
// publish has already been handled
ep_.on_mqtt_message_processed(
force_move(
std::get<0>(
any_cast<
std::tuple<any, process_type_sp>
>(session_life_keeper)
)
)
);
if (ep_.async_notify_) {
ep_.async_send_pubrec(
*packet_id_,
v5::pubrec_reason_code::success,
v5::properties{},
[](auto){}
);
}
else {
ep_.send_pubrec(
*packet_id_,
v5::pubrec_reason_code::success,
v5::properties{}
);
}
}
break;
}
Expand Down Expand Up @@ -10458,7 +10493,7 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
std::set<packet_id_t> sub_unsub_inflight_;
bool auto_pub_response_{true};
bool auto_pub_response_async_{false};
bool async_error_notify_{false};
bool async_notify_{false};
bool async_send_store_ { false };
bool async_read_on_message_processed_ { true };
bool disconnect_requested_{false};
Expand Down
81 changes: 24 additions & 57 deletions test/system/st_resend.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1080,7 +1080,6 @@ BOOST_AUTO_TEST_CASE( publish_qos2_from_broker ) {
deps("h_error", "h_pubcomp", "h_publish1"),
// connect
cont("h_connack3"),
cont("h_publish2"),
// disconnect
cont("h_close2"),
};
Expand All @@ -1105,6 +1104,7 @@ BOOST_AUTO_TEST_CASE( publish_qos2_from_broker ) {
[&] {
MQTT_CHK("h_connack3");
BOOST_TEST(sp == true);
c->disconnect();
}
);
BOOST_TEST(ret);
Expand All @@ -1127,34 +1127,17 @@ BOOST_AUTO_TEST_CASE( publish_qos2_from_broker ) {
MQTT_NS::publish_options pubopts,
MQTT_NS::buffer topic,
MQTT_NS::buffer contents) {
auto ret = MQTT_ORDERED(
[&] {
MQTT_CHK("h_publish1");
BOOST_TEST(pubopts.get_dup() == MQTT_NS::dup::no);
BOOST_TEST(pubopts.get_qos() == MQTT_NS::qos::exactly_once);
BOOST_TEST(pubopts.get_retain() == MQTT_NS::retain::no);
BOOST_CHECK(packet_id);
pid_pub = packet_id.value();
BOOST_TEST(topic == "topic1");
BOOST_TEST(contents == "topic1_contents");
if (chk.passed("h_pubcomp")) {
c->force_disconnect();
}
},
[&] {
MQTT_CHK("h_publish2");
BOOST_TEST(pubopts.get_dup() == MQTT_NS::dup::yes);
BOOST_TEST(pubopts.get_qos() == MQTT_NS::qos::exactly_once);
BOOST_TEST(pubopts.get_retain() == MQTT_NS::retain::no);
BOOST_CHECK(packet_id);
BOOST_TEST(packet_id.value() == pid_pub);
BOOST_TEST(topic == "topic1");
BOOST_TEST(contents == "topic1_contents");
c->pubrec(packet_id.value());
c->disconnect();
}
);
BOOST_TEST(ret);
MQTT_CHK("h_publish1");
BOOST_TEST(pubopts.get_dup() == MQTT_NS::dup::no);
BOOST_TEST(pubopts.get_qos() == MQTT_NS::qos::exactly_once);
BOOST_TEST(pubopts.get_retain() == MQTT_NS::retain::no);
BOOST_CHECK(packet_id);
pid_pub = packet_id.value();
BOOST_TEST(topic == "topic1");
BOOST_TEST(contents == "topic1_contents");
if (chk.passed("h_pubcomp")) {
c->force_disconnect();
}
return true;
}
);
Expand Down Expand Up @@ -1198,6 +1181,7 @@ BOOST_AUTO_TEST_CASE( publish_qos2_from_broker ) {
[&] {
MQTT_CHK("h_connack3");
BOOST_TEST(sp == true);
c->disconnect();
}
);
BOOST_TEST(ret);
Expand All @@ -1221,34 +1205,17 @@ BOOST_AUTO_TEST_CASE( publish_qos2_from_broker ) {
MQTT_NS::buffer topic,
MQTT_NS::buffer contents,
MQTT_NS::v5::properties /*props*/) {
auto ret = MQTT_ORDERED(
[&] {
MQTT_CHK("h_publish1");
BOOST_TEST(pubopts.get_dup() == MQTT_NS::dup::no);
BOOST_TEST(pubopts.get_qos() == MQTT_NS::qos::exactly_once);
BOOST_TEST(pubopts.get_retain() == MQTT_NS::retain::no);
BOOST_CHECK(packet_id);
pid_pub = packet_id.value();
BOOST_TEST(topic == "topic1");
BOOST_TEST(contents == "topic1_contents");
if (chk.passed("h_pubcomp")) {
c->force_disconnect();
}
},
[&] {
MQTT_CHK("h_publish2");
BOOST_TEST(pubopts.get_dup() == MQTT_NS::dup::yes);
BOOST_TEST(pubopts.get_qos() == MQTT_NS::qos::exactly_once);
BOOST_TEST(pubopts.get_retain() == MQTT_NS::retain::no);
BOOST_CHECK(packet_id);
BOOST_TEST(packet_id.value() == pid_pub);
BOOST_TEST(topic == "topic1");
BOOST_TEST(contents == "topic1_contents");
c->pubrec(packet_id.value());
c->disconnect();
}
);
BOOST_TEST(ret);
MQTT_CHK("h_publish1");
BOOST_TEST(pubopts.get_dup() == MQTT_NS::dup::no);
BOOST_TEST(pubopts.get_qos() == MQTT_NS::qos::exactly_once);
BOOST_TEST(pubopts.get_retain() == MQTT_NS::retain::no);
BOOST_CHECK(packet_id);
pid_pub = packet_id.value();
BOOST_TEST(topic == "topic1");
BOOST_TEST(contents == "topic1_contents");
if (chk.passed("h_pubcomp")) {
c->force_disconnect();
}
return true;
}
);
Expand Down