Restored sync_consume with auto reconnect.

src/samples/sync_consume.cpp

@@ -0,0 +1,121 @@
+// sync_consume.cpp
+//
+// This is a Paho MQTT C++ client, sample application.
+//
+// This application is an MQTT consumer/subscriber using the C++ synchronous
+// client interface, which uses the queuing API to receive messages.
+//
+// The sample demonstrates:
+//  - Connecting to an MQTT server/broker
+//  - Subscribing to multiple topics
+//  - Receiving messages through the queueing consumer API
+//  - Recieving and acting upon commands via MQTT topics
+//  - Auto reconnect
+//  - Using a persistent (non-clean) session
+//
+
+/*******************************************************************************
+ * Copyright (c) 2013-2020 Frank Pagliughi <[email protected]>
+ *
+ * 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
+ *******************************************************************************/
+
+#include <iostream>
+#include <cstdlib>
+#include <string>
+#include <cstring>
+#include <cctype>
+#include <thread>
+#include <chrono>
+#include "mqtt/client.h"
+
+using namespace std;
+using namespace std::chrono;
+
+const string SERVER_ADDRESS	{ "tcp://localhost:1883" };
+const string CLIENT_ID		{ "sync_consume_cpp" };
+
+/////////////////////////////////////////////////////////////////////////////
+
+int main(int argc, char* argv[])
+{
+	auto connOpts = mqtt::connect_options_builder()
+		.keep_alive_interval(seconds(30))
+		.automatic_reconnect(seconds(2), seconds(30))
+		.clean_session(false)
+		.finalize();
+
+	mqtt::client cli(SERVER_ADDRESS, CLIENT_ID);
+
+	cli.set_update_connection_handler(
+		[](mqtt::connect_data& connData) {
+			if (!connData.get_user_name().empty())
+				cout << "User: " << connData.get_user_name() << endl;
+			return false;
+		}
+	);
+
+	const vector<string> TOPICS { "data/#", "command" };
+	const vector<int> QOS { 0, 1 };
+
+	try {
+		cout << "Connecting to the MQTT server..." << flush;
+		mqtt::connect_response rsp = cli.connect(connOpts);
+		cout << "OK\n" << endl;
+
+		if (!rsp.is_session_present()) {
+			std::cout << "Subscribing to topics..." << std::flush;
+			cli.subscribe(TOPICS, QOS);
+			std::cout << "OK" << std::endl;
+		}
+		else {
+			cout << "Session already present. Skipping subscribe." << std::endl;
+		}
+
+		// Consume messages
+
+		while (true) {
+			auto msg = cli.consume_message();
+
+			if (msg) {
+				if (msg->get_topic() == "command" &&
+						msg->to_string() == "exit") {
+					cout << "Exit command received" << endl;
+					break;
+				}
+
+				cout << msg->get_topic() << ": " << msg->to_string() << endl;
+			}
+			else if (!cli.is_connected()) {
+				cout << "Lost connection" << endl;
+				while (!cli.is_connected()) {
+					this_thread::sleep_for(milliseconds(250));
+				}
+				cout << "Re-established connection" << endl;
+			}
+		}
+
+		// Disconnect
+
+		cout << "\nDisconnecting from the MQTT server..." << flush;
+		cli.disconnect();
+		cout << "OK" << endl;
+	}
+	catch (const mqtt::exception& exc) {
+		cerr << exc.what() << endl;
+		return 1;
+	}
+
+ 	return 0;
+}
+