Polling for incoming messages about async-mqtt5 HOT 1 CLOSED

namespace asio = boost::asio;

using stream_type = asio::ip::tcp::socket;
using client_type = async_mqtt5::mqtt_client<stream_type>;

void receive_message(client_type& client) {
	client.async_receive([&client](
		async_mqtt5::error_code ec, std::string topic,
		std::string payload, async_mqtt5::publish_props
	) {
                // session and our subscriptions are lost
		if (ec == async_mqtt5::client::error::session_expired) {
			// resubscribe...
		} else if (ec) {
			// break...
		} else {
			// process message
			receive_message(client);
		}
	});
}

int main(int argc, char* argv[]) {
	asio::io_context ioc;

        // using signals to exit
	asio::signal_set signals(ioc, SIGINT, SIGTERM, SIGBREAK);
	signals.async_wait(
		[&ioc](boost::system::error_code ec, int sig) {
			ioc.stop();
		}
	);

	client_type client(ioc, "");

        // configure the client, subscribe and check if the subscription is successfully established

        receive_message(client);

        ioc.run();
}

#include <boost/asio/as_tuple.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/use_awaitable.hpp>

#include <boost/asio/ip/tcp.hpp>

#include <async_mqtt5.hpp>

namespace asio = boost::asio;
constexpr auto use_nothrow_awaitable = asio::as_tuple(asio::use_awaitable);

asio::awaitable<void> client_receiver(asio::io_context& ioc) {
	async_mqtt5::mqtt_client<asio::ip::tcp::socket> client(ioc, "");

        // configure client
	client.credentials("tester", "", "")
		.brokers("mqtt-broker", 1883)
		.run();

       // subscribe and check if the subscription is successful
	auto [ec, sub_codes, sub_props] = co_await client.async_subscribe(...);

       // if we failed to subscribe, co_return here

	for(;;) {
		auto [rec, topic, payload, publish_props] = co_await client.async_receive(use_nothrow_awaitable);

		if (rec == async_mqtt5::client::error::session_expired) {
			// resubscribe
		} else if (rec) {
			break;
		} else {
			// process message
		}
	}

	co_return;
}

int main() {
	asio::io_context ioc;

        // using signals to exit
	asio::signal_set signals(ioc, SIGINT, SIGTERM);
	signals.async_wait(
		[&ioc](boost::system::error_code ec, int sig) {
			ioc.stop();
		}
	);

	co_spawn(ioc, client_receiver(ioc), asio::detached);

	ioc.run();
}