Loading...
Searching...
No Matches
client.h
Go to the documentation of this file.
/*******************************************************************************
* Copyright (c) 2013-2023 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
*******************************************************************************/
#ifndef __mqtt_client_h
#define __mqtt_client_h
#include <future>
#include "mqtt/async_client.h"
namespace mqtt {
class client : private callback
{
PAHO_MQTTPP_EXPORT static const std::chrono::seconds DFLT_TIMEOUT;
PAHO_MQTTPP_EXPORT static const int DFLT_QOS; // =1;
async_client cli_;
std::chrono::milliseconds timeout_;
callback* userCallback_;
template <typename T>
std::shared_ptr<T> ptr(const T& val) {
return std::shared_ptr<T>(const_cast<T*>(&val), [](T*) {});
}
// User callbacks
// Most are launched in a separate thread, for convenience, except
// message_arrived, for performance.
void connected(const string& cause) override {
std::async(std::launch::async, &callback::connected, userCallback_, cause).wait();
}
void connection_lost(const string& cause) override {
std::async(std::launch::async, &callback::connection_lost, userCallback_, cause)
.wait();
}
void message_arrived(const_message_ptr msg) override {
userCallback_->message_arrived(msg);
}
void delivery_complete(delivery_token_ptr tok) override {
std::async(std::launch::async, &callback::delivery_complete, userCallback_, tok)
.wait();
}
client() = delete;
client(const async_client&) = delete;
client& operator=(const async_client&) = delete;
public:
using ptr_t = std::shared_ptr<client>;
using qos_collection = async_client::qos_collection;
using update_connection_handler = async_client::update_connection_handler;
client(
const string& serverURI, const string& clientId = string{},
const persistence_type& persistence = NO_PERSISTENCE
);
client(
const string& serverURI, const string& clientId, int maxBufferedMessages,
const persistence_type& persistence = NO_PERSISTENCE
);
client(
const string& serverURI, const string& clientId, const create_options& opts,
const persistence_type& persistence = NO_PERSISTENCE
);
client(const create_options& opts);
virtual ~client() {}
virtual connect_response connect();
virtual connect_response connect(connect_options opts);
virtual connect_response reconnect();
virtual void disconnect();
virtual void disconnect(int timeoutMS);
template <class Rep, class Period>
void disconnect(const std::chrono::duration<Rep, Period>& to) {
disconnect((int)to_milliseconds_count(to));
}
virtual string get_client_id() const { return cli_.get_client_id(); }
virtual string get_server_uri() const { return cli_.get_server_uri(); }
virtual std::chrono::milliseconds get_timeout() const { return timeout_; }
connect_options get_connect_options() const { return cli_.get_connect_options(); }
virtual topic get_topic(
const string& top, int qos = message::DFLT_QOS, bool retained = message::DFLT_RETAINED
) {
return topic(cli_, top, qos, retained);
}
virtual bool is_connected() const { return cli_.is_connected(); }
void set_update_connection_handler(update_connection_handler cb) {
cli_.set_update_connection_handler(cb);
}
virtual void publish(
string_ref top, const void* payload, size_t n, int qos, bool retained
) {
if (!cli_.publish(std::move(top), payload, n, qos, retained)->wait_for(timeout_))
throw timeout_error();
}
virtual void publish(string_ref top, const void* payload, size_t n) {
if (!cli_.publish(std::move(top), payload, n)->wait_for(timeout_))
throw timeout_error();
}
virtual void publish(const_message_ptr msg) {
if (!cli_.publish(msg)->wait_for(timeout_))
throw timeout_error();
}
virtual void publish(const message& msg) { cli_.publish(ptr(msg))->wait(); }
virtual void set_callback(callback& cb);
virtual void set_timeout(int timeoutMS) {
timeout_ = std::chrono::milliseconds(timeoutMS);
}
template <class Rep, class Period>
void set_timeout(const std::chrono::duration<Rep, Period>& to) {
timeout_ = to_milliseconds(to);
}
virtual subscribe_response subscribe(
const string& topicFilter, const subscribe_options& opts = subscribe_options(),
const properties& props = properties()
);
virtual subscribe_response subscribe(
const string& topicFilter, int qos,
const subscribe_options& opts = subscribe_options(),
const properties& props = properties()
);
virtual subscribe_response subscribe(
const string_collection& topicFilters,
const std::vector<subscribe_options>& opts = std::vector<subscribe_options>(),
const properties& props = properties()
);
virtual subscribe_response subscribe(
const string_collection& topicFilters, const qos_collection& qos,
const std::vector<subscribe_options>& opts = std::vector<subscribe_options>(),
const properties& props = properties()
);
virtual unsubscribe_response unsubscribe(
const string& topicFilter, const properties& props = properties()
);
virtual unsubscribe_response unsubscribe(
const string_collection& topicFilters, const properties& props = properties()
);
virtual void start_consuming() { cli_.start_consuming(); }
virtual void stop_consuming() { cli_.stop_consuming(); }
virtual const_message_ptr consume_message() { return cli_.consume_message(); }
virtual bool try_consume_message(const_message_ptr* msg) {
return cli_.try_consume_message(msg);
}
template <typename Rep, class Period>
bool try_consume_message_for(
const_message_ptr* msg, const std::chrono::duration<Rep, Period>& relTime
) {
return cli_.try_consume_message_for(msg, relTime);
}
template <class Clock, class Duration>
bool try_consume_message_until(
const_message_ptr* msg, const std::chrono::time_point<Clock, Duration>& absTime
) {
return cli_.try_consume_message_until(msg, absTime);
}
};
using client_ptr = client::ptr_t;
} // namespace mqtt
#endif // __mqtt_client_h
async_client.h
mqtt::async_client
Definition async_client.h:137
mqtt::async_client::try_consume_message
bool try_consume_message(const_message_ptr *msg) override
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::update_connection_handler
std::function< bool(connect_data &)> update_connection_handler
Definition async_client.h:151
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::get_client_id
string get_client_id() const override
Definition async_client.h:506
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::consume_message
const_message_ptr consume_message() override
mqtt::async_client::get_server_uri
string get_server_uri() const override
Definition async_client.h:511
mqtt::async_client::set_update_connection_handler
void set_update_connection_handler(update_connection_handler cb)
mqtt::async_client::is_connected
bool is_connected() const override
Definition async_client.h:535
mqtt::async_client::start_consuming
void start_consuming() override
mqtt::buffer_ref< char >
mqtt::callback
Definition callback.h:43
mqtt::callback::connected
virtual void connected(const string &)
Definition callback.h:61
mqtt::callback::connection_lost
virtual void connection_lost(const string &)
Definition callback.h:65
mqtt::callback::delivery_complete
virtual void delivery_complete(delivery_token_ptr)
Definition callback.h:74
mqtt::callback::message_arrived
virtual void message_arrived(const_message_ptr)
Definition callback.h:69
mqtt::client
Definition client.h:40
mqtt::client::subscribe
virtual subscribe_response subscribe(const string &topicFilter, const subscribe_options &opts=subscribe_options(), const properties &props=properties())
mqtt::client::ptr_t
std::shared_ptr< client > ptr_t
Definition client.h:93
mqtt::client::client
client(const string &serverURI, const string &clientId=string{}, const persistence_type &persistence=NO_PERSISTENCE)
mqtt::client::stop_consuming
virtual void stop_consuming()
Definition client.h:379
mqtt::client::try_consume_message_until
bool try_consume_message_until(const_message_ptr *msg, const std::chrono::time_point< Clock, Duration > &absTime)
Definition client.h:416
mqtt::client::try_consume_message
virtual bool try_consume_message(const_message_ptr *msg)
Definition client.h:392
mqtt::client::publish
virtual void publish(const message &msg)
Definition client.h:278
mqtt::client::set_update_connection_handler
void set_update_connection_handler(update_connection_handler cb)
Definition client.h:233
mqtt::client::disconnect
virtual void disconnect(int timeoutMS)
mqtt::client::publish
virtual void publish(const_message_ptr msg)
Definition client.h:267
mqtt::client::subscribe
virtual subscribe_response subscribe(const string_collection &topicFilters, const qos_collection &qos, const std::vector< subscribe_options > &opts=std::vector< subscribe_options >(), const properties &props=properties())
mqtt::client::disconnect
virtual void disconnect()
mqtt::client::consume_message
virtual const_message_ptr consume_message()
Definition client.h:385
mqtt::client::try_consume_message_for
bool try_consume_message_for(const_message_ptr *msg, const std::chrono::duration< Rep, Period > &relTime)
Definition client.h:403
mqtt::client::set_callback
virtual void set_callback(callback &cb)
mqtt::client::disconnect
void disconnect(const std::chrono::duration< Rep, Period > &to)
Definition client.h:185
mqtt::client::get_server_uri
virtual string get_server_uri() const
Definition client.h:197
mqtt::client::start_consuming
virtual void start_consuming()
Definition client.h:373
mqtt::client::connect
virtual connect_response connect(connect_options opts)
mqtt::client::subscribe
virtual subscribe_response subscribe(const string &topicFilter, int qos, const subscribe_options &opts=subscribe_options(), const properties &props=properties())
mqtt::client::unsubscribe
virtual unsubscribe_response unsubscribe(const string_collection &topicFilters, const properties &props=properties())
mqtt::client::update_connection_handler
async_client::update_connection_handler update_connection_handler
Definition client.h:98
mqtt::client::get_topic
virtual topic get_topic(const string &top, int qos=message::DFLT_QOS, bool retained=message::DFLT_RETAINED)
Definition client.h:217
mqtt::client::set_timeout
void set_timeout(const std::chrono::duration< Rep, Period > &to)
Definition client.h:297
mqtt::client::get_connect_options
connect_options get_connect_options() const
Definition client.h:208
mqtt::client::get_timeout
virtual std::chrono::milliseconds get_timeout() const
Definition client.h:202
mqtt::client::subscribe
virtual subscribe_response subscribe(const string_collection &topicFilters, const std::vector< subscribe_options > &opts=std::vector< subscribe_options >(), const properties &props=properties())
mqtt::client::is_connected
virtual bool is_connected() const
Definition client.h:227
mqtt::client::unsubscribe
virtual unsubscribe_response unsubscribe(const string &topicFilter, const properties &props=properties())
mqtt::client::publish
virtual void publish(string_ref top, const void *payload, size_t n, int qos, bool retained)
Definition client.h:246
mqtt::client::connect
virtual connect_response connect()
mqtt::client::client
client(const string &serverURI, const string &clientId, const create_options &opts, const persistence_type &persistence=NO_PERSISTENCE)
mqtt::client::qos_collection
async_client::qos_collection qos_collection
Definition client.h:95
mqtt::client::reconnect
virtual connect_response reconnect()
mqtt::client::client
client(const create_options &opts)
mqtt::client::publish
virtual void publish(string_ref top, const void *payload, size_t n)
Definition client.h:259
mqtt::client::~client
virtual ~client()
Definition client.h:152
mqtt::client::client
client(const string &serverURI, const string &clientId, int maxBufferedMessages, const persistence_type &persistence=NO_PERSISTENCE)
mqtt::client::get_client_id
virtual string get_client_id() const
Definition client.h:192
mqtt::client::set_timeout
virtual void set_timeout(int timeoutMS)
Definition client.h:289
mqtt::connect_options
Definition connect_options.h:50
mqtt::connect_response
Definition server_response.h:73
mqtt::create_options
Definition create_options.h:77
mqtt::iasync_client::qos_collection
std::vector< int > qos_collection
Definition iasync_client.h:66
mqtt::message
Definition message.h:57
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::string_collection
Definition string_collection.h:45
mqtt::subscribe_options
Definition subscribe_options.h:49
mqtt::timeout_error
Definition exception.h:196
mqtt::topic
Definition topic.h:45
mqtt::unsubscribe_response
Definition server_response.h:172
PAHO_MQTTPP_EXPORT
#define PAHO_MQTTPP_EXPORT
Definition export.h:40
mqtt
Definition async_client.h:60
mqtt::client_ptr
client::ptr_t client_ptr
Definition client.h:424
mqtt::NO_PERSISTENCE
constexpr no_persistence NO_PERSISTENCE
Definition create_options.h:43
mqtt::to_milliseconds
std::chrono::milliseconds to_milliseconds(const std::chrono::duration< Rep, Period > &dur)
Definition types.h:84
mqtt::const_message_ptr
message::const_ptr_t const_message_ptr
Definition message.h:372
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::persistence_type
std::variant< no_persistence, string, iclient_persistence * > persistence_type
Definition create_options.h:53
mqtt::subscribe_response
Definition server_response.h:131