Loading...
Searching...
No Matches
async_client.h
Go to the documentation of this file.
/*******************************************************************************
* Copyright (c) 2013-2025 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 v2.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v20.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Frank Pagliughi - initial implementation and documentation
* Frank Pagliughi - MQTT v5 support
*******************************************************************************/
#ifndef __mqtt_async_client_h
#define __mqtt_async_client_h
#include <functional>
#include <list>
#include <memory>
#include <stdexcept>
#include <tuple>
#include <vector>
#include "MQTTAsync.h"
#include "mqtt/callback.h"
#include "mqtt/create_options.h"
#include "mqtt/delivery_token.h"
#include "mqtt/event.h"
#include "mqtt/exception.h"
#include "mqtt/iaction_listener.h"
#include "mqtt/iasync_client.h"
#include "mqtt/iclient_persistence.h"
#include "mqtt/message.h"
#include "mqtt/properties.h"
#include "mqtt/string_collection.h"
#include "mqtt/thread_queue.h"
#include "mqtt/token.h"
#include "mqtt/types.h"
namespace mqtt {
// OBSOLETE: The legacy constants that lacked the "PAHO_MQTTPP_" prefix
// clashed with #define's from other libraries and will be removed at the
// next major version upgrade.
#if defined(PAHO_MQTTPP_VERSIONS)
const uint32_t PAHO_MQTTPP_VERSION = 0x01050002;
const string PAHO_MQTTPP_VERSION_STR("Paho MQTT C++ (mqttpp) v. 1.5.2");
const string PAHO_MQTTPP_COPYRIGHT("Copyright (c) 2013-2025 Frank Pagliughi");
#else
const uint32_t VERSION = 0x01050002;
const string VERSION_STR("Paho MQTT C++ (mqttpp) v. 1.5.2");
const string COPYRIGHT("Copyright (c) 2013-2025 Frank Pagliughi");
#endif
class async_client : public virtual iasync_client
{
public:
using ptr_t = std::shared_ptr<async_client>;
using consumer_queue_type = std::unique_ptr<thread_queue<event>>;
using message_handler = std::function<void(const_message_ptr)>;
using connection_handler = std::function<void(const string& cause)>;
using disconnected_handler = std::function<void(const properties&, ReasonCode)>;
using update_connection_handler = std::function<bool(connect_data&)>;
private:
using guard = std::unique_lock<std::mutex>;
using unique_lock = std::unique_lock<std::mutex>;
mutable std::mutex lock_;
MQTTAsync cli_;
const create_options createOpts_;
int mqttVersion_;
std::unique_ptr<MQTTClient_persistence> persist_{};
callback* userCallback_{};
connection_handler connHandler_;
connection_handler connLostHandler_;
disconnected_handler disconnectedHandler_;
update_connection_handler updateConnectionHandler_;
message_handler msgHandler_;
connect_options connOpts_;
token_ptr connTok_;
std::list<token_ptr> pendingTokens_;
std::list<delivery_token_ptr> pendingDeliveryTokens_;
consumer_queue_type que_;
static void on_connected(void* context, char* cause);
static void on_connection_lost(void* context, char* cause);
static void on_disconnected(
void* context, MQTTProperties* cprops, MQTTReasonCodes reasonCode
);
static int on_message_arrived(
void* context, char* topicName, int topicLen, MQTTAsync_message* msg
);
static void on_delivery_complete(void* context, MQTTAsync_token tok);
static int on_update_connection(void* context, MQTTAsync_connectData* cdata);
friend class token;
virtual void add_token(token_ptr tok);
virtual void add_token(delivery_token_ptr tok);
virtual void remove_token(token* tok) override;
virtual void remove_token(token_ptr tok) { remove_token(tok.get()); }
void remove_token(delivery_token_ptr tok) { remove_token(tok.get()); }
async_client() = delete;
async_client(const async_client&) = delete;
async_client& operator=(const async_client&) = delete;
static void check_ret(int rc) {
if (rc != MQTTASYNC_SUCCESS)
throw exception(rc);
}
void create();
public:
explicit async_client(
const string& serverURI, const string& clientId = string{},
const persistence_type& persistence = NO_PERSISTENCE
)
: createOpts_{serverURI, clientId, persistence} {
create();
}
async_client(
const string& serverURI, const string& clientId, int maxBufferedMessages,
const persistence_type& persistence = NO_PERSISTENCE
)
: createOpts_{serverURI, clientId, maxBufferedMessages, persistence} {
create();
}
async_client(
const string& serverURI, const string& clientId, const create_options& opts,
const persistence_type& persistence = NO_PERSISTENCE
)
: createOpts_{serverURI, clientId, opts, persistence} {
create();
}
async_client(const create_options& opts) : createOpts_{opts} { create(); }
~async_client() override;
void set_callback(callback& cb) override;
void disable_callbacks() override;
void set_connected_handler(connection_handler cb) /*override*/;
void set_connection_lost_handler(connection_handler cb) /*override*/;
void set_disconnected_handler(disconnected_handler cb) /*override*/;
void set_message_callback(message_handler cb) /*override*/;
void set_update_connection_handler(update_connection_handler cb);
token_ptr connect() override;
token_ptr connect(connect_options options) override;
token_ptr connect(
connect_options options, void* userContext, iaction_listener& cb
) override;
token_ptr connect(void* userContext, iaction_listener& cb) override {
return connect(connect_options{}, userContext, cb);
}
token_ptr reconnect() override;
token_ptr disconnect() override { return disconnect(disconnect_options()); }
token_ptr disconnect(disconnect_options opts) override;
token_ptr disconnect(int timeout) override {
return disconnect(disconnect_options(timeout));
}
template <class Rep, class Period>
token_ptr disconnect(const std::chrono::duration<Rep, Period>& timeout) {
// TODO: check range
return disconnect((int)to_milliseconds_count(timeout));
}
token_ptr disconnect(int timeout, void* userContext, iaction_listener& cb) override;
template <class Rep, class Period>
token_ptr disconnect(
const std::chrono::duration<Rep, Period>& timeout, void* userContext,
iaction_listener& cb
) {
// TODO: check range
return disconnect((int)to_milliseconds_count(timeout), userContext, cb);
}
token_ptr disconnect(void* userContext, iaction_listener& cb) override {
return disconnect(0L, userContext, cb);
}
delivery_token_ptr get_pending_delivery_token(int msgID) const override;
std::vector<delivery_token_ptr> get_pending_delivery_tokens() const override;
string get_client_id() const override { return createOpts_.get_client_id(); }
string get_server_uri() const override { return createOpts_.get_server_uri(); }
int mqtt_version() const noexcept { return mqttVersion_; }
connect_options get_connect_options() const {
guard g(lock_);
return connOpts_;
}
bool is_connected() const override { return to_bool(MQTTAsync_isConnected(cli_)); }
delivery_token_ptr publish(
string_ref topic, const void* payload, size_t n, int qos, bool retained,
const properties& props = properties()
) override;
delivery_token_ptr publish(string_ref topic, const void* payload, size_t n) override {
return publish(
std::move(topic), payload, n, message::DFLT_QOS, message::DFLT_RETAINED
);
}
delivery_token_ptr publish(
string_ref topic, binary_ref payload, int qos, bool retained,
const properties& props = properties()
) override;
delivery_token_ptr publish(string_ref topic, binary_ref payload) override {
return publish(
std::move(topic), std::move(payload), message::DFLT_QOS, message::DFLT_RETAINED
);
}
delivery_token_ptr publish(
string_ref topic, const void* payload, size_t n, int qos, bool retained,
void* userContext, iaction_listener& cb
) override;
delivery_token_ptr publish(const_message_ptr msg) override;
delivery_token_ptr publish(
const_message_ptr msg, void* userContext, iaction_listener& cb
) override;
token_ptr subscribe(
const string& topicFilter, int qos,
const subscribe_options& opts = subscribe_options(),
const properties& props = properties()
) override;
token_ptr subscribe(
const string& topicFilter, int qos, void* userContext, iaction_listener& cb,
const subscribe_options& opts = subscribe_options(),
const properties& props = properties()
) override;
token_ptr subscribe(
const_string_collection_ptr topicFilters, const qos_collection& qos,
const std::vector<subscribe_options>& opts = std::vector<subscribe_options>(),
const properties& props = properties()
) override;
token_ptr subscribe(
const_string_collection_ptr topicFilters, const qos_collection& qos,
void* userContext, iaction_listener& cb,
const std::vector<subscribe_options>& opts = std::vector<subscribe_options>(),
const properties& props = properties()
) override;
token_ptr unsubscribe(
const string& topicFilter, const properties& props = properties()
) override;
token_ptr unsubscribe(
const_string_collection_ptr topicFilters, const properties& props = properties()
) override;
token_ptr unsubscribe(
const_string_collection_ptr topicFilters, void* userContext, iaction_listener& cb,
const properties& props = properties()
) override;
token_ptr unsubscribe(
const string& topicFilter, void* userContext, iaction_listener& cb,
const properties& props = properties()
) override;
void start_consuming() override;
void stop_consuming() override;
void clear_consumer() override {
if (que_)
que_->clear();
}
bool consumer_closed() noexcept override { return !que_ || que_->closed(); }
bool consumer_done() noexcept override { return !que_ || que_->done(); }
std::size_t consumer_queue_size() const override { return (que_) ? que_->size() : 0; }
event consume_event() override;
bool try_consume_event(event* evt) override;
template <typename Rep, class Period>
bool try_consume_event_for(
event* evt, const std::chrono::duration<Rep, Period>& relTime
) {
if (!que_)
throw mqtt::exception(-1, "Consumer not started");
try {
return que_->try_get_for(evt, relTime);
}
catch (queue_closed&) {
*evt = event{shutdown_event{}};
return true;
}
}
template <typename Rep, class Period>
event try_consume_event_for(const std::chrono::duration<Rep, Period>& relTime) {
event evt;
try {
que_->try_get_for(&evt, relTime);
}
catch (queue_closed&) {
evt = event{shutdown_event{}};
}
return evt;
}
template <class Clock, class Duration>
bool try_consume_event_until(
event* evt, const std::chrono::time_point<Clock, Duration>& absTime
) {
if (!que_)
throw mqtt::exception(-1, "Consumer not started");
try {
return que_->try_get_until(evt, absTime);
}
catch (queue_closed&) {
*evt = event{shutdown_event{}};
return true;
}
}
template <class Clock, class Duration>
event try_consume_event_until(const std::chrono::time_point<Clock, Duration>& absTime) {
event evt;
try {
que_->try_get_until(&evt, absTime);
}
catch (queue_closed&) {
evt = event{shutdown_event{}};
}
return evt;
}
const_message_ptr consume_message() override;
bool try_consume_message(const_message_ptr* msg) override;
template <typename Rep, class Period>
bool try_consume_message_for(
const_message_ptr* msg, const std::chrono::duration<Rep, Period>& relTime
) {
if (!que_)
throw mqtt::exception(-1, "Consumer not started");
event evt;
while (true) {
if (!try_consume_event_for(&evt, relTime))
return false;
if (const auto* pval = evt.get_message_if()) {
*msg = std::move(*pval);
break;
}
if (evt.is_any_disconnect()) {
*msg = const_message_ptr{};
break;
}
}
return true;
}
template <typename Rep, class Period>
const_message_ptr try_consume_message_for(
const std::chrono::duration<Rep, Period>& relTime
) {
const_message_ptr msg;
this->try_consume_message_for(&msg, relTime);
return msg;
}
template <class Clock, class Duration>
bool try_consume_message_until(
const_message_ptr* msg, const std::chrono::time_point<Clock, Duration>& absTime
) {
if (!que_)
throw mqtt::exception(-1, "Consumer not started");
event evt;
while (true) {
if (!try_consume_event_until(&evt, absTime))
return false;
if (const auto* pval = evt.get_message_if()) {
*msg = std::move(*pval);
break;
}
if (!evt.is_any_disconnect()) {
*msg = const_message_ptr{};
break;
}
}
return true;
}
template <class Clock, class Duration>
const_message_ptr try_consume_message_until(
const std::chrono::time_point<Clock, Duration>& absTime
) {
const_message_ptr msg;
this->try_consume_message_until(&msg, absTime);
return msg;
}
};
using async_client_ptr = async_client::ptr_t;
} // namespace mqtt
#endif // __mqtt_async_client_h
callback.h
mqtt::async_client
Definition async_client.h:137
mqtt::async_client::set_connection_lost_handler
void set_connection_lost_handler(connection_handler cb)
mqtt::async_client::try_consume_event_until
bool try_consume_event_until(event *evt, const std::chrono::time_point< Clock, Duration > &absTime)
Definition async_client.h:891
mqtt::async_client::publish
delivery_token_ptr publish(const_message_ptr msg, void *userContext, iaction_listener &cb) override
mqtt::async_client::consumer_done
bool consumer_done() noexcept override
Definition async_client.h:820
mqtt::async_client::consumer_queue_type
std::unique_ptr< thread_queue< event > > consumer_queue_type
Definition async_client.h:142
mqtt::async_client::set_disconnected_handler
void set_disconnected_handler(disconnected_handler cb)
mqtt::async_client::try_consume_message_until
const_message_ptr try_consume_message_until(const std::chrono::time_point< Clock, Duration > &absTime)
Definition async_client.h:1021
mqtt::async_client::subscribe
token_ptr subscribe(const string &topicFilter, int qos, const subscribe_options &opts=subscribe_options(), const properties &props=properties()) override
mqtt::async_client::disconnect
token_ptr disconnect(disconnect_options opts) override
mqtt::async_client::try_consume_message
bool try_consume_message(const_message_ptr *msg) override
mqtt::async_client::disconnect
token_ptr disconnect() override
Definition async_client.h:404
mqtt::async_client::publish
delivery_token_ptr publish(string_ref topic, const void *payload, size_t n, int qos, bool retained, void *userContext, iaction_listener &cb) override
mqtt::async_client::publish
delivery_token_ptr publish(string_ref topic, const void *payload, size_t n) override
Definition async_client.h:560
mqtt::async_client::publish
delivery_token_ptr publish(string_ref topic, binary_ref payload, int qos, bool retained, const properties &props=properties()) override
mqtt::async_client::subscribe
token_ptr subscribe(const string &topicFilter, int qos, void *userContext, iaction_listener &cb, const subscribe_options &opts=subscribe_options(), const properties &props=properties()) override
mqtt::async_client::ptr_t
std::shared_ptr< async_client > ptr_t
Definition async_client.h:140
mqtt::async_client::consumer_closed
bool consumer_closed() noexcept override
Definition async_client.h:812
mqtt::async_client::stop_consuming
void stop_consuming() override
mqtt::async_client::get_connect_options
connect_options get_connect_options() const
Definition async_client.h:527
mqtt::async_client::try_consume_message_for
bool try_consume_message_for(const_message_ptr *msg, const std::chrono::duration< Rep, Period > &relTime)
Definition async_client.h:944
mqtt::async_client::disconnect
token_ptr disconnect(int timeout) override
Definition async_client.h:423
mqtt::async_client::subscribe
token_ptr subscribe(const_string_collection_ptr topicFilters, const qos_collection &qos, const std::vector< subscribe_options > &opts=std::vector< subscribe_options >(), const properties &props=properties()) override
mqtt::async_client::update_connection_handler
std::function< bool(connect_data &)> update_connection_handler
Definition async_client.h:151
mqtt::async_client::set_message_callback
void set_message_callback(message_handler cb)
mqtt::async_client::disconnect
token_ptr disconnect(const std::chrono::duration< Rep, Period > &timeout, void *userContext, iaction_listener &cb)
Definition async_client.h:471
mqtt::async_client::connect
token_ptr connect(connect_options options, void *userContext, iaction_listener &cb) override
mqtt::async_client::publish
delivery_token_ptr publish(string_ref topic, const void *payload, size_t n, int qos, bool retained, const properties &props=properties()) override
mqtt::async_client::disconnected_handler
std::function< void(const properties &, ReasonCode)> disconnected_handler
Definition async_client.h:149
mqtt::async_client::message_handler
std::function< void(const_message_ptr)> message_handler
Definition async_client.h:145
mqtt::async_client::consumer_queue_size
std::size_t consumer_queue_size() const override
Definition async_client.h:829
mqtt::async_client::async_client
async_client(const string &serverURI, const string &clientId, int maxBufferedMessages, const persistence_type &persistence=NO_PERSISTENCE)
Definition async_client.h:263
mqtt::async_client::unsubscribe
token_ptr unsubscribe(const string &topicFilter, const properties &props=properties()) override
mqtt::async_client::publish
delivery_token_ptr publish(string_ref topic, binary_ref payload) override
Definition async_client.h:587
mqtt::async_client::get_client_id
string get_client_id() const override
Definition async_client.h:506
mqtt::async_client::set_callback
void set_callback(callback &cb) override
mqtt::async_client::try_consume_message_until
bool try_consume_message_until(const_message_ptr *msg, const std::chrono::time_point< Clock, Duration > &absTime)
Definition async_client.h:990
mqtt::async_client::subscribe
token_ptr subscribe(const_string_collection_ptr topicFilters, const qos_collection &qos, void *userContext, iaction_listener &cb, const std::vector< subscribe_options > &opts=std::vector< subscribe_options >(), const properties &props=properties()) override
mqtt::async_client::consume_message
const_message_ptr consume_message() override
mqtt::async_client::reconnect
token_ptr reconnect() override
mqtt::async_client::try_consume_message_for
const_message_ptr try_consume_message_for(const std::chrono::duration< Rep, Period > &relTime)
Definition async_client.h:975
mqtt::async_client::set_connected_handler
void set_connected_handler(connection_handler cb)
mqtt::async_client::try_consume_event_for
bool try_consume_event_for(event *evt, const std::chrono::duration< Rep, Period > &relTime)
Definition async_client.h:852
mqtt::async_client::async_client
async_client(const create_options &opts)
Definition async_client.h:297
mqtt::async_client::disconnect
token_ptr disconnect(int timeout, void *userContext, iaction_listener &cb) override
mqtt::async_client::get_pending_delivery_tokens
std::vector< delivery_token_ptr > get_pending_delivery_tokens() const override
mqtt::async_client::unsubscribe
token_ptr unsubscribe(const_string_collection_ptr topicFilters, const properties &props=properties()) override
mqtt::async_client::unsubscribe
token_ptr unsubscribe(const_string_collection_ptr topicFilters, void *userContext, iaction_listener &cb, const properties &props=properties()) override
mqtt::async_client::clear_consumer
void clear_consumer() override
Definition async_client.h:801
mqtt::async_client::try_consume_event
bool try_consume_event(event *evt) override
mqtt::async_client::mqtt_version
int mqtt_version() const noexcept
Definition async_client.h:521
mqtt::async_client::unsubscribe
token_ptr unsubscribe(const string &topicFilter, void *userContext, iaction_listener &cb, const properties &props=properties()) override
mqtt::async_client::try_consume_event_until
event try_consume_event_until(const std::chrono::time_point< Clock, Duration > &absTime)
Definition async_client.h:912
mqtt::async_client::consume_event
event consume_event() override
mqtt::async_client::get_server_uri
string get_server_uri() const override
Definition async_client.h:511
mqtt::async_client::~async_client
~async_client() override
mqtt::async_client::disconnect
token_ptr disconnect(const std::chrono::duration< Rep, Period > &timeout)
Definition async_client.h:437
mqtt::async_client::get_pending_delivery_token
delivery_token_ptr get_pending_delivery_token(int msgID) const override
mqtt::async_client::disconnect
token_ptr disconnect(void *userContext, iaction_listener &cb) override
Definition async_client.h:489
mqtt::async_client::connect
token_ptr connect(void *userContext, iaction_listener &cb) override
Definition async_client.h:389
mqtt::async_client::disable_callbacks
void disable_callbacks() override
mqtt::async_client::async_client
async_client(const string &serverURI, const string &clientId=string{}, const persistence_type &persistence=NO_PERSISTENCE)
Definition async_client.h:243
mqtt::async_client::set_update_connection_handler
void set_update_connection_handler(update_connection_handler cb)
mqtt::async_client::connect
token_ptr connect() override
mqtt::async_client::is_connected
bool is_connected() const override
Definition async_client.h:535
mqtt::async_client::try_consume_event_for
event try_consume_event_for(const std::chrono::duration< Rep, Period > &relTime)
Definition async_client.h:873
mqtt::async_client::publish
delivery_token_ptr publish(const_message_ptr msg) override
mqtt::async_client::start_consuming
void start_consuming() override
mqtt::async_client::connect
token_ptr connect(connect_options options) override
mqtt::async_client::async_client
async_client(const string &serverURI, const string &clientId, const create_options &opts, const persistence_type &persistence=NO_PERSISTENCE)
Definition async_client.h:282
mqtt::async_client::connection_handler
std::function< void(const string &cause)> connection_handler
Definition async_client.h:147
mqtt::buffer_ref< char >
mqtt::callback
Definition callback.h:43
mqtt::connect_data
Definition connect_options.h:571
mqtt::connect_options
Definition connect_options.h:50
mqtt::create_options
Definition create_options.h:77
mqtt::create_options::get_server_uri
const string & get_server_uri() const noexcept
Definition create_options.h:205
mqtt::create_options::get_client_id
const string & get_client_id() const noexcept
Definition create_options.h:215
mqtt::disconnect_options
Definition disconnect_options.h:41
mqtt::event
Definition event.h:87
mqtt::exception
Definition exception.h:48
mqtt::iaction_listener
Definition iaction_listener.h:50
mqtt::iasync_client
Definition iasync_client.h:60
mqtt::iasync_client::qos_collection
std::vector< int > qos_collection
Definition iasync_client.h:66
mqtt::message::DFLT_RETAINED
static constexpr bool DFLT_RETAINED
Definition message.h:62
mqtt::message::DFLT_QOS
static constexpr int DFLT_QOS
Definition message.h:60
mqtt::properties
Definition properties.h:293
mqtt::queue_closed
Definition thread_queue.h:43
mqtt::subscribe_options
Definition subscribe_options.h:49
mqtt::token
Definition token.h:54
mqtt::topic
Definition topic.h:45
create_options.h
delivery_token.h
event.h
exception.h
iaction_listener.h
iasync_client.h
iclient_persistence.h
message.h
mqtt
Definition async_client.h:60
mqtt::ReasonCode
ReasonCode
Definition reason_code.h:39
mqtt::COPYRIGHT
const string COPYRIGHT("Copyright (c) 2013-2025 Frank Pagliughi")
mqtt::NO_PERSISTENCE
constexpr no_persistence NO_PERSISTENCE
Definition create_options.h:43
mqtt::const_string_collection_ptr
string_collection::const_ptr_t const_string_collection_ptr
Definition string_collection.h:259
mqtt::to_bool
bool to_bool(int n)
Definition types.h:107
mqtt::token_ptr
token::ptr_t token_ptr
Definition token.h:513
mqtt::VERSION
const uint32_t VERSION
Definition async_client.h:75
mqtt::const_message_ptr
message::const_ptr_t const_message_ptr
Definition message.h:372
mqtt::VERSION_STR
const string VERSION_STR("Paho MQTT C++ (mqttpp) v. 1.5.2")
mqtt::to_milliseconds_count
long to_milliseconds_count(const std::chrono::duration< Rep, Period > &dur)
Definition types.h:95
mqtt::delivery_token_ptr
delivery_token::ptr_t delivery_token_ptr
Definition delivery_token.h:127
mqtt::async_client_ptr
async_client::ptr_t async_client_ptr
Definition async_client.h:1031
mqtt::persistence_type
std::variant< no_persistence, string, iclient_persistence * > persistence_type
Definition create_options.h:53
properties.h
string_collection.h
mqtt::shutdown_event
Definition event.h:59
thread_queue.h
token.h
types.h