eclipse/paho.mqtt.cpp
1
0
Fork 0
mirror of https://github.com/eclipse/paho.mqtt.cpp.git synced 2025-07-04 09:16:31 +08:00
Code Issues Releases Wiki Activity

Added backward compatibility for exception reason code.

Browse Source
This commit is contained in:
fpagliughi 2019-07-27 18:32:20 -04:00
parent 9a4c83e0bd
commit a2e6565a66
12 changed files with 1189 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/async_client.cpp
View File

@ -587,7 +587,7 @@ token_ptr async_client::subscribe(const_string_collection_ptr topicFilters,
if (n != qos.size()) if (n != qos.size())
throw std::invalid_argument("Collection sizes don't match"); throw std::invalid_argument("Collection sizes don't match");
auto tok = token::create(token::Type::SUBSCRIBE_MANY, *this, topicFilters); auto tok = token::create(token::Type::SUBSCRIBE, *this, topicFilters);
tok->set_num_expected(n); tok->set_num_expected(n);
add_token(tok); add_token(tok);
@ -614,7 +614,7 @@ token_ptr async_client::subscribe(const_string_collection_ptr topicFilters,
if (n != qos.size()) if (n != qos.size())
throw std::invalid_argument("Collection sizes don't match"); throw std::invalid_argument("Collection sizes don't match");
auto tok = token::create(token::Type::SUBSCRIBE_MANY, *this, auto tok = token::create(token::Type::SUBSCRIBE, *this,
topicFilters, userContext, cb); topicFilters, userContext, cb);
tok->set_num_expected(n); tok->set_num_expected(n);
add_token(tok); add_token(tok);
@ -635,6 +635,7 @@ token_ptr async_client::subscribe(const_string_collection_ptr topicFilters,
token_ptr async_client::subscribe(const string& topicFilter, int qos) token_ptr async_client::subscribe(const string& topicFilter, int qos)
{ {
auto tok = token::create(token::Type::SUBSCRIBE, *this, topicFilter); auto tok = token::create(token::Type::SUBSCRIBE, *this, topicFilter);
tok->set_num_expected(0); // Indicates non-array response for single val
add_token(tok); add_token(tok);
response_options opts(tok, mqttVersion_); response_options opts(tok, mqttVersion_);
@ -654,6 +655,7 @@ token_ptr async_client::subscribe(const string& topicFilter, int qos,
{ {
auto tok = token::create(token::Type::SUBSCRIBE, *this, topicFilter, auto tok = token::create(token::Type::SUBSCRIBE, *this, topicFilter,
userContext, cb); userContext, cb);
tok->set_num_expected(0);
add_token(tok); add_token(tok);
response_options opts(tok, mqttVersion_); response_options opts(tok, mqttVersion_);
@ -674,6 +676,7 @@ token_ptr async_client::subscribe(const string& topicFilter, int qos,
token_ptr async_client::unsubscribe(const string& topicFilter) token_ptr async_client::unsubscribe(const string& topicFilter)
{ {
auto tok = token::create(token::Type::UNSUBSCRIBE, *this, topicFilter); auto tok = token::create(token::Type::UNSUBSCRIBE, *this, topicFilter);
tok->set_num_expected(0); // Indicates non-array response for single val
add_token(tok); add_token(tok);
response_options opts(tok, mqttVersion_); response_options opts(tok, mqttVersion_);
@ -692,7 +695,7 @@ token_ptr async_client::unsubscribe(const_string_collection_ptr topicFilters)
{ {
size_t n = topicFilters->size(); size_t n = topicFilters->size();
auto tok = token::create(token::Type::UNSUBSCRIBE_MANY, *this, topicFilters); auto tok = token::create(token::Type::UNSUBSCRIBE, *this, topicFilters);
tok->set_num_expected(n); tok->set_num_expected(n);
add_token(tok); add_token(tok);
@ -714,7 +717,7 @@ token_ptr async_client::unsubscribe(const_string_collection_ptr topicFilters,
{ {
size_t n = topicFilters->size(); size_t n = topicFilters->size();
auto tok = token::create(token::Type::UNSUBSCRIBE_MANY, *this, topicFilters, auto tok = token::create(token::Type::UNSUBSCRIBE, *this, topicFilters,
userContext, cb); userContext, cb);
tok->set_num_expected(n); tok->set_num_expected(n);
add_token(tok); add_token(tok);

61
src/mqtt/exception.h
View File

@ -57,14 +57,21 @@ public:
* @param rc The error return code from the C library. * @param rc The error return code from the C library.
*/ */
explicit exception(int rc) explicit exception(int rc)
: exception(rc, string_error(rc)) {} : exception(rc, error_str(rc)) {}
/**
* Creates an MQTT exception.
* @param rc The error return code from the C library.
* @param reasonCode The reason code from the server response.
*/
explicit exception(int rc, ReasonCode reasonCode)
: exception(rc, reasonCode, error_str(rc)) {}
/** /**
* Creates an MQTT exception. * Creates an MQTT exception.
* @param rc The error return code from the C library. * @param rc The error return code from the C library.
* @param msg The text message for the error. * @param msg The text message for the error.
*/ */
exception(int rc, const string& msg) exception(int rc, const string& msg)
: std::runtime_error(printable_error(rc, msg)), : std::runtime_error(printable_error(rc, ReasonCode::SUCCESS, msg)),
rc_(rc), reasonCode_(ReasonCode::SUCCESS), msg_(msg) {} rc_(rc), reasonCode_(ReasonCode::SUCCESS), msg_(msg) {}
/** /**
* Creates an MQTT exception. * Creates an MQTT exception.
@ -73,31 +80,44 @@ public:
* @param msg The text message for the error. * @param msg The text message for the error.
*/ */
exception(int rc, ReasonCode reasonCode, const string& msg) exception(int rc, ReasonCode reasonCode, const string& msg)
: std::runtime_error(printable_error(rc, msg)), : std::runtime_error(printable_error(rc, reasonCode, msg)),
rc_(rc), reasonCode_(reasonCode), msg_(msg) {} rc_(rc), reasonCode_(reasonCode), msg_(msg) {}
/** /**
* Gets an error message from an error code. * Gets an error message from an error code.
* @param rc The error code from the C lib * @param rc The error code from the C lib
* @return A string explanation of the error * @return A string explanation of the error
*/ */
static string string_error(int rc) { static string error_str(int rc) {
const char *msg = ::MQTTAsync_strerror(rc); const char *msg = ::MQTTAsync_strerror(rc);
return msg ? string(msg) : string(); return msg ? string(msg) : string();
} }
/**
* Gets a string describing the MQTT v5 reason code.
* @param reasonCode The MQTT v5 reason code.
* @return A string describing the reason code.
*/
static string reason_code_str(int reasonCode) {
if (reasonCode != MQTTPP_V3_CODE) {
auto msg = ::MQTTReasonCode_toString(MQTTReasonCodes(reasonCode));
if (msg) return string(msg);
}
return string();
}
/** /**
* Gets a detailed error message for an error code. * Gets a detailed error message for an error code.
* @param rc The error code from the C lib * @param rc The error code from the C lib
* @param msg An optional additional message. If none is provided, the * @param msg An optional additional message. If none is provided, the
* string_error message is used. * error_str message is used.
* @return A string error message that includes the error code and an * @return A string error message that includes the error code and an
* explanation message. * explanation message.
*/ */
static string printable_error(int rc, const string& msg=string()) { static string printable_error(int rc, int reasonCode=ReasonCode::SUCCESS,
string s = "MQTT error [" + std::to_string(rc) + "]: "; const string& msg=string()) {
if (msg.empty()) string s = "MQTT error [" + std::to_string(rc) + "]";
s += string_error(rc); if (!msg.empty())
else s += string(": ") + msg;
s += msg; if (reasonCode != MQTTPP_V3_CODE && reasonCode != ReasonCode::SUCCESS)
s += string(". Reason: ") + reason_code_str(reasonCode);
return s; return s;
} }
/** /**
@ -105,9 +125,24 @@ public:
*/ */
int get_return_code() const { return rc_; } int get_return_code() const { return rc_; }
/** /**
* Returns the reason code for this exception. * Gets a string of the error code.
* @return A string of the error code.
*/ */
int get_reason_code() const { return reasonCode_; } string get_error_str() const { return error_str(rc_); }
/**
* Returns the reason code for this exception.
* For MQTT v3 connections, this is actually the return code.
*/
int get_reason_code() const {
return reasonCode_ == MQTTPP_V3_CODE ? rc_ : reasonCode_;
}
/**
* Gets a string for the reason code.
* @return A string for the reason code.
*/
string get_reason_code_str() const {
return reason_code_str(reasonCode_);
}
/** /**
* Returns the error message for this exception. * Returns the error message for this exception.
*/ */

49
src/mqtt/server_response.h
View File

@ -30,24 +30,39 @@
namespace mqtt { namespace mqtt {
/**
* Base class for responses from the server.
*/
class server_response class server_response
{ {
/** The properties from the acknowledge */ /** The properties from the acknowledge */
properties props_; properties props_;
friend class token;
public: public:
/**
* Creates a response with empty property list.
*/
server_response() {} server_response() {}
/**
* Creates a server response with the specified properties.
* @param props The properties in the response.
*/
server_response(const properties& props) server_response(const properties& props)
: props_(props) {} : props_(props) {}
/**
* Creates a server response with the specified properties.
* @param props The properties in the response.
*/
server_response(properties&& props) server_response(properties&& props)
: props_(std::move(props)) {} : props_(std::move(props)) {}
/**
* Virtual destructor.
*/
virtual ~server_response() {} virtual ~server_response() {}
/**
* Gets the properties from the response.
* @return The properties from the response.
*/
const properties& get_properties() const { return props_; } const properties& get_properties() const { return props_; }
}; };
@ -79,12 +94,29 @@ class connect_response : public server_response
} }
public: public:
/**
* Gets the URI of the broker to which we connected.
* @return The URI of the broker.
*/
string get_server_uri() const { return serverURI_; } string get_server_uri() const { return serverURI_; }
/**
* Gets the MQTT version for the connection.
* @return The MQTT version for the connection.
*/
int get_mqtt_version() const { return mqttVersion_; } int get_mqtt_version() const { return mqttVersion_; }
/**
* Determines whether a session already existed for this client on the
* server.
* This tells whether the server has a persistent session stored for the
* client, given the ClientID specified in the connect message.
* @return Whether a session already existed for this client on the server.
*/
bool is_session_present() const { return sessionPresent_; } bool is_session_present() const { return sessionPresent_; }
}; };
/** Response for subscribe messages */ /**
* Response for subscribe messages
*/
struct subscribe_response : public server_response struct subscribe_response : public server_response
{ {
/** The reason/result code for each topic request. */ /** The reason/result code for each topic request. */
@ -146,8 +178,9 @@ class unsubscribe_response : public server_response
} }
} }
public: unsubscribe_response(MQTTAsync_successData* rsp) {}
public:
/** /**
* Gets the reason codes from the server response. * Gets the reason codes from the server response.
* On an unsubscribe ack there is a reason code for each topic * On an unsubscribe ack there is a reason code for each topic

2
src/mqtt/token.h
View File

@ -63,10 +63,8 @@ public:
enum Type { enum Type {
CONNECT, CONNECT,
SUBSCRIBE, SUBSCRIBE,
SUBSCRIBE_MANY,
PUBLISH, PUBLISH,
UNSUBSCRIBE, UNSUBSCRIBE,
UNSUBSCRIBE_MANY,
DISCONNECT DISCONNECT
}; };

3
src/mqtt/types.h
View File

@ -99,7 +99,8 @@ enum ReasonCode {
CONNECTION_RATE_EXCEEDED = 159, CONNECTION_RATE_EXCEEDED = 159,
MAXIMUM_CONNECT_TIME = 160, MAXIMUM_CONNECT_TIME = 160,
SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED = 161, SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED = 161,
WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED = 162 WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED = 162,
MQTTPP_V3_CODE = 255 // This is not a protocol code; used internally by the library
}; };
///////////////////////////////////////////////////////////////////////////// /////////////////////////////////////////////////////////////////////////////

38
src/token.cpp
View File

@ -105,6 +105,9 @@ void token::on_failure5(void* context, MQTTAsync_failureData5* rsp)
// -------------------------------------------------------------------------- // --------------------------------------------------------------------------
// Object callbacks // Object callbacks
//
// The success callback for MQTT v3 connections
//
void token::on_success(MQTTAsync_successData* rsp) void token::on_success(MQTTAsync_successData* rsp)
{ {
::Log(TRACE_MIN, -1, "[cpp] on_success"); ::Log(TRACE_MIN, -1, "[cpp] on_success");
@ -117,15 +120,15 @@ void token::on_success(MQTTAsync_successData* rsp)
switch (type_) { switch (type_) {
case Type::CONNECT: case Type::CONNECT:
connRsp_ = std::unique_ptr<connect_response>(new connect_response(rsp)); connRsp_.reset(new connect_response(rsp));
break; break;
case Type::SUBSCRIBE: case Type::SUBSCRIBE:
subRsp_ = std::unique_ptr<subscribe_response>(new subscribe_response(0, rsp)); subRsp_.reset(new subscribe_response(nExpected_, rsp));
break; break;
case Type::SUBSCRIBE_MANY: case Type::UNSUBSCRIBE:
subRsp_ = std::unique_ptr<subscribe_response>(new subscribe_response(nExpected_, rsp)); unsubRsp_.reset(new unsubscribe_response(rsp));
break; break;
} }
} }
@ -142,6 +145,9 @@ void token::on_success(MQTTAsync_successData* rsp)
cli_->remove_token(this); cli_->remove_token(this);
} }
//
// The success callback for MQTT v5 connections
//
void token::on_success5(MQTTAsync_successData5* rsp) void token::on_success5(MQTTAsync_successData5* rsp)
{ {
::Log(TRACE_MIN, -1, "[cpp] on_success5"); ::Log(TRACE_MIN, -1, "[cpp] on_success5");
@ -154,17 +160,15 @@ void token::on_success5(MQTTAsync_successData5* rsp)
switch (type_) { switch (type_) {
case Type::CONNECT: case Type::CONNECT:
connRsp_ = std::unique_ptr<connect_response>(new connect_response(rsp)); connRsp_.reset(new connect_response(rsp));
break; break;
case Type::SUBSCRIBE: case Type::SUBSCRIBE:
case Type::SUBSCRIBE_MANY: subRsp_.reset(new subscribe_response(rsp));
subRsp_ = std::unique_ptr<subscribe_response>(new subscribe_response(rsp));
break; break;
case Type::UNSUBSCRIBE: case Type::UNSUBSCRIBE:
case Type::UNSUBSCRIBE_MANY: unsubRsp_.reset(new unsubscribe_response(rsp));
unsubRsp_ = std::unique_ptr<unsubscribe_response>(new unsubscribe_response(rsp));
break; break;
} }
} }
@ -180,6 +184,9 @@ void token::on_success5(MQTTAsync_successData5* rsp)
cli_->remove_token(this); cli_->remove_token(this);
} }
//
// The failure callback for MQTT v3 connections
//
void token::on_failure(MQTTAsync_failureData* rsp) void token::on_failure(MQTTAsync_failureData* rsp)
{ {
::Log(TRACE_MIN, -1, "[cpp] on_failure"); ::Log(TRACE_MIN, -1, "[cpp] on_failure");
@ -189,6 +196,10 @@ void token::on_failure(MQTTAsync_failureData* rsp)
if (rsp) { if (rsp) {
msgId_ = rsp->token; msgId_ = rsp->token;
rc_ = rsp->code; rc_ = rsp->code;
// HACK: For backward compatability with v3 connections
reasonCode_ = ReasonCode(MQTTPP_V3_CODE);
if (rsp->message) if (rsp->message)
errMsg_ = string(rsp->message); errMsg_ = string(rsp->message);
} }
@ -206,6 +217,9 @@ void token::on_failure(MQTTAsync_failureData* rsp)
cli_->remove_token(this); cli_->remove_token(this);
} }
//
// The failure callback for MQTT v5 connections
//
void token::on_failure5(MQTTAsync_failureData5* rsp) void token::on_failure5(MQTTAsync_failureData5* rsp)
{ {
::Log(TRACE_MIN, -1, "[cpp] on_failure"); ::Log(TRACE_MIN, -1, "[cpp] on_failure");
@ -241,7 +255,7 @@ void token::reset()
{ {
guard g(lock_); guard g(lock_);
complete_ = false; complete_ = false;
rc_ = 0; rc_ = MQTTASYNC_SUCCESS;
reasonCode_ = ReasonCode::SUCCESS; reasonCode_ = ReasonCode::SUCCESS;
errMsg_.clear(); errMsg_.clear();
} }
@ -270,7 +284,7 @@ connect_response token::get_connect_response() const
subscribe_response token::get_subscribe_response() const subscribe_response token::get_subscribe_response() const
{ {
if (type_ != Type::SUBSCRIBE && type_ != Type::SUBSCRIBE_MANY) if (type_ != Type::SUBSCRIBE)
throw bad_cast(); throw bad_cast();
unique_lock g(lock_); unique_lock g(lock_);
@ -285,7 +299,7 @@ subscribe_response token::get_subscribe_response() const
unsubscribe_response token::get_unsubscribe_response() const unsubscribe_response token::get_unsubscribe_response() const
{ {
if (type_ != Type::UNSUBSCRIBE && type_ != Type::UNSUBSCRIBE_MANY) if (type_ != Type::UNSUBSCRIBE)
throw bad_cast(); throw bad_cast();
unique_lock g(lock_); unique_lock g(lock_);

92
test/cppunit/async_client_test.h
View File

@ -131,13 +131,13 @@ public:
} }
void test_user_constructor_2_string_args_failure() { void test_user_constructor_2_string_args_failure() {
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::async_client cli { BAD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { BAD_SERVER_URI, CLIENT_ID };
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_PROTOCOL, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_PROTOCOL, return_code);
} }
void test_user_constructor_3_string_args() { void test_user_constructor_3_string_args() {
@ -199,17 +199,17 @@ public:
mqtt::will_options wo; mqtt::will_options wo;
wo.set_qos(BAD_QOS); // Invalid QoS causes connection failure wo.set_qos(BAD_QOS); // Invalid QoS causes connection failure
co.set_will(wo); co.set_will(wo);
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
token_conn = cli.connect(co); token_conn = cli.connect(co);
CPPUNIT_ASSERT(token_conn); CPPUNIT_ASSERT(token_conn);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT(nullptr == token_conn); CPPUNIT_ASSERT(nullptr == token_conn);
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, return_code);
} }
void test_connect_2_args() { void test_connect_2_args() {
@ -256,7 +256,7 @@ public:
token_conn->wait(); token_conn->wait();
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reasonCode = ex.get_reason_code(); reasonCode = ex.get_return_code();
} }
CPPUNIT_ASSERT(nullptr == token_conn); CPPUNIT_ASSERT(nullptr == token_conn);
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
@ -284,7 +284,7 @@ public:
tok->wait(); tok->wait();
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reasonCode = ex.get_reason_code(); reasonCode = ex.get_return_code();
} }
CPPUNIT_ASSERT(reasonCode != MQTTASYNC_SUCCESS); CPPUNIT_ASSERT(reasonCode != MQTTASYNC_SUCCESS);
} }
@ -328,16 +328,16 @@ public:
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_disconn; //{ nullptr }; mqtt::token_ptr token_disconn; //{ nullptr };
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
token_disconn = cli.disconnect(0); token_disconn = cli.disconnect(0);
CPPUNIT_ASSERT(token_disconn); CPPUNIT_ASSERT(token_disconn);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_disconnect_2_args() { void test_disconnect_2_args() {
@ -380,16 +380,16 @@ public:
mqtt::token_ptr token_disconn; //{ nullptr }; mqtt::token_ptr token_disconn; //{ nullptr };
mqtt::test::dummy_action_listener listener; mqtt::test::dummy_action_listener listener;
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
token_disconn = cli.disconnect(0, &CONTEXT, listener); token_disconn = cli.disconnect(0, &CONTEXT, listener);
CPPUNIT_ASSERT(token_disconn); CPPUNIT_ASSERT(token_disconn);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
//---------------------------------------------------------------------- //----------------------------------------------------------------------
@ -534,7 +534,7 @@ public:
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) }; mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) };
mqtt::delivery_token_ptr token_pub { cli.publish(msg) }; mqtt::delivery_token_ptr token_pub { cli.publish(msg) };
@ -542,9 +542,9 @@ public:
token_pub->wait_for(TIMEOUT); token_pub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_publish_4_args() { void test_publish_4_args() {
@ -573,7 +573,7 @@ public:
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) }; mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) };
mqtt::test::dummy_action_listener listener; mqtt::test::dummy_action_listener listener;
@ -582,9 +582,9 @@ public:
token_pub->wait_for(TIMEOUT); token_pub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_publish_5_args() { void test_publish_5_args() {
@ -672,16 +672,16 @@ public:
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::token_ptr token_sub { cli.subscribe(TOPIC, BAD_QOS) }; mqtt::token_ptr token_sub { cli.subscribe(TOPIC, BAD_QOS) };
CPPUNIT_ASSERT(token_sub); CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT); token_sub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_subscribe_single_topic_4_args() { void test_subscribe_single_topic_4_args() {
@ -709,7 +709,7 @@ public:
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::test::dummy_action_listener listener; mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_sub { cli.subscribe(TOPIC, BAD_QOS, &CONTEXT, listener) }; mqtt::token_ptr token_sub { cli.subscribe(TOPIC, BAD_QOS, &CONTEXT, listener) };
@ -717,9 +717,9 @@ public:
token_sub->wait_for(TIMEOUT); token_sub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_subscribe_many_topics_2_args() { void test_subscribe_many_topics_2_args() {
@ -750,19 +750,19 @@ public:
token_sub->wait_for(TIMEOUT); token_sub->wait_for(TIMEOUT);
} }
catch (const mqtt::exception& ex) { catch (const mqtt::exception& ex) {
//CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, ex.get_reason_code()); //CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, ex.get_return_code());
} }
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL) }; mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL) };
CPPUNIT_ASSERT(token_sub); CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT); token_sub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_subscribe_many_topics_4_args() { void test_subscribe_many_topics_4_args() {
@ -796,19 +796,19 @@ public:
cli.subscribe(TOPIC_COLL, BAD_QOS_COLL, &CONTEXT, listener)->wait_for(TIMEOUT); cli.subscribe(TOPIC_COLL, BAD_QOS_COLL, &CONTEXT, listener)->wait_for(TIMEOUT);
} }
catch (const mqtt::exception& ex) { catch (const mqtt::exception& ex) {
//CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, ex.get_reason_code()); //CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, ex.get_return_code());
} }
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL, &CONTEXT, listener) }; mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_sub); CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT); token_sub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
//---------------------------------------------------------------------- //----------------------------------------------------------------------
@ -838,16 +838,16 @@ public:
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC) }; mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC) };
CPPUNIT_ASSERT(token_unsub); CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT); token_unsub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_unsubscribe_single_topic_3_args() { void test_unsubscribe_single_topic_3_args() {
@ -875,7 +875,7 @@ public:
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::test::dummy_action_listener listener; mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC, &CONTEXT, listener) }; mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC, &CONTEXT, listener) };
@ -883,9 +883,9 @@ public:
token_unsub->wait_for(TIMEOUT); token_unsub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_unsubscribe_many_topics_1_arg() { void test_unsubscribe_many_topics_1_arg() {
@ -911,16 +911,16 @@ public:
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL) }; mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL) };
CPPUNIT_ASSERT(token_unsub); CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT); token_unsub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_unsubscribe_many_topics_3_args() { void test_unsubscribe_many_topics_3_args() {
@ -949,16 +949,16 @@ public:
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::test::dummy_action_listener listener; mqtt::test::dummy_action_listener listener;
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL, &CONTEXT, listener) }; mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_unsub); CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT); token_unsub->wait_for(TIMEOUT);
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
}; };

974
test/cppunit/async_client_v3_test.h Normal file
View File

@ -0,0 +1,974 @@
// async_client_v3_test.h
// Unit tests for the MQTT v3 async_client class in the Paho MQTT C++
// library.
/*******************************************************************************
* Copyright (c) 2017 Guilherme M. Ferreira <guilherme.maciel.ferreira@gmail.com>
* Copyright (c) 2019 Frank Pagliughi <fpagliughi@mindspring.com>
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Guilherme M. Ferreira - initial implementation and documentation
* Frank Pagliughi - updates
*******************************************************************************/
#ifndef __mqtt_async_client_v3_test_h
#define __mqtt_async_client_v3_test_h
#include <stdexcept>
#include <vector>
#include <cppunit/ui/text/TestRunner.h>
#include <cppunit/extensions/HelperMacros.h>
#include "mqtt/iasync_client.h"
#include "mqtt/async_client.h"
#include "dummy_client_persistence.h"
#include "dummy_action_listener.h"
#include "dummy_callback.h"
namespace mqtt {
/////////////////////////////////////////////////////////////////////////////
class async_client_v3_test : public CppUnit::TestFixture
{
CPPUNIT_TEST_SUITE( async_client_test );
CPPUNIT_TEST( test_user_constructor_2_string_args );
CPPUNIT_TEST( test_user_constructor_3_string_args );
CPPUNIT_TEST( test_user_constructor_3_args );
CPPUNIT_TEST( test_connect_0_arg );
CPPUNIT_TEST( test_connect_1_arg );
CPPUNIT_TEST( test_connect_1_arg_failure );
CPPUNIT_TEST( test_connect_2_args );
CPPUNIT_TEST( test_connect_3_args );
CPPUNIT_TEST( test_connect_3_args_failure );
CPPUNIT_TEST( test_connect_uninitialized_ssl );
CPPUNIT_TEST( test_disconnect_0_arg );
CPPUNIT_TEST( test_disconnect_1_arg );
CPPUNIT_TEST( test_disconnect_1_arg_failure );
CPPUNIT_TEST( test_disconnect_2_args );
CPPUNIT_TEST( test_disconnect_3_args );
CPPUNIT_TEST( test_disconnect_3_args_failure );
CPPUNIT_TEST( test_get_pending_delivery_token );
CPPUNIT_TEST( test_get_pending_delivery_tokens );
CPPUNIT_TEST( test_publish_2_args );
CPPUNIT_TEST( test_publish_2_args_failure );
CPPUNIT_TEST( test_publish_4_args );
CPPUNIT_TEST( test_publish_4_args_failure );
CPPUNIT_TEST( test_publish_5_args );
CPPUNIT_TEST( test_publish_7_args );
CPPUNIT_TEST( test_set_callback );
CPPUNIT_TEST( test_subscribe_single_topic_2_args );
CPPUNIT_TEST( test_subscribe_single_topic_2_args_failure );
CPPUNIT_TEST( test_subscribe_single_topic_4_args );
CPPUNIT_TEST( test_subscribe_single_topic_4_args_failure );
CPPUNIT_TEST( test_subscribe_many_topics_2_args );
CPPUNIT_TEST( test_subscribe_many_topics_2_args_failure );
CPPUNIT_TEST( test_subscribe_many_topics_4_args );
CPPUNIT_TEST( test_subscribe_many_topics_4_args_failure );
CPPUNIT_TEST( test_unsubscribe_single_topic_1_arg );
CPPUNIT_TEST( test_unsubscribe_single_topic_1_arg_failure );
CPPUNIT_TEST( test_unsubscribe_single_topic_3_args );
CPPUNIT_TEST( test_unsubscribe_single_topic_3_args_failure );
CPPUNIT_TEST( test_unsubscribe_many_topics_1_arg );
CPPUNIT_TEST( test_unsubscribe_many_topics_1_arg_failure );
CPPUNIT_TEST( test_unsubscribe_many_topics_3_args );
CPPUNIT_TEST( test_unsubscribe_many_topics_3_args_failure );
CPPUNIT_TEST_SUITE_END();
// NOTE: This test case requires network access. It uses one of
// the public available MQTT brokers
#if defined(TEST_EXTERNAL_SERVER)
const std::string GOOD_SERVER_URI { "tcp://mqtt.eclipse.org:1883" };
#else
const std::string GOOD_SERVER_URI { "tcp://localhost:1883" };
const std::string GOOD_SSL_SERVER_URI { "ssl://localhost:18885" };
#endif
const std::string BAD_SERVER_URI { "one://invalid.address" };
const std::string CLIENT_ID { "" }; // { "async_client_unit_test" };
const std::string PERSISTENCE_DIR { "/tmp" };
const std::string TOPIC { "TOPIC" };
const int GOOD_QOS { 0 };
const int BAD_QOS { 3 };
const_string_collection_ptr TOPIC_COLL { string_collection::create({ "TOPIC0", "TOPIC1", "TOPIC2" }) };
mqtt::iasync_client::qos_collection GOOD_QOS_COLL { 0, 1, 2 };
mqtt::iasync_client::qos_collection BAD_QOS_COLL { BAD_QOS, 1, 2 };
const std::string PAYLOAD { "PAYLOAD" };
const int TIMEOUT { 1000 };
int CONTEXT { 4 };
mqtt::test::dummy_action_listener listener;
const bool RETAINED { false };
public:
void setUp() {}
void tearDown() {}
//----------------------------------------------------------------------
// Test constructors async_client::async_client()
//----------------------------------------------------------------------
void test_user_constructor_2_string_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(GOOD_SERVER_URI, cli.get_server_uri());
CPPUNIT_ASSERT_EQUAL(CLIENT_ID, cli.get_client_id());
}
void test_user_constructor_2_string_args_failure() {
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::async_client cli { BAD_SERVER_URI, CLIENT_ID };
} catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_PROTOCOL, reason_code);
}
void test_user_constructor_3_string_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID, PERSISTENCE_DIR };
CPPUNIT_ASSERT_EQUAL(GOOD_SERVER_URI, cli.get_server_uri());
CPPUNIT_ASSERT_EQUAL(CLIENT_ID, cli.get_client_id());
}
void test_user_constructor_3_args() {
mqtt::test::dummy_client_persistence cp;
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID, &cp };
CPPUNIT_ASSERT_EQUAL(GOOD_SERVER_URI, cli.get_server_uri());
CPPUNIT_ASSERT_EQUAL(CLIENT_ID, cli.get_client_id());
mqtt::async_client cli_no_persistence { GOOD_SERVER_URI, CLIENT_ID, nullptr };
CPPUNIT_ASSERT_EQUAL(GOOD_SERVER_URI, cli_no_persistence.get_server_uri());
CPPUNIT_ASSERT_EQUAL(CLIENT_ID, cli_no_persistence.get_client_id());
}
//----------------------------------------------------------------------
// Test async_client::connect()
//----------------------------------------------------------------------
void test_connect_0_arg() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
try {
mqtt::token_ptr token_conn = cli.connect();
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
}
catch (const std::exception& exc) {
CPPUNIT_FAIL(std::string("Connection failure: ")+exc.what());
}
}
void test_connect_1_arg() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::connect_options co;
mqtt::token_ptr token_conn { cli.connect(co) };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
}
void test_connect_1_arg_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn; //{ nullptr };
mqtt::connect_options co;
mqtt::will_options wo;
wo.set_qos(BAD_QOS); // Invalid QoS causes connection failure
co.set_will(wo);
int reason_code = MQTTASYNC_SUCCESS;
try {
token_conn = cli.connect(co);
CPPUNIT_ASSERT(token_conn);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT(nullptr == token_conn);
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, reason_code);
}
void test_connect_2_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_conn { cli.connect(&CONTEXT, listener) };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_conn->get_user_context()));
CPPUNIT_ASSERT(listener.on_success_called);
}
void test_connect_3_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::connect_options co;
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_conn { cli.connect(co, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_conn->get_user_context()));
CPPUNIT_ASSERT(listener.on_success_called);
}
void test_connect_3_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn; //{ nullptr };
mqtt::connect_options co;
mqtt::will_options wo;
wo.set_qos(BAD_QOS); // Invalid QoS causes connection failure
co.set_will(wo);
mqtt::test::dummy_action_listener listener;
int reasonCode = MQTTASYNC_SUCCESS;
try {
token_conn = cli.connect(co, &CONTEXT, listener);
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
}
catch (mqtt::exception& ex) {
reasonCode = ex.get_reason_code();
}
CPPUNIT_ASSERT(nullptr == token_conn);
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, reasonCode);
// TODO Why listener.on_failure() is not called?
//CPPUNIT_ASSERT(listener.on_failure_called);
}
// An improperly initialized SSL connect request should fail gracefully
void test_connect_uninitialized_ssl() {
int reasonCode = MQTTASYNC_SUCCESS;
try {
// Compiled against a non-SSL library should throw here.
mqtt::async_client cli { GOOD_SSL_SERVER_URI, CLIENT_ID };
mqtt::connect_options opts;
opts.set_keep_alive_interval(10);
opts.set_clean_session(true);
// Note that we're not setting SSL options.
mqtt::token_ptr tok;
// Compiled against the SSL library should throw here
tok = cli.connect(opts);
tok->wait();
}
catch (mqtt::exception& ex) {
reasonCode = ex.get_reason_code();
}
CPPUNIT_ASSERT(reasonCode != MQTTASYNC_SUCCESS);
}
//----------------------------------------------------------------------
// Test async_client::disconnect()
//----------------------------------------------------------------------
void test_disconnect_0_arg() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_disconnect_1_arg() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::token_ptr token_disconn { cli.disconnect(0) };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_disconnect_1_arg_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_disconn; //{ nullptr };
int reason_code = MQTTASYNC_SUCCESS;
try {
token_disconn = cli.disconnect(0);
CPPUNIT_ASSERT(token_disconn);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_disconnect_2_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_disconn { cli.disconnect(&CONTEXT, listener) };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_disconn->get_user_context()));
}
void test_disconnect_3_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_disconn { cli.disconnect(0, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_disconn->get_user_context()));
}
void test_disconnect_3_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_disconn; //{ nullptr };
mqtt::test::dummy_action_listener listener;
int reason_code = MQTTASYNC_SUCCESS;
try {
token_disconn = cli.disconnect(0, &CONTEXT, listener);
CPPUNIT_ASSERT(token_disconn);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
//----------------------------------------------------------------------
// Test async_client::get_pending_delivery_token()
//----------------------------------------------------------------------
void test_get_pending_delivery_token() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(0, GOOD_QOS_COLL[0]);
CPPUNIT_ASSERT_EQUAL(1, GOOD_QOS_COLL[1]);
CPPUNIT_ASSERT_EQUAL(2, GOOD_QOS_COLL[2]);
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
// NOTE: async_client::publish() is the only method that adds
// delivery_token via async_client::add_token(delivery_token_ptr tok).
// The other functions add token async_client::add_token(token_ptr tok).
mqtt::delivery_token_ptr token_pub; // { nullptr };
mqtt::delivery_token_ptr token_pending; // { nullptr };
// NOTE: message IDs are 16-bit numbers sequentially incremented, from
// 1 to 65535 (MAX_MSG_ID). See MQTTAsync_assignMsgId() at Paho MQTT C.
int message_id = 1;
// NOTE: All of the MQTT messages that require a response/acknowledge
// should have a non-zero 16-bit message ID. This mainly applies to a
// message with QOS=1 or QOS=2. The C++ library keeps a collection of
// pointers to token objects for all of these messages that are in
// flight. When the acknowledge comes back from the broker, the C++
// library can look up the token from the msgID and signal it, indicating
// completion.
// Messages with QOS=2 are kept by the library
mqtt::message_ptr msg2 { mqtt::message::create(TOPIC, PAYLOAD, GOOD_QOS_COLL[2], RETAINED) };
token_pub = cli.publish(msg2);
CPPUNIT_ASSERT(token_pub);
token_pending = cli.get_pending_delivery_token(message_id++);
CPPUNIT_ASSERT(token_pending);
// Messages with QOS=1 are kept by the library
mqtt::message_ptr msg1 { mqtt::message::create(TOPIC, PAYLOAD, GOOD_QOS_COLL[1], RETAINED) };
token_pub = cli.publish(msg1);
CPPUNIT_ASSERT(token_pub);
token_pending = cli.get_pending_delivery_token(message_id++);
CPPUNIT_ASSERT(token_pending);
// NOTE: Messages with QOS=0 are fire-and-forget. These just get sent
// to the broker without any tracking. Their tokens are signaled as
// "complete" in the send function (by the calling thread). So, as
// soon as send returns, the message is considered completed. These
// have a msgID that is always zero.
// Messages with QOS=0 are NOT kept by the library
mqtt::message_ptr msg0 { mqtt::message::create(TOPIC, PAYLOAD, GOOD_QOS_COLL[0], RETAINED) };
token_pub = cli.publish(msg0);
CPPUNIT_ASSERT(token_pub);
token_pending = cli.get_pending_delivery_token(message_id++);
CPPUNIT_ASSERT(!token_pending);
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_get_pending_delivery_tokens() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(0, GOOD_QOS_COLL[0]);
CPPUNIT_ASSERT_EQUAL(1, GOOD_QOS_COLL[1]);
CPPUNIT_ASSERT_EQUAL(2, GOOD_QOS_COLL[2]);
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::delivery_token_ptr token_pub; // { nullptr };
// NOTE: async_client::publish() is the only method that adds
// delivery_token via async_client::add_token(delivery_token_ptr tok).
// The other functions add token async_client::add_token(token_ptr tok).
// Messages with QOS=0 are NOT kept by the library
mqtt::message_ptr msg0 { mqtt::message::create(TOPIC, PAYLOAD, GOOD_QOS_COLL[0], RETAINED) };
token_pub = cli.publish(msg0);
CPPUNIT_ASSERT(token_pub);
// Messages with QOS=1 are kept by the library
mqtt::message_ptr msg1 { mqtt::message::create(TOPIC, PAYLOAD, GOOD_QOS_COLL[1], RETAINED) };
token_pub = cli.publish(msg1);
CPPUNIT_ASSERT(token_pub);
// Messages with QOS=2 are kept by the library
mqtt::message_ptr msg2 { mqtt::message::create(TOPIC, PAYLOAD, GOOD_QOS_COLL[2], RETAINED) };
token_pub = cli.publish(msg2);
CPPUNIT_ASSERT(token_pub);
// NOTE: Only tokens for messages with QOS=1 and QOS=2 are kept. That's
// why the vector's size does not account for QOS=0 message tokens
std::vector<mqtt::delivery_token_ptr> tokens_pending { cli.get_pending_delivery_tokens() };
CPPUNIT_ASSERT_EQUAL(2, static_cast<int>(tokens_pending.size()));
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
//----------------------------------------------------------------------
// Test async_client::publish()
//----------------------------------------------------------------------
void test_publish_2_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) };
mqtt::delivery_token_ptr token_pub { cli.publish(msg) };
CPPUNIT_ASSERT(token_pub);
token_pub->wait_for(TIMEOUT);
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_publish_2_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) };
mqtt::delivery_token_ptr token_pub { cli.publish(msg) };
CPPUNIT_ASSERT(token_pub);
token_pub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_publish_4_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) };
mqtt::test::dummy_action_listener listener;
mqtt::delivery_token_ptr token_pub { cli.publish(msg, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_pub);
token_pub->wait_for(TIMEOUT);
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_pub->get_user_context()));
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_publish_4_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) };
mqtt::test::dummy_action_listener listener;
mqtt::delivery_token_ptr token_pub { cli.publish(msg, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_pub);
token_pub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_publish_5_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
const void* payload { PAYLOAD.data() };
const size_t payload_size { PAYLOAD.size() };
mqtt::delivery_token_ptr token_pub { cli.publish(TOPIC, payload, payload_size, GOOD_QOS, RETAINED) };
CPPUNIT_ASSERT(token_pub);
token_pub->wait_for(TIMEOUT);
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_publish_7_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
const void* payload { PAYLOAD.c_str() };
const size_t payload_size { PAYLOAD.size() };
mqtt::test::dummy_action_listener listener;
mqtt::delivery_token_ptr token_pub { cli.publish(TOPIC, payload, payload_size, GOOD_QOS, RETAINED, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_pub);
token_pub->wait_for(TIMEOUT);
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_pub->get_user_context()));
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
//----------------------------------------------------------------------
// Test async_client::set_callback()
//----------------------------------------------------------------------
void test_set_callback() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::test::dummy_callback cb;
cli.set_callback(cb);
//CPPUNIT_ASSERT(cb.delivery_complete_called);
}
//----------------------------------------------------------------------
// Test async_client::subscribe()
//----------------------------------------------------------------------
void test_subscribe_single_topic_2_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::token_ptr token_sub { cli.subscribe(TOPIC, GOOD_QOS) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_subscribe_single_topic_2_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::token_ptr token_sub { cli.subscribe(TOPIC, BAD_QOS) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_subscribe_single_topic_4_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_sub { cli.subscribe(TOPIC, GOOD_QOS, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_sub->get_user_context()));
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_subscribe_single_topic_4_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_sub { cli.subscribe(TOPIC, BAD_QOS, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_subscribe_many_topics_2_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
cli.connect()->wait();
CPPUNIT_ASSERT(cli.is_connected());
try {
cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL)->wait_for(TIMEOUT);
}
catch (const mqtt::exception& exc) {
CPPUNIT_FAIL(exc.what());
}
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_subscribe_many_topics_2_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
try {
mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, BAD_QOS_COLL) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
}
catch (const mqtt::exception& ex) {
//CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, ex.get_reason_code());
}
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_subscribe_many_topics_4_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_sub->get_user_context()));
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_subscribe_many_topics_4_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::test::dummy_action_listener listener;
try {
cli.subscribe(TOPIC_COLL, BAD_QOS_COLL, &CONTEXT, listener)->wait_for(TIMEOUT);
}
catch (const mqtt::exception& ex) {
//CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, ex.get_reason_code());
}
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::token_ptr token_sub { cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_sub);
token_sub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
//----------------------------------------------------------------------
// Test async_client::unsubscribe()
//----------------------------------------------------------------------
void test_unsubscribe_single_topic_1_arg() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_unsubscribe_single_topic_1_arg_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_unsubscribe_single_topic_3_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_unsub->get_user_context()));
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_unsubscribe_single_topic_3_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_unsubscribe_many_topics_1_arg() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_unsubscribe_many_topics_1_arg_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
void test_unsubscribe_many_topics_3_args() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::token_ptr token_conn { cli.connect() };
CPPUNIT_ASSERT(token_conn);
token_conn->wait();
CPPUNIT_ASSERT(cli.is_connected());
mqtt::test::dummy_action_listener listener;
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
CPPUNIT_ASSERT_EQUAL(CONTEXT, *static_cast<int*>(token_unsub->get_user_context()));
mqtt::token_ptr token_disconn { cli.disconnect() };
CPPUNIT_ASSERT(token_disconn);
token_disconn->wait();
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
}
void test_unsubscribe_many_topics_3_args_failure() {
mqtt::async_client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
mqtt::test::dummy_action_listener listener;
int reason_code = MQTTASYNC_SUCCESS;
try {
mqtt::token_ptr token_unsub { cli.unsubscribe(TOPIC_COLL, &CONTEXT, listener) };
CPPUNIT_ASSERT(token_unsub);
token_unsub->wait_for(TIMEOUT);
}
catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code();
}
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code);
}
};
/////////////////////////////////////////////////////////////////////////////
// end namespace mqtt
}
#endif // __mqtt_async_client_v3_test_h

62
test/cppunit/client_test.h
View File

@ -83,7 +83,7 @@ class client_test : public CppUnit::TestFixture
// NOTE: This test case requires network access. It uses one of // NOTE: This test case requires network access. It uses one of
// the public available MQTT brokers // the public available MQTT brokers
#if defined(TEST_EXTERNAL_SERVER) #if defined(TEST_EXTERNAL_SERVER)
const std::string GOOD_SERVER_URI { "tcp://m2m.eclipse.org:1883" }; const std::string GOOD_SERVER_URI { "tcp://mqtt.eclipse.org:1883" };
#else #else
const std::string GOOD_SERVER_URI { "tcp://localhost:1883" }; const std::string GOOD_SERVER_URI { "tcp://localhost:1883" };
#endif #endif
@ -118,13 +118,13 @@ public:
} }
void test_user_constructor_2_string_args_failure() { void test_user_constructor_2_string_args_failure() {
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::client cli { BAD_SERVER_URI, CLIENT_ID }; mqtt::client cli { BAD_SERVER_URI, CLIENT_ID };
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_PROTOCOL, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_PROTOCOL, return_code);
} }
void test_user_constructor_3_string_args() { void test_user_constructor_3_string_args() {
@ -176,14 +176,14 @@ public:
mqtt::will_options wo; mqtt::will_options wo;
wo.set_qos(BAD_QOS); // Invalid QoS causes connection failure wo.set_qos(BAD_QOS); // Invalid QoS causes connection failure
co.set_will(wo); co.set_will(wo);
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.connect(co); cli.connect(co);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_BAD_QOS, return_code);
} }
//---------------------------------------------------------------------- //----------------------------------------------------------------------
@ -216,14 +216,14 @@ public:
mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.disconnect(0); cli.disconnect(0);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
//---------------------------------------------------------------------- //----------------------------------------------------------------------
@ -295,14 +295,14 @@ public:
mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) }; mqtt::message_ptr msg { mqtt::message::create(TOPIC, PAYLOAD) };
cli.publish(msg); cli.publish(msg);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_publish_reference_2_args() { void test_publish_reference_2_args() {
@ -367,13 +367,13 @@ public:
mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.subscribe(TOPIC); cli.subscribe(TOPIC);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_subscribe_single_topic_2_args() { void test_subscribe_single_topic_2_args() {
@ -393,13 +393,13 @@ public:
mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.subscribe(TOPIC, BAD_QOS); cli.subscribe(TOPIC, BAD_QOS);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_subscribe_many_topics_1_arg() { void test_subscribe_many_topics_1_arg() {
@ -419,13 +419,13 @@ public:
mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.subscribe(TOPIC_COLL); cli.subscribe(TOPIC_COLL);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_subscribe_many_topics_2_args() { void test_subscribe_many_topics_2_args() {
@ -449,13 +449,13 @@ public:
cli.subscribe(TOPIC_COLL, BAD_QOS_COLL); cli.subscribe(TOPIC_COLL, BAD_QOS_COLL);
} catch (std::invalid_argument& ex) {} } catch (std::invalid_argument& ex) {}
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL); cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
//---------------------------------------------------------------------- //----------------------------------------------------------------------
@ -479,13 +479,13 @@ public:
mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.unsubscribe(TOPIC); cli.unsubscribe(TOPIC);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
void test_unsubscribe_many_topics_1_arg() { void test_unsubscribe_many_topics_1_arg() {
@ -505,13 +505,13 @@ public:
mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID }; mqtt::client cli { GOOD_SERVER_URI, CLIENT_ID };
CPPUNIT_ASSERT_EQUAL(false, cli.is_connected()); CPPUNIT_ASSERT_EQUAL(false, cli.is_connected());
int reason_code = MQTTASYNC_SUCCESS; int return_code = MQTTASYNC_SUCCESS;
try { try {
cli.unsubscribe(TOPIC_COLL); cli.unsubscribe(TOPIC_COLL);
} catch (mqtt::exception& ex) { } catch (mqtt::exception& ex) {
reason_code = ex.get_reason_code(); return_code = ex.get_return_code();
} }
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, reason_code); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_DISCONNECTED, return_code);
} }
}; };

14
test/cppunit/exception_test.h
View File

@ -35,7 +35,7 @@ class exception_test : public CppUnit::TestFixture
CPPUNIT_TEST( test_user_constructor ); CPPUNIT_TEST( test_user_constructor );
CPPUNIT_TEST( test_get_message ); CPPUNIT_TEST( test_get_message );
CPPUNIT_TEST( test_get_reason_code ); CPPUNIT_TEST( test_get_return_code );
CPPUNIT_TEST( test_to_str ); CPPUNIT_TEST( test_to_str );
CPPUNIT_TEST( test_what ); CPPUNIT_TEST( test_what );
@ -53,13 +53,13 @@ public:
void test_user_constructor() { void test_user_constructor() {
mqtt::exception ex1(MQTTASYNC_FAILURE); mqtt::exception ex1(MQTTASYNC_FAILURE);
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex1.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex1.get_return_code());
mqtt::exception ex2(MQTTASYNC_PERSISTENCE_ERROR); mqtt::exception ex2(MQTTASYNC_PERSISTENCE_ERROR);
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_PERSISTENCE_ERROR, ex2.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_PERSISTENCE_ERROR, ex2.get_return_code());
mqtt::exception ex3(MQTTASYNC_OPERATION_INCOMPLETE); mqtt::exception ex3(MQTTASYNC_OPERATION_INCOMPLETE);
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_OPERATION_INCOMPLETE, ex3.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_OPERATION_INCOMPLETE, ex3.get_return_code());
} }
// ---------------------------------------------------------------------- // ----------------------------------------------------------------------
@ -72,12 +72,12 @@ public:
} }
// ---------------------------------------------------------------------- // ----------------------------------------------------------------------
// Test get_reason_code() // Test get_return_code()
// ---------------------------------------------------------------------- // ----------------------------------------------------------------------
void test_get_reason_code() { void test_get_return_code() {
mqtt::exception ex1(MQTTASYNC_FAILURE); mqtt::exception ex1(MQTTASYNC_FAILURE);
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex1.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex1.get_return_code());
} }
// ---------------------------------------------------------------------- // ----------------------------------------------------------------------

2
test/cppunit/test.cpp
View File

@ -20,6 +20,7 @@
#include "buffer_ref_test.h" #include "buffer_ref_test.h"
#include "string_collection_test.h" #include "string_collection_test.h"
#include "async_client_test.h" #include "async_client_test.h"
#include "async_client_v3_test.h"
#include "client_test.h" #include "client_test.h"
#include "message_test.h" #include "message_test.h"
#include "will_options_test.h" #include "will_options_test.h"
@ -59,6 +60,7 @@ int main(int argc, char* argv[])
CPPUNIT_TEST_SUITE_REGISTRATION( mqtt::exception_test ); CPPUNIT_TEST_SUITE_REGISTRATION( mqtt::exception_test );
CPPUNIT_TEST_SUITE_REGISTRATION( mqtt::async_client_test ); CPPUNIT_TEST_SUITE_REGISTRATION( mqtt::async_client_test );
CPPUNIT_TEST_SUITE_REGISTRATION( mqtt::async_client_v3_test );
CPPUNIT_TEST_SUITE_REGISTRATION( mqtt::client_test ); CPPUNIT_TEST_SUITE_REGISTRATION( mqtt::client_test );
TextUi::TestRunner runner; TextUi::TestRunner runner;

9
test/cppunit/token_test.h
View File

@ -143,6 +143,7 @@ public:
MQTTAsync_successData data = { MQTTAsync_successData data = {
.token = MESSAGE_ID, .token = MESSAGE_ID,
}; };
data.alt.connect.serverURI = const_cast<char*>("tcp://some_server.com");
CPPUNIT_ASSERT_EQUAL(false, tok.is_complete()); CPPUNIT_ASSERT_EQUAL(false, tok.is_complete());
token::on_success(&tok, &data); token::on_success(&tok, &data);
@ -300,7 +301,7 @@ public:
CPPUNIT_FAIL("token::wait() should throw on failure"); CPPUNIT_FAIL("token::wait() should throw on failure");
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_return_code());
} }
try { try {
@ -308,7 +309,7 @@ public:
CPPUNIT_FAIL("token::try_wait() should throw on failure"); CPPUNIT_FAIL("token::try_wait() should throw on failure");
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_return_code());
} }
try { try {
@ -316,7 +317,7 @@ public:
CPPUNIT_FAIL("token::wait_for() should throw on failure"); CPPUNIT_FAIL("token::wait_for() should throw on failure");
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_return_code());
} }
try { try {
@ -324,7 +325,7 @@ public:
CPPUNIT_FAIL("token::wait_until() should throw on failure"); CPPUNIT_FAIL("token::wait_until() should throw on failure");
} }
catch (mqtt::exception& ex) { catch (mqtt::exception& ex) {
CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_reason_code()); CPPUNIT_ASSERT_EQUAL(MQTTASYNC_FAILURE, ex.get_return_code());
} }
} }