token.h

Go to the documentation of this file.

/*******************************************************************************
 * Copyright (c) 2013-2017 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:
 *    Frank Pagliughi - initial implementation and documentation
 *******************************************************************************/

#ifndef __mqtt_token_h
#define __mqtt_token_h

#include "MQTTAsync.h"
#include "mqtt/iaction_listener.h"
#include "mqtt/exception.h"
#include "mqtt/types.h"
#include "mqtt/buffer_ref.h"
#include "mqtt/string_collection.h"
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>

namespace mqtt {

class iasync_client;


class token
{
    using guard = std::lock_guard<std::mutex>;
    using unique_lock = std::unique_lock<std::mutex>;

    mutable std::mutex lock_;
    std::condition_variable cond_;

    iasync_client* cli_;
    MQTTAsync_token tok_;
    const_string_collection_ptr topics_;
    void* userContext_;
    iaction_listener* listener_;
    bool complete_;
    int rc_;
    string errMsg_;

    friend class async_client;
    friend class token_test;

    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_);
        tok_ = msgid;
    }
    static void on_success(void* tokObj, MQTTAsync_successData* rsp);
    static void on_failure(void* tokObj, MQTTAsync_failureData* rsp);
    static void on_connected(void* tokObj, char* /*cause*/);
    void on_success(MQTTAsync_successData* rsp);
    void on_failure(MQTTAsync_failureData* rsp);

    void check_rc() {
        if (rc_ != MQTTASYNC_SUCCESS)
            throw exception(rc_, errMsg_);
    }

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>;

    token(iasync_client& cli);
    token(iasync_client& cli, void* userContext, iaction_listener& cb);
    token(iasync_client& cli, const string& topic);
    token(iasync_client& cli, const string& topic,
          void* userContext, iaction_listener& cb);
    token(iasync_client& cli, const_string_collection_ptr topics);
    token(iasync_client& cli, const_string_collection_ptr topics,
          void* userContext, iaction_listener& cb);
    token(iasync_client& cli, MQTTAsync_token tok);
    virtual ~token() {}
    static ptr_t create(iasync_client& cli) {
        return std::make_shared<token>(cli);
    }
    static ptr_t create(iasync_client& cli, void* userContext, iaction_listener& cb) {
        return std::make_shared<token>(cli, userContext, cb);
    }
    static ptr_t create(iasync_client& cli, const string& topic) {
        return std::make_shared<token>(cli, topic);
    }
    static ptr_t create(iasync_client& cli, const string& topic,
                        void* userContext, iaction_listener& cb) {
        return std::make_shared<token>(cli, topic, userContext, cb);
    }
    static ptr_t create(iasync_client& cli, const_string_collection_ptr topics) {
        return std::make_shared<token>(cli, topics);
    }
    static ptr_t create(iasync_client& cli, const_string_collection_ptr topics,
                        void* userContext, iaction_listener& cb) {
        return std::make_shared<token>(cli, topics, userContext, cb);
    }
    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(tok_) <= sizeof(int), "MQTTAsync_token must fit into int");
        return int(tok_);
    }
    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_; }
    virtual int get_return_code() const { return rc_; }
    virtual void set_action_callback(iaction_listener& listener) {
        guard g(lock_);
        listener_ = &listener;
    }
    virtual void set_user_context(void* userContext) {
        guard g(lock_);
        userContext_ = userContext;
    }
    virtual void wait();
    virtual bool try_wait() {
        guard g(lock_);
        if (complete_)
            check_rc();
        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_rc();
        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_rc();
        return true;
    }
};

using token_ptr = token::ptr_t;

using const_token_ptr = token::const_ptr_t;

// end namespace mqtt
}

#endif      // __mqtt_token_h