samples: nrf_cloud: switch to Zephyr MQTT

As MQTT library is now available in Zephyr upstream, align nrf_cloud
library to use it instead.

Signed-off-by: Robert Lubos <[email protected]>

Author: rlubos

applications/asset_tracker/prj.conf

@@ -18,11 +18,6 @@ CONFIG_NET_NATIVE=n
 CONFIG_NET_SOCKETS=y
 CONFIG_NET_SOCKETS_OFFLOAD=y
 
-# MQTT
-CONFIG_MQTT_SOCKET_LIB=y
-CONFIG_MQTT_LIB_TLS=y
-CONFIG_MQTT_MAX_PACKET_LENGTH=2048
-
 # LTE link control
 CONFIG_POWER_OPTIMIZATION_ENABLE=n
 CONFIG_LTE_LINK_CONTROL=y

applications/asset_tracker/prj_nrf9160_pca20035ns.conf

@@ -18,11 +18,6 @@ CONFIG_NET_NATIVE=n
 CONFIG_NET_SOCKETS=y
 CONFIG_NET_SOCKETS_OFFLOAD=y
 
-# MQTT
-CONFIG_MQTT_SOCKET_LIB=y
-CONFIG_MQTT_LIB_TLS=y
-CONFIG_MQTT_MAX_PACKET_LENGTH=2048
-
 # LTE link control
 CONFIG_POWER_OPTIMIZATION_ENABLE=n
 CONFIG_LTE_LINK_CONTROL=y

applications/asset_tracker/prj_qemu_x86.conf

@@ -33,9 +33,6 @@ CONFIG_CONSOLE_SUBSYS=y
 CONFIG_CONSOLE_HANDLER=y
 CONFIG_CONSOLE_GETCHAR=y
 
-CONFIG_MQTT_SOCKET_LIB=y
-CONFIG_MQTT_MAX_PACKET_LENGTH=2048
-CONFIG_MQTT_LIB_TLS=y
 CONFIG_NET_SOCKETS_SOCKOPT_TLS=y
 
 # TLS configuration

include/net/mqtt_socket.h

@@ -320,7 +320,7 @@ struct mqtt_sec_config {
 	u32_t sec_tag_count;
 
 	/** Indicates the list of security tags to be used for the session. */
-	sec_tag_t *seg_tag_list;
+	sec_tag_t *sec_tag_list;
 
 	/** Peer hostname for ceritificate verification.
 	 *  May be NULL to skip hostname verification.

samples/nrf9160/lte_ble_gateway/prj.conf

@@ -15,11 +15,6 @@ CONFIG_NET_NATIVE=n
 CONFIG_NET_SOCKETS=y
 CONFIG_NET_SOCKETS_OFFLOAD=y
 
-# MQTT
-CONFIG_MQTT_SOCKET_LIB=y
-CONFIG_MQTT_LIB_TLS=y
-CONFIG_MQTT_MAX_PACKET_LENGTH=2048
-
 # LTE link control
 CONFIG_LTE_LINK_CONTROL=y
 CONFIG_LTE_AUTO_INIT_AND_CONNECT=n

subsys/net/lib/nrf_cloud/Kconfig

@@ -6,7 +6,8 @@
 menuconfig NRF_CLOUD
 	bool "nRF Cloud library"
 	select CJSON_LIB
-	select MQTT_SOCKET_LIB
+	select MQTT_LIB
+	select MQTT_LIB_TLS
 
 if NRF_CLOUD
 
@@ -45,6 +46,17 @@ config NRF_CLOUD_PORT
 config NRF_CLOUD_IPV6
 	bool "Configure nRF Cloud library to use IPv6 addressing. Otherwise IPv4 is used."
 
+config NRF_CLOUD_MQTT_MESSAGE_BUFFER_LEN
+	int "Size of the buffer for MQTT library."
+	default 256
+	help
+		Specifies maximum message size can be transmitted/received through
+		MQTT (exluding MQTT PUBLISH payload).
+
+config NRF_CLOUD_MQTT_PAYLOAD_BUFFER_LEN
+	int "Size of the buffer for MQTT PUBLISH payload."
+	default 2048
+
 module=NRF_CLOUD
 module-dep=LOG
 module-str=Log level for nRF Cloud

subsys/net/lib/nrf_cloud/src/nrf_cloud_transport.c

@@ -9,7 +9,7 @@
 
 #include <zephyr.h>
 #include <stdio.h>
-#include <net/mqtt_socket.h>
+#include <net/mqtt.h>
 #include <net/socket.h>
 #include <logging/log.h>
 #include <misc/util.h>
@@ -90,6 +90,9 @@ static struct nct {
 	struct mqtt_utf8 dc_rx_endp;
 	struct mqtt_utf8 dc_m_endp;
 	u32_t message_id;
+	u8_t rx_buf[CONFIG_NRF_CLOUD_MQTT_MESSAGE_BUFFER_LEN];
+	u8_t tx_buf[CONFIG_NRF_CLOUD_MQTT_MESSAGE_BUFFER_LEN];
+	u8_t payload_buf[CONFIG_NRF_CLOUD_MQTT_PAYLOAD_BUFFER_LEN];
 } nct;
 
 static const struct mqtt_topic nct_cc_rx_list[] = {
@@ -343,7 +346,7 @@ static int nct_provision(void)
 	nct.tls_config.cipher_count = 0;
 	nct.tls_config.cipher_list = NULL;
 	nct.tls_config.sec_tag_count = ARRAY_SIZE(sec_tag_list);
-	nct.tls_config.seg_tag_list = sec_tag_list;
+	nct.tls_config.sec_tag_list = sec_tag_list;
 	nct.tls_config.hostname = NRF_CLOUD_HOSTNAME;
 
 #if defined(CONFIG_NRF_CLOUD_PROVISION_CERTIFICATES)
@@ -446,6 +449,10 @@ int nct_mqtt_connect(void)
 	nct.client.user_name = NULL;
 #if defined(CONFIG_MQTT_LIB_TLS)
 	nct.client.transport.type = MQTT_TRANSPORT_SECURE;
+	nct.client.rx_buf = nct.rx_buf;
+	nct.client.rx_buf_size = sizeof(nct.rx_buf);
+	nct.client.tx_buf = nct.tx_buf;
+	nct.client.tx_buf_size = sizeof(nct.tx_buf);
 
 	struct mqtt_sec_config *tls_config = &nct.client.transport.tls.config;
 
@@ -457,6 +464,15 @@ int nct_mqtt_connect(void)
 	return mqtt_connect(&nct.client);
 }
 
+static int publish_get_payload(struct mqtt_client *client, size_t length)
+{
+	if (length > sizeof(nct.payload_buf)) {
+		return -EMSGSIZE;
+	}
+
+	return mqtt_readall_publish_payload(client, nct.payload_buf, length);
+}
+
 /* Handle MQTT events. */
 static void nct_mqtt_evt_handler(struct mqtt_client *const mqtt_client,
 				 const struct mqtt_evt *_mqtt_evt)
@@ -483,14 +499,24 @@ static void nct_mqtt_evt_handler(struct mqtt_client *const mqtt_client,
 			p->message_id,
 			p->message.payload.len);
 
+		int err = publish_get_payload(mqtt_client,
+					      p->message.payload.len);
+
+		if (err < 0) {
+			LOG_ERR("publish_get_payload: failed %d", err);
+			mqtt_disconnect(mqtt_client);
+			event_notify = false;
+			break;
+		}
+
 		/* If the data arrives on one of the subscribed control channel
 		 * topic. Then we notify the same.
 		 */
 		if (control_channel_topic_match(
 			NCT_RX_LIST, &p->message.topic, &cc.opcode)) {
 
 			cc.id = p->message_id;
-			cc.data.ptr = p->message.payload.data;
+			cc.data.ptr = nct.payload_buf;
 			cc.data.len = p->message.payload.len;
 
 			evt.type = NCT_EVT_CC_RX_DATA;
@@ -582,12 +608,7 @@ int nct_init(void)
 		return err;
 	}
 
-	err = nct_provision();
-	if (err) {
-		return err;
-	}
-
-	return mqtt_init();
+	return nct_provision();
 }
 
 #if defined(CONFIG_NRF_CLOUD_STATIC_IPV4)
@@ -845,7 +866,7 @@ int nct_disconnect(void)
 void nct_process(void)
 {
 	mqtt_input(&nct.client);
-	mqtt_live();
+	mqtt_live(&nct.client);
 }
 
 int nct_socket_get(void)