Cleaned up some of the sample apps. Rejiggered create_options contructors, breaking previous commits

src/async_client.cpp

@@ -54,7 +54,7 @@ async_client::async_client(const string& serverURI, const string& clientId,
 				: serverURI_(serverURI), clientId_(clientId),
 					mqttVersion_(MQTTVERSION_DEFAULT), userCallback_(nullptr)
 {
-	create_options opts(maxBufferedMessages);
+	create_options opts(MQTTVERSION_DEFAULT, maxBufferedMessages);
 
 	int rc = MQTTAsync_createWithOptions(&cli_, serverURI.c_str(), clientId.c_str(),
 										 MQTTCLIENT_PERSISTENCE_DEFAULT,
@@ -71,7 +71,7 @@ async_client::async_client(const string& serverURI, const string& clientId,
 						   iclient_persistence* persistence /*=nullptr*/,
 						   ipersistence_encoder* encoder /*=nullptr*/)
 				: async_client(serverURI, clientId,
-							   create_options(maxBufferedMessages),
+							   create_options(MQTTVERSION_DEFAULT, maxBufferedMessages),
 							   persistence, encoder)
 {
 }

src/client.cpp

@@ -31,35 +31,50 @@ const std::chrono::minutes client::DFLT_TIMEOUT = std::chrono::minutes(5);
 
 /////////////////////////////////////////////////////////////////////////////
 
-
 client::client(const string& serverURI, const string& clientId,
-			   iclient_persistence* persistence /*=nullptr*/)
-		: cli_(serverURI, clientId, persistence),
+			   iclient_persistence* persistence /*=nullptr*/,
+			   ipersistence_encoder* encoder /*=nullptr*/)
+		: cli_(serverURI, clientId, persistence, encoder),
 			timeout_(DFLT_TIMEOUT), userCallback_(nullptr)
 {
 }
 
 client::client(const string& serverURI, const string& clientId,
-			   const string& persistDir)
-		: cli_(serverURI, clientId, persistDir),
+			   const string& persistDir,
+			   ipersistence_encoder* encoder /*=nullptr*/)
+		: cli_(serverURI, clientId, persistDir, encoder),
 			timeout_(DFLT_TIMEOUT), userCallback_(nullptr)
 {
 }
 
 client::client(const string& serverURI, const string& clientId,
-			   int maxBufferedMessages, iclient_persistence* persistence /*=nullptr*/)
-		: cli_(serverURI, clientId, maxBufferedMessages, persistence),
+			   int maxBufferedMessages,
+			   iclient_persistence* persistence /*=nullptr*/,
+			   ipersistence_encoder* encoder /*=nullptr*/)
+		: cli_(serverURI, clientId, maxBufferedMessages, persistence, encoder),
 			timeout_(DFLT_TIMEOUT), userCallback_(nullptr)
 {
 }
 
 client::client(const string& serverURI, const string& clientId,
-			   int maxBufferedMessages, const string& persistDir)
-		: cli_(serverURI, clientId, maxBufferedMessages, persistDir),
+			   int maxBufferedMessages, const string& persistDir,
+			   ipersistence_encoder* encoder /*=nullptr*/)
+		: cli_(serverURI, clientId, maxBufferedMessages, persistDir, encoder),
 			timeout_(DFLT_TIMEOUT), userCallback_(nullptr)
 {
 }
 
+client::client(const string& serverURI, const string& clientId,
+			   const create_options& opts,
+			   iclient_persistence* persistence /*=nullptr*/,
+			   ipersistence_encoder* encoder /*=nullptr*/)
+		: cli_(serverURI, clientId, opts, persistence, encoder),
+			timeout_(DFLT_TIMEOUT), userCallback_(nullptr)
+{
+}
+
+// --------------------------------------------------------------------------
+
 void client::set_callback(callback& cb)
 {
 	userCallback_ = &cb;

src/connect_options.cpp

@@ -25,7 +25,8 @@ namespace mqtt {
 
 /////////////////////////////////////////////////////////////////////////////
 
-const MQTTAsync_connectOptions connect_options::DFLT_C_STRUCT = MQTTAsync_connectOptions_initializer;
+const MQTTAsync_connectOptions connect_options::DFLT_C_STRUCT =
+		MQTTAsync_connectOptions_initializer;
 
 connect_options::connect_options() : opts_(DFLT_C_STRUCT)
 {

src/create_options.cpp

@@ -22,11 +22,18 @@ namespace mqtt {
 /////////////////////////////////////////////////////////////////////////////
 
 const MQTTAsync_createOptions create_options::DFLT_C_STRUCT =
-			MQTTAsync_createOptions_initializer5;
+			MQTTAsync_createOptions_initializer;
 
 
-create_options::create_options(int maxBufferedMessages) : create_options()
+create_options::create_options(int mqttVersion) : create_options()
 {
+	opts_.MQTTVersion = mqttVersion;
+}
+
+create_options::create_options(int mqttVersion, int maxBufferedMessages) : create_options()
+{
+	opts_.MQTTVersion = mqttVersion;
+
 	if (maxBufferedMessages != 0) {
 		opts_.sendWhileDisconnected = to_int(true);
 		opts_.maxBufferedMessages = maxBufferedMessages;

src/mqtt/async_client.h

@@ -6,7 +6,7 @@
 /////////////////////////////////////////////////////////////////////////////
 
 /*******************************************************************************
- * Copyright (c) 2013-2019 Frank Pagliughi <fpagliughi@mindspring.com>
+ * Copyright (c) 2013-2020 Frank Pagliughi <fpagliughi@mindspring.com>
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -253,8 +253,7 @@ public:
 	 *  				as a URI.
 	 * @param clientId a client identifier that is unique on the server
 	 *  			   being connected to
-	 * @param maxBufferedMessages the maximum number of messages allowed to
-	 *  						  be buffered while not connected
+	 * @param opts The create options
 	 * @param persistence The user persistence structure. If this is null,
 	 *  				  then no persistence is used.
 	 * @param encoder An object to encode and decode the persistence data.

src/mqtt/client.h

@@ -104,7 +104,8 @@ public:
 	 *  				  then no persistence is used.
 	 */
 	client(const string& serverURI, const string& clientId,
-		   iclient_persistence* persistence=nullptr);
+		   iclient_persistence* persistence=nullptr,
+		   ipersistence_encoder* encoder=nullptr);
 	/**
 	 * Create an async_client that can be used to communicate with an MQTT
 	 * server.
@@ -116,7 +117,8 @@ public:
 	 * @param persistDir The directory to use for persistence data
 	 */
 	client(const string& serverURI, const string& clientId,
-		   const string& persistDir);
+		   const string& persistDir,
+		   ipersistence_encoder* encoder=nullptr);
 	/**
 	 * Create a client that can be used to communicate with an MQTT server,
 	 * which allows for off-line message buffering.
@@ -132,7 +134,9 @@ public:
 	 *  				  then no persistence is used.
 	 */
 	client(const string& serverURI, const string& clientId,
-		   int maxBufferedMessages, iclient_persistence* persistence=nullptr);
+		   int maxBufferedMessages,
+		   iclient_persistence* persistence=nullptr,
+		   ipersistence_encoder* encoder=nullptr);
 	/**
 	 * Create a client that can be used to communicate with an MQTT server,
 	 * which allows for off-line message buffering.
@@ -144,9 +148,30 @@ public:
 	 * @param maxBufferedMessages the maximum number of messages allowed to
 	 *  						  be buffered while not connected
 	 * @param persistDir The directory to use for persistence data
+	 * @param encoder An object to encode and decode the persistence data.
 	 */
 	client(const string& serverURI, const string& clientId,
-		   int maxBufferedMessages, const string& persistDir);
+		   int maxBufferedMessages, const string& persistDir,
+		   ipersistence_encoder* encoder=nullptr);
+	/**
+	 * Create an async_client that can be used to communicate with an MQTT
+	 * server, which allows for off-line message buffering.
+	 * This allows the caller to specify a user-defined persistence object,
+	 * or use no persistence.
+	 * @param serverURI the address of the server to connect to, specified
+	 *  				as a URI.
+	 * @param clientId a client identifier that is unique on the server
+	 *  			   being connected to
+	 * @param opts The create options
+	 * @param persistence The user persistence structure. If this is null,
+	 *  				  then no persistence is used.
+	 * @param encoder An object to encode and decode the persistence data.
+	 * @param encoder An object to encode and decode the persistence data.
+	 */
+	client(const string& serverURI, const string& clientId,
+		   const create_options& opts,
+		   iclient_persistence* persistence=nullptr,
+		   ipersistence_encoder* encoder=nullptr);
 	/**
 	 * Virtual destructor
 	 */

src/mqtt/create_options.h

@@ -56,12 +56,18 @@ public:
 	 * Default set of client create options.
 	 */
 	create_options() : opts_(DFLT_C_STRUCT) {}
+	/**
+	 * Default create optionsfor the specified version of MQTT.
+	 * @param mqttVersion The MQTT version used to create the client.
+	 */
+	explicit create_options(int mqttVersion);
 	/**
 	 * Default create options, but with off-line buffering enabled.
+	 * @param mqttVersion The MQTT version used to create the client.
 	 * @param maxBufferedMessages the maximum number of messages allowed to
 	 *  						  be buffered while not connected
 	 */
-	explicit create_options(int maxBufferedMessages);
+	create_options(int mqttVersion, int maxBufferedMessages);
 	/**
 	 * Gets whether the client will accept message to publish while
 	 * disconnected.

src/samples/async_publish.cpp

@@ -32,7 +32,7 @@
 #include <iostream>
 #include <cstdlib>
 #include <string>
-#include <thread>	// For sleep
+#include <thread>
 #include <atomic>
 #include <chrono>
 #include <cstring>
@@ -132,16 +132,15 @@ int main(int argc, char* argv[])
 	callback cb;
 	client.set_callback(cb);
 
-	mqtt::connect_options conopts;
-	mqtt::message willmsg(TOPIC, LWT_PAYLOAD, 1, true);
-	mqtt::will_options will(willmsg);
-	conopts.set_will(will);
+	auto connopts = mqtt::connect_options_builder()
+		.will(mqtt::message(TOPIC, LWT_PAYLOAD, QOS))
+		.finalize();
 
 	cout << "  ...OK" << endl;
 
 	try {
 		cout << "\nConnecting..." << endl;
-		mqtt::token_ptr conntok = client.connect(conopts);
+		mqtt::token_ptr conntok = client.connect(connopts);
 		cout << "Waiting for the connection..." << endl;
 		conntok->wait();
 		cout << "  ...OK" << endl;

src/samples/async_subscribe.cpp

@@ -168,12 +168,12 @@ public:
 
 int main(int argc, char* argv[])
 {
+	mqtt::async_client client(SERVER_ADDRESS, CLIENT_ID);
+
 	mqtt::connect_options connOpts;
 	connOpts.set_keep_alive_interval(20);
 	connOpts.set_clean_session(true);
 
-	mqtt::async_client client(SERVER_ADDRESS, CLIENT_ID);
-
 	callback cb(client, connOpts);
 	client.set_callback(cb);
 

src/samples/data_publish.cpp

@@ -113,10 +113,11 @@ int main(int argc, char* argv[])
 	mqtt::async_client cli(address, CLIENT_ID, MAX_BUFFERED_MSGS,
 						   PERSIST_DIR, &encoder);
 
-	mqtt::connect_options connOpts;
-	connOpts.set_keep_alive_interval(MAX_BUFFERED_MSGS * PERIOD);
-	connOpts.set_clean_session(true);
-	connOpts.set_automatic_reconnect(true);
+	auto connOpts = mqtt::connect_options_builder()
+		.keep_alive_interval(MAX_BUFFERED_MSGS * PERIOD)
+		.clean_session(true)
+		.automatic_reconnect(true)
+		.finalize();
 
 	// Create a topic object. This is a conventience since we will
 	// repeatedly publish messages with the same parameters.

src/samples/mqttpp_chat.cpp

@@ -70,17 +70,19 @@ int main(int argc, char* argv[])
 
 	// LWT message is broadcast to other users if out connection is lost
 
-	auto lwt = mqtt::make_message(chatTopic, "<<<"+chatUser+" was disconnected>>>", QOS, false);
+	auto lwt = mqtt::message(chatTopic, "<<<"+chatUser+" was disconnected>>>", QOS, false);
 
 	// Set up the connect options
 
-	mqtt::connect_options connOpts;
-	connOpts.set_keep_alive_interval(20);
-	connOpts.set_mqtt_version(MQTTVERSION_5);
-	connOpts.set_clean_start(true);
-	connOpts.set_will_message(lwt);
+	auto connOpts = mqtt::connect_options_builder()
+		.keep_alive_interval(std::chrono::seconds(20))
+		.mqtt_version(MQTTVERSION_5)
+		.clean_start(true)
+		.will(std::move(lwt))
+		.finalize();
 
-	mqtt::async_client cli(SERVER_ADDRESS, "");
+	mqtt::async_client cli(SERVER_ADDRESS, "",
+						   mqtt::create_options(MQTTVERSION_5));
 
 	// Set a callback for connection lost.
 	// This just exits the app.

src/samples/rpc_math_cli.cpp

@@ -34,7 +34,7 @@
 #include <sstream>
 #include <cstdlib>
 #include <string>
-#include <thread>	// For sleep
+#include <thread>
 #include <atomic>
 #include <chrono>
 #include <cstring>
@@ -59,9 +59,10 @@ int main(int argc, char* argv[])
 	constexpr int QOS = 1;
 	const string REQ_TOPIC_HDR { "requests/math/" };
 
-	mqtt::async_client cli(SERVER_ADDRESS, "");
+	mqtt::create_options createOpts(MQTTVERSION_5);
+	mqtt::async_client cli(SERVER_ADDRESS, "", createOpts);
 
-	auto connopts = mqtt::connect_options_builder()
+	auto connOpts = mqtt::connect_options_builder()
 					    .mqtt_version(MQTTVERSION_5)
 					    .clean_start()
 						.finalize();
@@ -70,7 +71,7 @@ int main(int argc, char* argv[])
 
 	try {
 		cout << "Connecting..." << flush;
-		mqtt::token_ptr tok = cli.connect(connopts);
+		mqtt::token_ptr tok = cli.connect(connOpts);
 		auto connRsp = tok->get_connect_response();
 		cout << "OK (" << connRsp.get_server_uri() << ")" << endl;
 

src/samples/rpc_math_srvr.cpp

@@ -91,11 +91,8 @@ double mult(const std::vector<double>& nums)
 
 int main(int argc, char* argv[])
 {
-	//auto createOpts = mqtt::create_options_builder()
-	//				      .mqtt_version(MQTTVERSION_5)
-	//					  .finalize();
-
-	mqtt::client cli(SERVER_ADDRESS, CLIENT_ID);
+	mqtt::create_options createOpts(MQTTVERSION_5);
+	mqtt::client cli(SERVER_ADDRESS, CLIENT_ID, createOpts);
 
 	auto connOpts = mqtt::connect_options_builder()
 					    .mqtt_version(MQTTVERSION_5)
@@ -115,6 +112,7 @@ int main(int argc, char* argv[])
 
 		// Consume messages
 
+		cout << "Waiting for RPC requests..." << endl;
 		while (true) {
 			auto msg = cli.consume_message();