async_client.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_async_client_h
#define __mqtt_async_client_h

#include "MQTTAsync.h"
#include "mqtt/types.h"
#include "mqtt/token.h"
#include "mqtt/string_collection.h"
#include "mqtt/delivery_token.h"
#include "mqtt/iclient_persistence.h"
#include "mqtt/iaction_listener.h"
#include "mqtt/exception.h"
#include "mqtt/message.h"
#include "mqtt/callback.h"
#include "mqtt/thread_queue.h"
#include "mqtt/iasync_client.h"
#include <vector>
#include <list>
#include <memory>
#include <tuple>
#include <stdexcept>

namespace mqtt {

const uint32_t VERSION = 0x00090000;
const string VERSION_STR("Paho MQTT C++ (mqttpp) v. 0.9");
const string COPYRIGHT("Copyright (c) 2013-2017 Frank Pagliughi");


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

private:
    using guard = std::unique_lock<std::mutex>;
    using unique_lock = std::unique_lock<std::mutex>;

    mutable std::mutex lock_;
    MQTTAsync cli_;
    string serverURI_;
    string clientId_;
    std::unique_ptr<MQTTClient_persistence> persist_;
    callback* userCallback_;
    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 int  on_message_arrived(void* context, char* topicName, int topicLen,
                                   MQTTAsync_message* msg);
    static void on_delivery_complete(void* context, MQTTAsync_token tok);

    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;

public:
    async_client(const string& serverURI, const string& clientId,
                 const string& persistDir);
    async_client(const string& serverURI, const string& clientId,
                 iclient_persistence* persistence=nullptr);
    async_client(const string& serverURI, const string& clientId,
                 int maxBufferedMessages, const string& persistDir);
    async_client(const string& serverURI, const string& clientId,
                 int maxBufferedMessages, iclient_persistence* persistence=nullptr);
    ~async_client() override;
    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;
    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 clientId_; }
    string get_server_uri() const override { return serverURI_; }
    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) 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) 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;
    void set_callback(callback& cb) override;
    void disable_callbacks() override;
    token_ptr subscribe(const_string_collection_ptr topicFilters,
                         const qos_collection& qos) override;
    token_ptr subscribe(const_string_collection_ptr topicFilters,
                         const qos_collection& qos,
                         void* userContext, iaction_listener& cb) override;
    token_ptr subscribe(const string& topicFilter, int qos) override;
    token_ptr subscribe(const string& topicFilter, int qos,
                        void* userContext, iaction_listener& cb) override;
    token_ptr unsubscribe(const string& topicFilter) override;
    token_ptr unsubscribe(const_string_collection_ptr topicFilters) override;
    token_ptr unsubscribe(const_string_collection_ptr topicFilters,
                           void* userContext, iaction_listener& cb) override;
    token_ptr unsubscribe(const string& topicFilter,
                          void* userContext, iaction_listener& cb) override;

    void start_consuming();
    void stop_consuming();
    const_message_ptr consume_message() { return que_->get(); }
    bool try_consume_message(const_message_ptr* msg) {
        return que_->try_get(msg);
    }
    template <typename Rep, class Period>
    bool try_consume_message_for(const_message_ptr* msg,
                                 const std::chrono::duration<Rep, Period>& relTime) {
        return que_->try_get_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 que_->try_get_until(msg, absTime);
    }

};

using async_client_ptr = async_client::ptr_t;

// end namespace mqtt
}

#endif      // __mqtt_async_client_h