Add client support for MQTT v3.1.1.

ChangeLog.txt

@@ -18,6 +18,7 @@ Important changes:
 - The client library and clients now have experimental SOCKS5 support.
 - Wildcard TLS certificates are now supported for bridges and clients.
 - The clients have support for config files with default options.
+- Client and client libraries have support for MQTT v3.1.1.
 
 
 Broker:
@@ -63,6 +64,7 @@ Clients:
 - Add --proxy SOCKS5 support for both clients.
 - Pub client supports setting its keepalive. Closes bug #454852.
 - Add support for config files with default options.
+- Add support for MQTT v3.1.1.
 
 Client library:
 - Add experimental SOCKS5 support.
@@ -71,6 +73,7 @@ Client library:
 - SRV support is now not compiled in by default.
 - Wildcard TLS certificates are now supported.
 - mosquittopp now has a virtual destructor. Closes bug #452915.
+- Add support for MQTT v3.1.1.
 
 
 1.3.5 - 20141008

client/client_shared.c

@@ -42,6 +42,7 @@ void init_config(struct mosq_config *cfg)
  cfg->keepalive = 60;
  cfg->clean_session = true;
  cfg->eol = true;
+ cfg->protocol_version = MQTT_PROTOCOL_v31;
 }
 
 void client_config_cleanup(struct mosq_config *cfg)
@@ -420,6 +421,21 @@ int client_config_line_proc(struct mosq_config *cfg, int pub_or_sub, int argc, c
  }else{
  cfg->pub_mode = MSGMODE_NULL;
  }
+ }else if(!strcmp(argv[i], "-V") || !strcmp(argv[i], "--protocol-version")){
+ if(i==argc-1){
+ fprintf(stderr, "Error: --protocol-version argument given but no version specified.\n\n");
+ return 1;
+ }else{
+ if(!strcmp(argv[i+1], "mqttv31")){
+ cfg->protocol_version = MQTT_PROTOCOL_v31;
+ }else if(!strcmp(argv[i+1], "mqttv311")){
+ cfg->protocol_version = MQTT_PROTOCOL_v311;
+ }else{
+ fprintf(stderr, "Error: Invalid protocol version argument given.\n\n");
+ return 1;
+ }
+ i++;
+ }
 #ifdef WITH_SOCKS
  }else if(!strcmp(argv[i], "--proxy")){
  if(i==argc-1){
@@ -668,6 +684,7 @@ int client_opts_set(struct mosquitto *mosq, struct mosq_config *cfg)
  }
  }
 #endif
+ mosquitto_opts_set(mosq, MOSQ_OPT_PROTOCOL_VERSION, &(cfg->protocol_version));
  return MOSQ_ERR_SUCCESS;
 }
 

client/client_shared.h

@@ -33,6 +33,7 @@ and the Eclipse Distribution License is available at
 struct mosq_config {
  char *id;
  char *id_prefix;
+ int protocol_version;
  int keepalive;
  char *host;
  int port;

client/pub_client.c

@@ -245,6 +245,8 @@ void print_usage(void)
 #endif
  printf(" -t : mqtt topic to publish to.\n");
  printf(" -u : provide a username (requires MQTT 3.1 broker)\n");
+ printf(" -V : specify the version of the MQTT protocol to use when connecting.\n");
+ printf(" Can be mqttv31 or mqttv311. Defaults to mqttv31.\n");
  printf(" --help : display this message.\n");
  printf(" --quiet : don't print error messages.\n");
  printf(" --will-payload : payload for the client Will, which is sent by the broker in case of\n");

client/sub_client.c

@@ -169,6 +169,8 @@ void print_usage(void)
  printf(" -T : topic string to filter out of results. May be repeated.\n");
  printf(" -u : provide a username (requires MQTT 3.1 broker)\n");
  printf(" -v : print published messages verbosely.\n");
+ printf(" -V : specify the version of the MQTT protocol to use when connecting.\n");
+ printf(" Can be mqttv31 or mqttv311. Defaults to mqttv31.\n");
  printf(" --help : display this message.\n");
  printf(" --quiet : don't print error messages.\n");
  printf(" --will-payload : payload for the client Will, which is sent by the broker in case of\n");

lib/cpp/mosquittopp.cpp

@@ -264,6 +264,11 @@ bool mosquittopp::want_write()
  return mosquitto_want_write(m_mosq);
 }
 
+int mosquittopp::opts_set(enum mosq_opt_t option, void *value)
+{
+ return mosquitto_opts_set(m_mosq, option, value);
+}
+
 int mosquittopp::threaded_set(bool threaded)
 {
  return mosquitto_threaded_set(m_mosq, threaded);

lib/cpp/mosquittopp.h

@@ -78,6 +78,7 @@ class mosqpp_EXPORT mosquittopp {
  int tls_opts_set(int cert_reqs, const char *tls_version=NULL, const char *ciphers=NULL);
  int tls_insecure_set(bool value);
  int tls_psk_set(const char *psk, const char *identity, const char *ciphers=NULL);
+ int opts_set(enum mosq_opt_t option, void *value);
 
  int loop(int timeout=-1, int max_packets=1);
  int loop_misc();

lib/linker.version

@@ -73,6 +73,7 @@ MOSQ_1.3 {
 MOSQ_1.4 {
  global:
  mosquitto_threaded_set;
+ mosquitto_opts_set;
  mosquitto_pub_topic_check;
  mosquitto_sub_topic_check;
  mosquitto_socks5_set;

lib/mosquitto.c

@@ -1154,6 +1154,29 @@ bool mosquitto_want_write(struct mosquitto *mosq)
  }
 }
 
+int mosquitto_opts_set(struct mosquitto *mosq, enum mosq_opt_t option, void *value)
+{
+ if(!mosq || !value) return MOSQ_ERR_INVAL;
+ int ival;
+
+ switch(option){
+ case MOSQ_OPT_PROTOCOL_VERSION:
+ ival = *((int *)value);
+ if(ival == MQTT_PROTOCOL_V31){
+ mosq->protocol = mosq_p_mqtt31;
+ }else if(ival == MQTT_PROTOCOL_V311){
+ mosq->protocol = mosq_p_mqtt311;
+ }else{
+ return MOSQ_ERR_INVAL;
+ }
+ break;
+ default:
+ return MOSQ_ERR_INVAL;
+ }
+ return MOSQ_ERR_SUCCESS;
+}
+
+
 void mosquitto_connect_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int))
 {
  pthread_mutex_lock(&mosq->callback_mutex);

lib/mosquitto.h

@@ -82,9 +82,17 @@ enum mosq_err_t {
  MOSQ_ERR_PROXY = 16
 };
 
+/* Error values */
+enum mosq_opt_t {
+ MOSQ_OPT_PROTOCOL_VERSION = 1,
+};
+
 /* MQTT specification restricts client ids to a maximum of 23 characters */
 #define MOSQ_MQTT_ID_MAX_LENGTH 23
 
+#define MQTT_PROTOCOL_V31 3
+#define MQTT_PROTOCOL_V311 4
+
 struct mosquitto_message{
  int mid;
  char *topic;
@@ -918,6 +926,25 @@ libmosq_EXPORT bool mosquitto_want_write(struct mosquitto *mosq);
  */
 libmosq_EXPORT int mosquitto_threaded_set(struct mosquitto *mosq, bool threaded);
 
+/*
+ * Function: mosquitto_opts_set
+ *
+ * Used to set options for the client.
+ *
+ * Parameters:
+ * mosq - a valid mosquitto instance.
+ * option - the option to set.
+ * value - the option specific value.
+ *
+ * Options:
+ * MOSQ_OPT_PROTOCOL_VERSION - value must be an int, set to either
+ * MQTT_PROTOCOL_V31 or MQTT_PROTOCOL_V311. Must
+ * be set before the client connects. Defaults to
+ * MQTT_PROTOCOL_V31.
+ */
+libmosq_EXPORT int mosquitto_opts_set(struct mosquitto *mosq, enum mosq_opt_t option, void *value);
+
+
 /*
  * Function: mosquitto_tls_set
  *

lib/send_client_mosq.c

@@ -36,7 +36,7 @@ int _mosquitto_send_connect(struct mosquitto *mosq, uint16_t keepalive, bool cle
  uint8_t will = 0;
  uint8_t byte;
  int rc;
- uint8_t version = PROTOCOL_VERSION_v31;
+ uint8_t version;
  char *clientid, *username, *password;
 
  assert(mosq);
@@ -58,6 +58,14 @@ int _mosquitto_send_connect(struct mosquitto *mosq, uint16_t keepalive, bool cle
  password = mosq->password;
 #endif
 
+ if(mosq->protocol == mosq_p_mqtt31){
+ version = MQTT_PROTOCOL_V31;
+ }else if(mosq->protocol == mosq_p_mqtt311){
+ version = MQTT_PROTOCOL_V311;
+ }else{
+ return MOSQ_ERR_INVAL;
+ }
+
  packet = _mosquitto_calloc(1, sizeof(struct _mosquitto_packet));
  if(!packet) return MOSQ_ERR_NOMEM;
 
@@ -84,7 +92,14 @@ int _mosquitto_send_connect(struct mosquitto *mosq, uint16_t keepalive, bool cle
  }
 
  /* Variable header */
- _mosquitto_write_string(packet, PROTOCOL_NAME_v31, strlen(PROTOCOL_NAME_v31));
+ if(version == MQTT_PROTOCOL_V31){
+ _mosquitto_write_string(packet, PROTOCOL_NAME_v31, strlen(PROTOCOL_NAME_v31));
+ }else if(version == MQTT_PROTOCOL_V311){
+ _mosquitto_write_string(packet, PROTOCOL_NAME_v311, strlen(PROTOCOL_NAME_v311));
+ }else{
+ _mosquitto_free(packet);
+ return MOSQ_ERR_INVAL;
+ }
 #if defined(WITH_BROKER) && defined(WITH_BRIDGE)
  if(mosq->bridge && mosq->bridge->try_private && mosq->bridge->try_private_accepted){
  version |= 0x80;

man/mosquitto_pub.1.xml

@@ -11,7 +11,7 @@
 
  <refnamediv>
  <refname>mosquitto_pub</refname>
- <refpurpose>an MQTT version 3.1 client for publishing simple messages</refpurpose>
+ <refpurpose>an MQTT version 3.1/3.1.1 client for publishing simple messages</refpurpose>
  </refnamediv>
 
  <refsynopsisdiv>
@@ -65,6 +65,7 @@
  </arg>
  </group>
  <arg><option>--proxy</option> <replaceable>socks-url</replaceable></arg>
+ <arg><option>-V</option> <replaceable>protocol-version</replaceable></arg>
 
  <arg choice='plain'><option>-t</option> <replaceable>message-topic</replaceable></arg>
  </cmdsynopsis>
@@ -374,6 +375,16 @@
  See also the <option>--pw</option> argument.</para>
  </listitem>
  </varlistentry>
+ <varlistentry>
+ <term><option>-V</option></term>
+ <term><option>--protocol-version</option></term>
+ <listitem>
+ <para>Specify which version of the MQTT protocol should be
+ used when connecting to the rmeote broker. Can be
+ <option>mqttv31</option> or <option>mqttv311</option>.
+ Defaults to <option>mqttv31</option>.</para>
+ </listitem>
+ </varlistentry>
  <varlistentry>
  <term><option>--will-payload</option></term>
  <listitem>

man/mosquitto_sub.1.xml

@@ -60,6 +60,7 @@
  </arg>
  </group>
  <arg><option>--proxy</option> <replaceable>socks-url</replaceable></arg>
+ <arg><option>-V</option> <replaceable>protocol-version</replaceable></arg>
  <arg choice='opt' rep='repeat'><option>-T</option> <replaceable>filter-out</replaceable></arg>
  <arg choice='plain' rep='repeat'><option>-t</option> <replaceable>message-topic</replaceable></arg>
  </cmdsynopsis>
@@ -414,6 +415,16 @@
  "payload".</para>
  </listitem>
  </varlistentry>
+ <varlistentry>
+ <term><option>-V</option></term>
+ <term><option>--protocol-version</option></term>
+ <listitem>
+ <para>Specify which version of the MQTT protocol should be
+ used when connecting to the rmeote broker. Can be
+ <option>mqttv31</option> or <option>mqttv311</option>.
+ Defaults to <option>mqttv31</option>.</para>
+ </listitem>
+ </varlistentry>
  <varlistentry>
  <term><option>--will-payload</option></term>
  <listitem>