Wait for connection before attempting to publish

mqtt_cpp_demo.cpp

#include <iostream>
#include <thread>
#include <chrono>

#include <boost/asio.hpp>
#include <mqtt_client_cpp.hpp>

struct App {
    boost::asio::io_context ioc {};
    using client_t = decltype(MQTT_NS::make_sync_client(std::declval<boost::asio::io_context &>(), "", 0));
    client_t c;
};

void task(std::shared_ptr<App> app, std::promise<void> & pro) {

    app->c->set_client_id("mqtt_cpp_demo");
    app->c->set_clean_session(true);

    app->c->set_connack_handler(
        [&c = app->c, &pro](bool sp, mqtt::connect_return_code connack_return_code){
            std::cout << "Connack handler called" << std::endl;
            std::cout << "Session Present: " << std::boolalpha << sp << std::endl;
            std::cout << "Connack Return Code: " << connack_return_code << std::endl;

            pro.set_value();  // signal to main thread that connection is up
            return true;
        }
    );

    app->c->set_close_handler( [](){} );

    app->c->set_error_handler(
        [](boost::system::error_code const & ec) {}
    );

    app->c->set_puback_handler(
        [](std::uint16_t packet_id) { return true; }
    );

    app->c->set_pubrec_handler(
        [](std::uint16_t packet_id) { return true; }
    );

    app->c->set_pubcomp_handler(
        [](std::uint16_t packet_id) { return true; }
    );

    app->c->set_suback_handler(
        [&c = app->c](std::uint16_t packet_id, std::vector<mqtt::suback_return_code> results){
            std::cout << "suback received. packet_id: " << packet_id << std::endl;
            for (auto const& e : results) {
                std::cout << "subscribe result: " << e << std::endl;
            }
            return true;
        }
    );

    app->c->set_publish_handler(
        [&c = app->c]
        (mqtt::optional<std::uint16_t> packet_id,
         mqtt::publish_options pubopts,
         mqtt::buffer topic_name,
         mqtt::buffer contents) {
            std::cout << "publish received."
                      << " dup: "    << pubopts.get_dup()
                      << " qos: "    << pubopts.get_qos()
                      << " retain: " << pubopts.get_retain() << std::endl;
            if (packet_id)
                std::cout << "packet_id: " << *packet_id << std::endl;
            std::cout << "topic_name: " << topic_name << std::endl;
            std::cout << "contents: " << contents << std::endl;
            //c->disconnect();
            return true;
        }
    );

    app->c->connect();
    app->ioc.run();
}

int main() {
    App appInstance;

    std::shared_ptr<App> app {&appInstance};

    app->c = mqtt::make_sync_client(app->ioc, "localhost", 1883);

    std::promise<void> pro;
    auto fut = pro.get_future();

    std::thread thread1 {&task, std::ref(app), std::ref(pro)};

    // wait for the connection
    fut.wait();

    while (true) {

//        std::this_thread::sleep_for(std::chrono::milliseconds(1));

//        std::cout << "publish" << std::endl;

        app->c->socket()->post(
            [&c = app->c] {
                c->publish("mqtt_cpp_demo/topic1", "0123456789ABCDEF", mqtt::qos::at_most_once);
            }
        );
    }

    return 0;
}