Loading...
Searching...
No Matches
token.h
Go to the documentation of this file.
/*******************************************************************************
* Copyright (c) 2013-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 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 & server responses
*******************************************************************************/
#ifndef __mqtt_token_h
#define __mqtt_token_h
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <vector>
#include "MQTTAsync.h"
#include "mqtt/buffer_ref.h"
#include "mqtt/exception.h"
#include "mqtt/iaction_listener.h"
#include "mqtt/properties.h"
#include "mqtt/server_response.h"
#include "mqtt/string_collection.h"
#include "mqtt/types.h"
namespace mqtt {
class iasync_client;
class token
{
public:
using ptr_t = std::shared_ptr<token>;
using const_ptr_t = std::shared_ptr<const token>;
using weak_ptr_t = std::weak_ptr<token>;
enum Type { CONNECT, SUBSCRIBE, PUBLISH, UNSUBSCRIBE, DISCONNECT };
private:
using guard = std::lock_guard<std::mutex>;
using unique_lock = std::unique_lock<std::mutex>;
mutable std::mutex lock_;
mutable std::condition_variable cond_;
Type type_;
iasync_client* cli_;
int rc_{0};
ReasonCode reasonCode_{ReasonCode::SUCCESS};
string errMsg_;
MQTTAsync_token msgId_;
const_string_collection_ptr topics_;
void* userContext_;
iaction_listener* listener_;
size_t nExpected_;
bool complete_;
std::unique_ptr<connect_response> connRsp_;
std::unique_ptr<subscribe_response> subRsp_;
std::unique_ptr<unsubscribe_response> unsubRsp_;
friend class async_client;
friend class mock_async_client;
friend class connect_options;
friend class response_options;
friend class delivery_response_options;
friend class disconnect_options;
void reset();
void set_message_id(MQTTAsync_token msgId) {
guard g(lock_);
msgId_ = msgId;
}
static void on_success(void* tokObj, MQTTAsync_successData* rsp);
static void on_success5(void* tokObj, MQTTAsync_successData5* rsp);
static void on_failure(void* tokObj, MQTTAsync_failureData* rsp);
static void on_failure5(void* tokObj, MQTTAsync_failureData5* rsp);
static void on_connected(void* tokObj, char* /*cause*/);
void on_success(MQTTAsync_successData* rsp);
void on_success5(MQTTAsync_successData5* rsp);
void on_failure(MQTTAsync_failureData* rsp);
void on_failure5(MQTTAsync_failureData5* rsp);
void check_ret() const {
if (rc_ != MQTTASYNC_SUCCESS || reasonCode_ >= 0x80)
throw exception(rc_, reasonCode_, errMsg_);
}
public:
token(Type typ, iasync_client& cli) : token{typ, cli, MQTTAsync_token(0)} {}
token(Type typ, iasync_client& cli, void* userContext, iaction_listener& cb)
: token{typ, cli, const_string_collection_ptr(), userContext, cb} {}
token(Type typ, iasync_client& cli, const string& topic)
: token{typ, cli, string_collection::create(topic)} {}
token(
Type typ, iasync_client& cli, const string& topic, void* userContext,
iaction_listener& cb
)
: token{typ, cli, string_collection::create(topic), userContext, cb} {}
token(Type typ, iasync_client& cli, const_string_collection_ptr topics);
token(
Type typ, iasync_client& cli, const_string_collection_ptr topics, void* userContext,
iaction_listener& cb
);
token(Type typ, iasync_client& cli, MQTTAsync_token tok);
virtual ~token() {}
static ptr_t create(Type typ, iasync_client& cli) {
return std::make_shared<token>(typ, cli);
}
static ptr_t create(
Type typ, iasync_client& cli, void* userContext, iaction_listener& cb
) {
return std::make_shared<token>(typ, cli, userContext, cb);
}
static ptr_t create(Type typ, iasync_client& cli, const string& topic) {
return std::make_shared<token>(typ, cli, topic);
}
static ptr_t create(
Type typ, iasync_client& cli, const string& topic, void* userContext,
iaction_listener& cb
) {
return std::make_shared<token>(typ, cli, topic, userContext, cb);
}
static ptr_t create(Type typ, iasync_client& cli, const_string_collection_ptr topics) {
return std::make_shared<token>(typ, cli, topics);
}
static ptr_t create(
Type typ, iasync_client& cli, const_string_collection_ptr topics, void* userContext,
iaction_listener& cb
) {
return std::make_shared<token>(typ, cli, topics, userContext, cb);
}
Type get_type() const { return type_; }
virtual iaction_listener* get_action_callback() const {
guard g(lock_);
return listener_;
}
virtual iasync_client* get_client() const { return cli_; }
virtual int get_message_id() const {
static_assert(sizeof(msgId_) <= sizeof(int), "MQTTAsync_token must fit into int");
return int(msgId_);
}
virtual const_string_collection_ptr get_topics() const { return topics_; }
virtual void* get_user_context() const {
guard g(lock_);
return userContext_;
}
virtual bool is_complete() const { return complete_; }
explicit operator bool() const {
guard g(lock_);
return rc_ == MQTTASYNC_SUCCESS && reasonCode_ < 0x80;
}
virtual int get_return_code() const { return rc_; }
virtual void set_action_callback(iaction_listener& listener);
virtual void set_user_context(void* userContext) {
guard g(lock_);
userContext_ = userContext;
}
void set_num_expected(size_t n) { nExpected_ = n; }
ReasonCode get_reason_code() const { return reasonCode_; }
string get_error_message() const { return errMsg_; }
virtual void wait();
virtual bool try_wait() {
guard g(lock_);
if (complete_)
check_ret();
return complete_;
}
virtual bool wait_for(long timeout) {
return wait_for(std::chrono::milliseconds(timeout));
}
template <class Rep, class Period>
bool wait_for(const std::chrono::duration<Rep, Period>& relTime) {
unique_lock g(lock_);
if (!cond_.wait_for(g, std::chrono::milliseconds(relTime), [this] {
return complete_;
}))
return false;
check_ret();
return true;
}
template <class Clock, class Duration>
bool wait_until(const std::chrono::time_point<Clock, Duration>& absTime) {
unique_lock g(lock_);
if (!cond_.wait_until(g, absTime, [this] { return complete_; }))
return false;
check_ret();
return true;
}
connect_response get_connect_response() const;
subscribe_response get_subscribe_response() const;
unsubscribe_response get_unsubscribe_response() const;
};
using token_ptr = token::ptr_t;
using const_token_ptr = token::const_ptr_t;
} // namespace mqtt
#endif // __mqtt_token_h
buffer_ref.h
mqtt::async_client
Definition async_client.h:121
mqtt::connect_options
Definition connect_options.h:50
mqtt::connect_response
Definition server_response.h:73
mqtt::delivery_response_options
Definition response_options.h:264
mqtt::disconnect_options
Definition disconnect_options.h:41
mqtt::exception
Definition exception.h:48
mqtt::iaction_listener
Definition iaction_listener.h:50
mqtt::iasync_client
Definition iasync_client.h:60
mqtt::response_options
Definition response_options.h:51
mqtt::string_collection
Definition string_collection.h:45
mqtt::token
Definition token.h:54
mqtt::token::token
token(Type typ, iasync_client &cli, void *userContext, iaction_listener &cb)
Definition token.h:201
mqtt::token::create
static ptr_t create(Type typ, iasync_client &cli, const string &topic)
Definition token.h:289
mqtt::token::create
static ptr_t create(Type typ, iasync_client &cli)
Definition token.h:266
mqtt::token::get_message_id
virtual int get_message_id() const
Definition token.h:357
mqtt::token::get_user_context
virtual void * get_user_context() const
Definition token.h:372
mqtt::token::get_return_code
virtual int get_return_code() const
Definition token.h:398
mqtt::token::wait
virtual void wait()
mqtt::token::token
token(Type typ, iasync_client &cli, const string &topic, void *userContext, iaction_listener &cb)
Definition token.h:222
mqtt::token::create
static ptr_t create(Type typ, iasync_client &cli, const string &topic, void *userContext, iaction_listener &cb)
Definition token.h:302
mqtt::token::get_type
Type get_type() const
Definition token.h:338
mqtt::token::get_reason_code
ReasonCode get_reason_code() const
Definition token.h:423
mqtt::token::ptr_t
std::shared_ptr< token > ptr_t
Definition token.h:57
mqtt::token::get_unsubscribe_response
unsubscribe_response get_unsubscribe_response() const
mqtt::token::create
static ptr_t create(Type typ, iasync_client &cli, const_string_collection_ptr topics)
Definition token.h:314
mqtt::token::get_subscribe_response
subscribe_response get_subscribe_response() const
mqtt::token::const_ptr_t
std::shared_ptr< const token > const_ptr_t
Definition token.h:59
mqtt::token::wait_for
virtual bool wait_for(long timeout)
Definition token.h:452
mqtt::token::get_topics
virtual const_string_collection_ptr get_topics() const
Definition token.h:367
mqtt::token::is_complete
virtual bool is_complete() const
Definition token.h:380
mqtt::token::set_user_context
virtual void set_user_context(void *userContext)
Definition token.h:409
mqtt::token::token
token(Type typ, iasync_client &cli)
Definition token.h:191
mqtt::token::Type
Type
Definition token.h:64
mqtt::token::SUBSCRIBE
@ SUBSCRIBE
Definition token.h:64
mqtt::token::CONNECT
@ CONNECT
Definition token.h:64
mqtt::token::UNSUBSCRIBE
@ UNSUBSCRIBE
Definition token.h:64
mqtt::token::PUBLISH
@ PUBLISH
Definition token.h:64
mqtt::token::DISCONNECT
@ DISCONNECT
Definition token.h:64
mqtt::token::get_error_message
string get_error_message() const
Definition token.h:428
mqtt::token::token
token(Type typ, iasync_client &cli, MQTTAsync_token tok)
mqtt::token::wait_until
bool wait_until(const std::chrono::time_point< Clock, Duration > &absTime)
Definition token.h:478
mqtt::token::token
token(Type typ, iasync_client &cli, const_string_collection_ptr topics)
mqtt::token::get_connect_response
connect_response get_connect_response() const
mqtt::token::get_action_callback
virtual iaction_listener * get_action_callback() const
Definition token.h:343
mqtt::token::get_client
virtual iasync_client * get_client() const
Definition token.h:352
mqtt::token::try_wait
virtual bool try_wait()
Definition token.h:439
mqtt::token::~token
virtual ~token()
Definition token.h:259
mqtt::token::weak_ptr_t
std::weak_ptr< token > weak_ptr_t
Definition token.h:61
mqtt::token::token
token(Type typ, iasync_client &cli, const string &topic)
Definition token.h:210
mqtt::token::set_num_expected
void set_num_expected(size_t n)
Definition token.h:418
mqtt::token::token
token(Type typ, iasync_client &cli, const_string_collection_ptr topics, void *userContext, iaction_listener &cb)
mqtt::token::set_action_callback
virtual void set_action_callback(iaction_listener &listener)
mqtt::token::mock_async_client
friend class mock_async_client
Definition token.h:114
mqtt::token::create
static ptr_t create(Type typ, iasync_client &cli, const_string_collection_ptr topics, void *userContext, iaction_listener &cb)
Definition token.h:327
mqtt::token::create
static ptr_t create(Type typ, iasync_client &cli, void *userContext, iaction_listener &cb)
Definition token.h:278
mqtt::token::wait_for
bool wait_for(const std::chrono::duration< Rep, Period > &relTime)
Definition token.h:462
mqtt::topic
Definition topic.h:45
mqtt::unsubscribe_response
Definition server_response.h:172
exception.h
iaction_listener.h
mqtt
Definition async_client.h:60
mqtt::ReasonCode
ReasonCode
Definition reason_code.h:39
mqtt::SUCCESS
@ SUCCESS
Definition reason_code.h:40
mqtt::const_token_ptr
token::const_ptr_t const_token_ptr
Definition token.h:516
mqtt::const_string_collection_ptr
string_collection::const_ptr_t const_string_collection_ptr
Definition string_collection.h:259
mqtt::token_ptr
token::ptr_t token_ptr
Definition token.h:513
properties.h
server_response.h
string_collection.h
mqtt::subscribe_response
Definition server_response.h:131
types.h