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_client_h
#define __mqtt_client_h

#include "mqtt/async_client.h"
#include <future>

namespace mqtt {


class client : private callback
{
    static const std::chrono::minutes DFLT_TIMEOUT;
    static constexpr 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);
    }
    void connection_lost(const string& cause) override {
        std::async(std::launch::async,
                   &callback::connection_lost, userCallback_, cause);
    }
    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);
    }

    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;

    client(const string& serverURI, const string& clientId,
           iclient_persistence* persistence=nullptr);
    client(const string& serverURI, const string& clientId,
           const string& persistDir);
    client(const string& serverURI, const string& clientId,
           int maxBufferedMessages, iclient_persistence* persistence=nullptr);
    client(const string& serverURI, const string& clientId,
           int maxBufferedMessages, const string& persistDir);
    virtual ~client() {}
    virtual void connect() {
        cli_.connect()->wait_for(timeout_);
        cli_.start_consuming();
    }
    virtual void connect(connect_options opts) {
        cli_.connect(std::move(opts))->wait_for(timeout_);
        cli_.start_consuming();
    }
    virtual void reconnect() { cli_.reconnect()->wait_for(timeout_); }
    virtual void disconnect() {
        cli_.stop_consuming();
        cli_.disconnect()->wait_for(timeout_);
    }
    virtual void disconnect(int timeoutMS) {
        cli_.stop_consuming();
        cli_.disconnect(timeoutMS)->wait_for(timeout_);
    }
    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_; }
    virtual topic get_topic(const string& top) { return topic(cli_, top); }
    virtual bool is_connected() const { return cli_.is_connected(); }

    virtual void publish(string_ref top, const void* payload, size_t n,
                         int qos, bool retained) {
        cli_.publish(std::move(top), payload, n, qos, retained)->wait_for(timeout_);
    }
    virtual void publish(string_ref top, const void* payload, size_t n) {
        cli_.publish(std::move(top), payload, n)->wait_for(timeout_);
    }
    virtual void publish(const_message_ptr msg) {
        cli_.publish(msg)->wait_for(timeout_);
    }
    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 void subscribe(const string& topicFilter) {
        cli_.subscribe(topicFilter, DFLT_QOS)->wait_for(timeout_);
    }
    virtual void subscribe(const string& topicFilter, int qos) {
        cli_.subscribe(topicFilter, qos)->wait_for(timeout_);
    }
    virtual void subscribe(const string_collection& topicFilters);
    virtual void subscribe(const string_collection& topicFilters,
                           const qos_collection& qos) {
        cli_.subscribe(ptr(topicFilters), qos)->wait_for(timeout_);
    }
    virtual void unsubscribe(const string& topicFilter) {
        cli_.unsubscribe(topicFilter)->wait_for(timeout_);
    }
    virtual void unsubscribe(const string_collection& topicFilters) {
        cli_.unsubscribe(ptr(topicFilters))->wait_for(timeout_);
    }
    void start_consuming() { cli_.start_consuming(); }
    void stop_consuming() { cli_.stop_consuming(); }
    const_message_ptr consume_message() { return cli_.consume_message(); }
    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;

// end namespace mqtt
}

#endif      // __mqtt_client_h