boostorg/mqtt5
1
0
Fork 1
mirror of https://github.com/boostorg/mqtt5.git synced 2025-08-01 05:24:50 +02:00
Code Issues Packages Projects Releases Wiki Activity

publisher & receiver examples, better completion token examples

Browse Source 
Reviewers: iljazovic, ivica

Reviewed By: ivica

Subscribers: miljen

Differential Revision: https://repo.mireo.local/D26582
This commit is contained in:
Korina Šimičević
2023-11-17 09:46:07 +01:00
parent 376393fbfa
commit 73620e5653
12 changed files with 372 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
example/callbacks.cpp
View File

@@ -1,5 +1,3 @@
//[callbacks_examples
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
@@ -12,13 +10,10 @@ namespace asio = boost::asio;
using stream_type = asio::ip::tcp::socket;
using client_type = async_mqtt5::mqtt_client<stream_type>;
/**
* This function showcases how to call each asynchronous function
* in mqtt_client using callbacks. Note that this example is not
* intended for direct execution, as the async_disconnect call
* will promptly close the client.
*/
void run_with_callbacks(client_type& client) {
//[publish_callback
// Publish an Application Message with QoS 0.
client.async_publish<async_mqtt5::qos_e::at_most_once>(
"test/mqtt-test", "Hello world!",
@@ -50,7 +45,9 @@ void run_with_callbacks(client_type& client) {
std::cout << "reason_code: " << rc.message() << std::endl;
}
);
//]
//[subscribe_callback
// Subscribe to a single Topic.
client.async_subscribe(
{ "test/mqtt-test", { async_mqtt5::qos_e::exactly_once } }, async_mqtt5::subscribe_props {},
@@ -62,7 +59,9 @@ void run_with_callbacks(client_type& client) {
std::cout << "subscribe reason_code: " << codes[0].message() << std::endl;
}
);
//]
//[receive_callback
// Receive an Application Message.
client.async_receive(
// Callback with signature void (error_code, std::string, std::string, publish_props)
@@ -74,7 +73,9 @@ void run_with_callbacks(client_type& client) {
std::cout << "payload: " << payload << std::endl;
}
);
//]
//[unsubscribe_callback
// Unsubscribe from the Topic.
client.async_unsubscribe("test/mqtt-test", async_mqtt5::unsubscribe_props {},
//Callback with signature void (error_code, std::vector<reason_code>, unsuback_props)
@@ -85,7 +86,9 @@ void run_with_callbacks(client_type& client) {
std::cout << "unsubscribe reason_code: " << codes[0].message() << std::endl;
}
);
//]
//[disconnect_callback
// Disconnect the Client.
client.async_disconnect(
async_mqtt5::disconnect_rc_e::disconnect_with_will_message,
@@ -93,7 +96,7 @@ void run_with_callbacks(client_type& client) {
// Callback with signature void (error_code)
[](async_mqtt5::error_code) {}
);
//]
}
int main(int argc, char** argv) {
@@ -110,5 +113,3 @@ int main(int argc, char** argv) {
ioc.run();
}
//]

81
example/cpp20_coroutines.cpp
View File

@@ -1,5 +1,3 @@
//[cpp20_coroutines_examples
#include <boost/asio/as_tuple.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
@@ -15,16 +13,8 @@ namespace asio = boost::asio;
using stream_type = asio::ip::tcp::socket;
using client_type = async_mqtt5::mqtt_client<stream_type>;
/**
* An example of a coroutine. It must have a return type of boost::asio::awaitable<T>.
* When an asynchronous function is called, the coroutine is suspended.
* After the asynchronous operation finishes, the coroutine resumes from the point it was suspended.
*
* In this example, each asynchronous function is invoked with ``__USE_AWAITABLE__`` completion token.
* When using this completion token, co_await will throw exceptions instead of returning an error code.
* If you do not wish to throw exceptions, refer to the following use_nothrow_awaitable and nothrow_coroutine() example.
*/
asio::awaitable<void> coroutine(client_type& client) {
//[publish_coro
// Publish an Application Message with QoS 0.
// The handler signature for this function is void (error_code).
// However, when using asio::use_awaitable as a completion token,
@@ -52,32 +42,29 @@ asio::awaitable<void> coroutine(client_type& client) {
async_mqtt5::retain_e::yes, async_mqtt5::publish_props {},
asio::use_awaitable
);
//]
auto sub_topic = async_mqtt5::subscribe_topic {
"test/mqtt-test", {
async_mqtt5::qos_e::exactly_once,
async_mqtt5::subscribe_options::no_local_e::no,
async_mqtt5::subscribe_options::retain_as_published_e::retain,
async_mqtt5::subscribe_options::retain_handling_e::send
}
};
//[subscribe_coro
// Subscribe to a single Topic.
// The handler signature for this function is void (error_code, std::vector<reason_code>, suback_props).
// With asio::use_awaitable as a completion token, the co_await
// will return std::vector<reason_code> and suback_props.
auto [sub_codes, sub_props] = co_await client.async_subscribe(
sub_topic, async_mqtt5::subscribe_props {}, asio::use_awaitable
{ "test/mqtt-test", { async_mqtt5::qos_e::exactly_once } },
async_mqtt5::subscribe_props {}, asio::use_awaitable
);
//]
//[receive_coro
// Receive an Application Message.
// The co_await call will return std::string (topic), std::string (payload) and publish_props.
// Note: the coroutine will be suspended until an Application Message is ready to be received
// or an error has occurred. In theory, the coroutine could be suspended indefinitely.
// Avoid calling this if you have not successfully subscribed to a Topic.
if (!sub_codes[0])
auto [topic, payload, publish_props] = co_await client.async_receive(asio::use_awaitable);
auto [topic, payload, publish_props] = co_await client.async_receive(asio::use_awaitable);
//]
//[unsubscribe_coro
// Unsubscribe from the Topic.
// The handler signature for this function is void (error_code, std::vector<reason_code>, unsuback_props).
// With asio::use_awaitable as a completion token, the co_await
@@ -86,7 +73,9 @@ asio::awaitable<void> coroutine(client_type& client) {
"test/mqtt-test", async_mqtt5::unsubscribe_props {},
asio::use_awaitable
);
//]
//[disconnect_coro
// Disconnect the Client.
// With asio::use_awaitable as a completion token and void (error_code) as the completion signature,
// the co_await has nothing to return.
@@ -95,23 +84,18 @@ asio::awaitable<void> coroutine(client_type& client) {
async_mqtt5::disconnect_props {},
asio::use_awaitable
);
//]
co_return;
}
/**
* A modified completion token. Using this completion token instead of ``__USE_AWAITABLE__``
* will prevent co_await from throwing exceptions. Instead, co_await will return the error code
* along with other values specified in the handler signature.
*/
//[no_throw_awaitable
constexpr auto use_nothrow_awaitable = asio::as_tuple(asio::use_awaitable);
//]
/**
* In this coroutine, each asynchronous function is called with use_nothrow_awaitable completion token.
* Each co_await will return an error_code, similar to the behavior of using callbacks
* (see ``__EXAMPLE_CALLBACK__``).
*/
asio::awaitable<void> nothrow_coroutine(client_type& client) {
//[publish_coro_nothrow
async_mqtt5::error_code ec;
async_mqtt5::reason_code rc;
@@ -134,39 +118,38 @@ asio::awaitable<void> nothrow_coroutine(client_type& client) {
async_mqtt5::retain_e::yes, async_mqtt5::publish_props {},
use_nothrow_awaitable
);
//]
auto sub_topic = async_mqtt5::subscribe_topic{
"test/mqtt-test", {
async_mqtt5::qos_e::exactly_once,
async_mqtt5::subscribe_options::no_local_e::no,
async_mqtt5::subscribe_options::retain_as_published_e::retain,
async_mqtt5::subscribe_options::retain_handling_e::send
}
};
//[subscribe_coro_nothrow
std::vector<async_mqtt5::reason_code> rcs;
async_mqtt5::suback_props suback_props;
std::tie(ec, rcs, suback_props) = co_await client.async_subscribe(
sub_topic, async_mqtt5::subscribe_props {}, use_nothrow_awaitable
{ "test/mqtt-test", { async_mqtt5::qos_e::exactly_once } },
async_mqtt5::subscribe_props {}, use_nothrow_awaitable
);
//]
if (!rcs[0]) {
std::string topic, payload;
async_mqtt5::publish_props publish_props;
std::tie(ec, topic, payload, publish_props) = co_await client.async_receive(use_nothrow_awaitable);
}
//[receive_coro_nothrow
std::string topic, payload;
async_mqtt5::publish_props publish_props;
std::tie(ec, topic, payload, publish_props) = co_await client.async_receive(use_nothrow_awaitable);
//]
//[unsubscribe_coro_nothrow
async_mqtt5::unsuback_props unsuback_props;
std::tie(ec, rcs, unsuback_props) = co_await client.async_unsubscribe(
std::vector<std::string>{ "test/mqtt-test" }, async_mqtt5::unsubscribe_props {},
use_nothrow_awaitable
);
//]
//[disconnect_coro_nothrow
std::tie(ec) = co_await client.async_disconnect(
async_mqtt5::disconnect_rc_e::disconnect_with_will_message,
async_mqtt5::disconnect_props {},
use_nothrow_awaitable
);
//]
co_return;
}
@@ -188,5 +171,3 @@ int main(int argc, char** argv) {
ioc.run();
}
//]

46
example/futures.cpp
View File

@@ -1,5 +1,3 @@
//[futures_examples
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_future.hpp>
@@ -15,13 +13,9 @@ namespace asio = boost::asio;
using stream_type = asio::ip::tcp::socket;
using client_type = async_mqtt5::mqtt_client<stream_type>;
/**
* This function is a reference on how to use the mqtt_client with ``__USE_FUTURE__``
* as the completion token. Each get() call on std::future will block the current
* thread and wait until the future has a valid result. That is why it is essential
* to ensure that the execution context is running in more than one thread.
*/
void run_with_future(client_type& client) {
//[publish_future
// Just like the asio::use_awaitable completion token
// (see ``__EXAMPLE_COROUTINE__``), the ``__USE_FUTURE__`` completion token
// will not return the error_code. Instead, it will throw an exception if an error has occurred.
@@ -52,42 +46,40 @@ void run_with_future(client_type& client) {
);
auto [qos2_rc, pubcomp_props] = pub_qos2_fut.get();
std::cout << "Publish QoS 2 Reason Code: " << qos2_rc.message() << std::endl;
//]
auto sub_topic = async_mqtt5::subscribe_topic{
"test/mqtt-test", {
async_mqtt5::qos_e::exactly_once,
async_mqtt5::subscribe_options::no_local_e::no,
async_mqtt5::subscribe_options::retain_as_published_e::retain,
async_mqtt5::subscribe_options::retain_handling_e::send
}
};
//[subscribe_future
using sub_fut_type = std::tuple<std::vector<async_mqtt5::reason_code>, async_mqtt5::suback_props>;
std::future<sub_fut_type> sub_fut = client.async_subscribe(
sub_topic, async_mqtt5::subscribe_props {}, asio::use_future
{ "test/mqtt-test", { async_mqtt5::qos_e::exactly_once } },
async_mqtt5::subscribe_props {}, asio::use_future
);
auto [sub_rcs, suback_props] = sub_fut.get();
std::cout << "Subscribe Reason Code: " << sub_rcs[0].message() << std::endl;
//]
// Note: the get() call on async_receive future could block indefinitely if the mqtt_client
//[receive_future
// Note: the get() call on async_receive future could block indefinitely if the ``__Client__``
// failed to subscribe or there are no Application Messages to be received from the subscribed Topic!
if (!sub_rcs[0]) {
using rec_fut_type = std::tuple<std::string, std::string, async_mqtt5::publish_props>;
std::future<rec_fut_type> rec_fut = client.async_receive(asio::use_future);
auto [topic, payload, publish_props] = rec_fut.get();
std::cout << "Received message from Topic: " << topic << ", " << payload << std::endl;
}
using rec_fut_type = std::tuple<std::string, std::string, async_mqtt5::publish_props>;
std::future<rec_fut_type> rec_fut = client.async_receive(asio::use_future);
auto [topic, payload, publish_props] = rec_fut.get();
std::cout << "Received message from Topic: " << topic << ", " << payload << std::endl;
//]
//[unsubscribe_future
using unsub_fut_type = std::tuple<std::vector<async_mqtt5::reason_code>, async_mqtt5::unsuback_props>;
std::future<unsub_fut_type> unsub_fut = client.async_unsubscribe(
"test/mqtt-test", async_mqtt5::unsubscribe_props {}, asio::use_future
);
auto [unsub_rcs, unsuback_props] = unsub_fut.get();
std::cout << "Unubscribe Reason Code: " << unsub_rcs[0].message() << std::endl;
//]
//[disconnect_future
std::future<void> dc_fut = client.async_disconnect(asio::use_future);
dc_fut.get();
//]
return;
}
@@ -117,5 +109,3 @@ int main(int argc, char** argv) {
if (t.joinable()) t.join();
}
//]

55
example/publisher.cpp Normal file
View File

@@ -0,0 +1,55 @@
//[publisher
#include <iostream>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <async_mqtt5.hpp>
namespace asio = boost::asio;
asio::awaitable<void> client_publisher(asio::io_context& ioc) {
// Initialise the ``__Client__``, establish connection to the Broker over TCP.
async_mqtt5::mqtt_client<asio::ip::tcp::socket> client(ioc, "");
// Configure the ``__Client__``.
// It is mandatory to call brokers() and run() to configure the Brokers to connect to and start the Client.
client.brokers("mqtt.mireo.local", 1883) // Broker that we want to connect to. 1883 is the default TCP port.
.run(); // Start the client.
// Publish an Application Message with QoS 1.
auto [rc, props] = co_await client.async_publish<async_mqtt5::qos_e::at_least_once>(
"test/mqtt-test", "my application message",
async_mqtt5::retain_e::yes, async_mqtt5::publish_props {}, asio::use_awaitable
);
if (rc)
std::cout << "MQTT protocol error occurred: " << rc.message() << std::endl;
// Publish some more messages...
// After we are done with publishing all the messages, disconnect the Client.
// Alternatively, you can also use mqtt_client::cancel.
// Regardless, you should ensure all the operations are completed before disconnecting the Client.
co_await client.async_disconnect(
async_mqtt5::disconnect_rc_e::normal_disconnection, async_mqtt5::disconnect_props {}, asio::use_awaitable
);
co_return;
}
int main() {
// Initialise execution context.
asio::io_context ioc;
// Spawn the coroutine.
asio::co_spawn(ioc.get_executor(), client_publisher(ioc), asio::detached);
// Start the execution.
ioc.run();
}
//]

80
example/receiver.cpp Normal file
View File

@@ -0,0 +1,80 @@
//[receiver
#include <iostream>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <async_mqtt5.hpp>
namespace asio = boost::asio;
asio::awaitable<void> client_receiver(asio::io_context& ioc) {
// Initialise the ``__Client__``, establish connection to the Broker over TCP.
async_mqtt5::mqtt_client<asio::ip::tcp::socket> client(ioc, "");
// Configure the``__Client__``.
// It is mandatory to call brokers() and run() to configure the Brokers to connect to and start the Client.
client.brokers("mqtt.broker", 1883) // Broker that we want to connect to. 1883 is the default TCP port.
.run(); // Start the client.
// Configure the request to subscribe to a Topic.
async_mqtt5::subscribe_topic sub_topic = async_mqtt5::subscribe_topic {
"test/mqtt-test",
async_mqtt5::subscribe_options {
async_mqtt5::qos_e::exactly_once, // All messages will arrive at QoS 2.
async_mqtt5::subscribe_options::no_local_e::no, // Forward message from Clients with same ID.
async_mqtt5::subscribe_options::retain_as_published_e::retain, // Keep the original RETAIN flag.
async_mqtt5::subscribe_options::retain_handling_e::send // Send retained messages when the subscription is established.
}
};
// Subscribe to a single Topic.
auto [sub_codes, sub_props] = co_await client.async_subscribe(
sub_topic, async_mqtt5::subscribe_props {}, asio::use_awaitable
);
// Note: you can subscribe to multiple Topics in one mqtt_client::async_subscribe call.
// std::vector<async_mqtt5::reason_code> sub_codes contain the result of the subscribe action for every Topic.
// Before attempting to receive an Application Message from the Topic we just subscribed to,
// it is advisable to verify that the subscription succeeded.
// It is not recommended to call mqtt_client::async_receive if you do not have any
// Subscription established as the corresponding handler will never be invoked.
if (!sub_codes[0])
auto [topic, payload, publish_props] = co_await client.async_receive(asio::use_awaitable);
// Receive more messages...
// Unsubscribe from the Topic.
// Similar to mqtt_client::async_subscribe call, std::vector<async_mqtt5::reason_code> unsub_codes contain
// the result of the unsubscribe action for every Topic.
auto [unsub_codes, unsub_props] = co_await client.async_unsubscribe(
"test/mqtt-test", async_mqtt5::unsubscribe_props {},
asio::use_awaitable
);
// Note: you can unsubscribe from multiple Topics in one mqtt_client::async_unsubscribe call.
// Disconnect the Client.
co_await client.async_disconnect(
async_mqtt5::disconnect_rc_e::disconnect_with_will_message,
async_mqtt5::disconnect_props {},
asio::use_awaitable
);
co_return;
}
int main() {
// Initialise execution context.
asio::io_context ioc;
// Spawn the coroutine.
asio::co_spawn(ioc, client_receiver(ioc), asio::detached);
// Start the execution.
ioc.run();
}
//]