embox

1
// async_publish.cpp
2
//
3
// This is a Paho MQTT C++ client, sample application.
4
//
5
// It's an example of how to send messages as an MQTT publisher using the
6
// C++ asynchronous client interface.
7
//
8
// The sample demonstrates:
9
//  - Connecting to an MQTT server/broker
10
//  - Publishing messages
11
//  - Default file persistence
12
//  - Last will and testament
13
//  - Using asynchronous tokens
14
//  - Implementing callbacks and action listeners
15
//
16

17
/*******************************************************************************
18
 * Copyright (c) 2013-2020 Frank Pagliughi <fpagliughi@mindspring.com>
19
 *
20
 * All rights reserved. This program and the accompanying materials
21
 * are made available under the terms of the Eclipse Public License v1.0
22
 * and Eclipse Distribution License v1.0 which accompany this distribution.
23
 *
24
 * The Eclipse Public License is available at
25
 *    http://www.eclipse.org/legal/epl-v10.html
26
 * and the Eclipse Distribution License is available at
27
 *   http://www.eclipse.org/org/documents/edl-v10.php.
28
 *
29
 * Contributors:
30
 *    Frank Pagliughi - initial implementation and documentation
31
 *******************************************************************************/
32

33
#include <iostream>
34
#include <cstdlib>
35
#include <string>
36
#include <thread>
37
#include <atomic>
38
#include <chrono>
39
#include <cstring>
40
#include "mqtt/async_client.h"
41

42
#include <paho_mqtt_cpp_publish.inc>
43

44
using namespace std;
45

46
static const string DFLT_SERVER_ADDRESS	{ MQTT_SERVER_ADDRESS };
47
static const string CLIENT_ID				{ "paho_cpp_async_publish" };
48
static const string TOPIC { MQTT_SERVER_TOPIC };
49

50
static const char* PAYLOAD1 = "Hello World!";
51
static const char* PAYLOAD2 = "Hi there!";
52
static const char* PAYLOAD3 = "Is anyone listening?";
53
static const char* PAYLOAD4 = "Someone is always listening.";
54

55
static const char* LWT_PAYLOAD = "Last will and testament.";
56

57
static const int  QOS = 1;
58

59
static const auto TIMEOUT = std::chrono::seconds(10);
60

61
/////////////////////////////////////////////////////////////////////////////
62

63
/**
64
 * A callback class for use with the main MQTT client.
65
 */
66
class callback : public virtual mqtt::callback
67
{
68
public:
69
	void connection_lost(const string& cause) override {
70
		cout << "\nConnection lost" << endl;
71
		if (!cause.empty())
72
			cout << "\tcause: " << cause << endl;
73
	}
74

75
	void delivery_complete(mqtt::delivery_token_ptr tok) override {
76
		cout << "\tDelivery complete for token: "
77
			<< (tok ? tok->get_message_id() : -1) << endl;
78
	}
79
};
80

81
/////////////////////////////////////////////////////////////////////////////
82

83
/**
84
 * A base action listener.
85
 */
86
class action_listener : public virtual mqtt::iaction_listener
87
{
88
protected:
89
	void on_failure(const mqtt::token& tok) override {
90
		cout << "\tListener failure for token: "
91
			<< tok.get_message_id() << endl;
92
	}
93

94
	void on_success(const mqtt::token& tok) override {
95
		cout << "\tListener success for token: "
96
			<< tok.get_message_id() << endl;
97
	}
98
};
99

100
/////////////////////////////////////////////////////////////////////////////
101

102
/**
103
 * A derived action listener for publish events.
104
 */
105
class delivery_action_listener : public action_listener
106
{
107
	atomic<bool> done_;
108

109
	void on_failure(const mqtt::token& tok) override {
110
		action_listener::on_failure(tok);
111
		done_ = true;
112
	}
113

114
	void on_success(const mqtt::token& tok) override {
115
		action_listener::on_success(tok);
116
		done_ = true;
117
	}
118

119
public:
120
	delivery_action_listener() : done_(false) {}
121
	bool is_done() const { return done_; }
122
};
123

124
/////////////////////////////////////////////////////////////////////////////
125

126
int main(int argc, char* argv[])
127
{
128
	// A client that just publishes normally doesn't need a persistent
129
	// session or Client ID unless it's using persistence, then the local
130
	// library requires an ID to identify the persistence files.
131

132
	string	address  = (argc > 1) ? string(argv[1]) : DFLT_SERVER_ADDRESS,
133
			clientID = (argc > 2) ? string(argv[2]) : CLIENT_ID;
134

135
	cout << "Initializing for server '" << address << "'..." << endl;
136
	mqtt::async_client client(address, clientID);
137

138
	callback cb;
139
	client.set_callback(cb);
140

141
	auto connOpts = mqtt::connect_options_builder()
142
		.clean_session()
143
		.will(mqtt::message(TOPIC, LWT_PAYLOAD, QOS))
144
		.finalize();
145

146
	cout << "  ...OK" << endl;
147

148
	try {
149
		cout << "\nConnecting..." << endl;
150
		mqtt::token_ptr conntok = client.connect(connOpts);
151
		cout << "Waiting for the connection..." << endl;
152
		conntok->wait();
153
		cout << "  ...OK" << endl;
154

155
		// First use a message pointer.
156

157
		cout << "\nSending message..." << endl;
158
		mqtt::message_ptr pubmsg = mqtt::make_message(TOPIC, PAYLOAD1);
159
		pubmsg->set_qos(QOS);
160
		client.publish(pubmsg)->wait_for(TIMEOUT);
161
		cout << "  ...OK" << endl;
162

163
		// Now try with itemized publish.
164

165
		cout << "\nSending next message..." << endl;
166
		mqtt::delivery_token_ptr pubtok;
167
		pubtok = client.publish(TOPIC, PAYLOAD2, strlen(PAYLOAD2), QOS, false);
168
		cout << "  ...with token: " << pubtok->get_message_id() << endl;
169
		cout << "  ...for message with " << pubtok->get_message()->get_payload().size()
170
			<< " bytes" << endl;
171
		pubtok->wait_for(TIMEOUT);
172
		cout << "  ...OK" << endl;
173

174
		// Now try with a listener
175

176
		cout << "\nSending next message..." << endl;
177
		action_listener listener;
178
		pubmsg = mqtt::make_message(TOPIC, PAYLOAD3);
179
		pubtok = client.publish(pubmsg, nullptr, listener);
180
		pubtok->wait();
181
		cout << "  ...OK" << endl;
182

183
		// Finally try with a listener, but no token
184

185
		cout << "\nSending final message..." << endl;
186
		delivery_action_listener deliveryListener;
187
		pubmsg = mqtt::make_message(TOPIC, PAYLOAD4);
188
		client.publish(pubmsg, nullptr, deliveryListener);
189

190
		while (!deliveryListener.is_done()) {
191
			this_thread::sleep_for(std::chrono::milliseconds(100));
192
		}
193
		cout << "OK" << endl;
194

195
		// Double check that there are no pending tokens
196

197
		auto toks = client.get_pending_delivery_tokens();
198
		if (!toks.empty())
199
			cout << "Error: There are pending delivery tokens!" << endl;
200

201
		// Disconnect
202
		cout << "\nDisconnecting..." << endl;
203
		client.disconnect()->wait();
204
		cout << "  ...OK" << endl;
205
	}
206
	catch (const mqtt::exception& exc) {
207
		cerr << exc.what() << endl;
208
		return 1;
209
	}
210

211
 	return 0;
212
}
213