From f5591256d5fa4465e87339bf161962573cf603e8 Mon Sep 17 00:00:00 2001
From: rechrtb <rechrtb@gmail.com>
Date: Thu, 24 Nov 2022 17:54:57 +0800
Subject: [PATCH] MQTT proof-of-concept for outgoing connections

---
 .cproject                                    |   5 +
 src/Config/Pins.h                            |   4 +
 src/Config/Pins_Duet3Mini.h                  |   1 +
 src/GCodes/GCodes2.cpp                       |  14 +-
 src/Networking/ESP8266WiFi/WiFiInterface.cpp |   8 +
 src/Networking/MqttClient.cpp                | 276 +++++++++++++++++++
 src/Networking/MqttClient.h                  | 171 ++++++++++++
 src/Networking/Network.cpp                   |  28 ++
 src/Networking/Network.h                     |   7 +
 src/Networking/NetworkClient.cpp             |  12 +-
 src/Networking/NetworkClient.h               |   4 -
 src/Networking/NetworkDefs.h                 |  11 +-
 src/Networking/NetworkResponder.h            |   5 +-
 src/Platform/MessageType.h                   |   1 +
 src/Platform/Platform.cpp                    |   5 +
 15 files changed, 537 insertions(+), 15 deletions(-)
 create mode 100644 src/Networking/MqttClient.cpp
 create mode 100644 src/Networking/MqttClient.h

diff --git a/.cproject b/.cproject
index 550b2484a6..09132ac254 100644
--- a/.cproject
+++ b/.cproject
@@ -1163,6 +1163,7 @@
 								<option id="gnu.c.compiler.option.misc.other.1176394435" name="Other flags" superClass="gnu.c.compiler.option.misc.other" useByScannerDiscovery="true" value="-c -mcpu=cortex-m4 -mthumb -mfpu=fpv4-sp-d16 -mfloat-abi=hard -mfp16-format=ieee -ffunction-sections -fdata-sections -nostdlib -Wundef -Wdouble-promotion -Werror=return-type -Werror=implicit -fsingle-precision-constant &quot;-Wa,-ahl=$*.s&quot;" valueType="string"/>
 								<option IS_BUILTIN_EMPTY="false" IS_VALUE_EMPTY="false" id="gnu.c.compiler.option.include.paths.468883064" name="Include paths (-I)" superClass="gnu.c.compiler.option.include.paths" useByScannerDiscovery="false" valueType="includePath">
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/CoreN2G}&quot;"/>
+									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/RRFLibraries/src/ThirdParty/MQTT-C/include}&quot;"/>
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/CoreN2G/src}&quot;"/>
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/CoreN2G/src//SAME5x_C21/SAME5x}&quot;"/>
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/CoreN2G/src/atmel/SAME54_DFP/1.1.134/include}&quot;"/>
@@ -1174,6 +1175,7 @@
 								</option>
 								<option IS_BUILTIN_EMPTY="false" IS_VALUE_EMPTY="false" id="gnu.c.compiler.option.preprocessor.def.symbols.126387254" name="Defined symbols (-D)" superClass="gnu.c.compiler.option.preprocessor.def.symbols" useByScannerDiscovery="false" valueType="definedSymbols">
 									<listOptionValue builtIn="false" value="__SAME54P20A__"/>
+									<listOptionValue builtIn="false" value="MQTTC_PAL_FILE=${workspace_loc:/RepRapFirmware/src/Networking/MqttClient.h}"/>
 									<listOptionValue builtIn="false" value="RTOS"/>
 									<listOptionValue builtIn="false" value="DUET3MINI_V04"/>
 									<listOptionValue builtIn="false" value="noexcept="/>
@@ -1212,6 +1214,8 @@
 								<option id="gnu.cpp.compiler.option.other.other.1951413447" name="Other flags" superClass="gnu.cpp.compiler.option.other.other" useByScannerDiscovery="true" value="-c -mcpu=cortex-m4 -mthumb -mfpu=fpv4-sp-d16 -mfloat-abi=hard -mfp16-format=ieee -ffunction-sections -fdata-sections -fno-threadsafe-statics -fno-rtti -fexceptions -nostdlib -Wundef -Wdouble-promotion -Werror=return-type -Wsuggest-override -fsingle-precision-constant &quot;-Wa,-ahl=$*.s&quot; -fstack-usage" valueType="string"/>
 								<option IS_BUILTIN_EMPTY="false" IS_VALUE_EMPTY="false" id="gnu.cpp.compiler.option.include.paths.1562950186" name="Include paths (-I)" superClass="gnu.cpp.compiler.option.include.paths" useByScannerDiscovery="false" valueType="includePath">
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/CoreN2G}&quot;"/>
+									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/RRFLibraries/src/ThirdParty/MQTT-C/include}&quot;"/>
+									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/RRFLibraries/MQTT}&quot;"/>
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/RRFLibraries}&quot;"/>
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/FreeRTOS/src/include}&quot;"/>
 									<listOptionValue builtIn="false" value="&quot;${workspace_loc:/FreeRTOS/src/portable/GCC/ARM_CM4F}&quot;"/>
@@ -1237,6 +1241,7 @@
 								</option>
 								<option IS_BUILTIN_EMPTY="false" IS_VALUE_EMPTY="false" id="gnu.cpp.compiler.option.preprocessor.def.1275759816" name="Defined symbols (-D)" superClass="gnu.cpp.compiler.option.preprocessor.def" useByScannerDiscovery="false" valueType="definedSymbols">
 									<listOptionValue builtIn="false" value="__SAME54P20A__"/>
+									<listOptionValue builtIn="false" value="MQTTC_PAL_FILE=${workspace_loc:/RepRapFirmware/src/Networking/MqttClient.h}"/>
 									<listOptionValue builtIn="false" value="RTOS"/>
 									<listOptionValue builtIn="false" value="DUET3MINI_V04"/>
 									<listOptionValue builtIn="false" value="_XOPEN_SOURCE"/>
diff --git a/src/Config/Pins.h b/src/Config/Pins.h
index 90574b631d..24a60c09a2 100644
--- a/src/Config/Pins.h
+++ b/src/Config/Pins.h
@@ -195,6 +195,10 @@
 # define SUPPORT_TELNET			HAS_NETWORKING
 #endif
 
+#ifndef SUPPORT_MQTT
+# define SUPPORT_MQTT			HAS_NETWORKING
+#endif
+
 #ifndef SUPPORT_MULTICAST_DISCOVERY
 # define SUPPORT_MULTICAST_DISCOVERY	0
 #endif
diff --git a/src/Config/Pins_Duet3Mini.h b/src/Config/Pins_Duet3Mini.h
index 2dd19a1f4f..85c1fa6fc9 100644
--- a/src/Config/Pins_Duet3Mini.h
+++ b/src/Config/Pins_Duet3Mini.h
@@ -64,6 +64,7 @@ constexpr uint32_t IAP_IMAGE_START = 0x20038000;
 #define SUPPORT_OBJECT_MODEL	1
 #define SUPPORT_FTP				1
 #define SUPPORT_TELNET			1
+#define SUPPORT_MQTT			1
 #define SUPPORT_ASYNC_MOVES		1
 #define SUPPORT_PROBE_POINTS_FILE	1
 
diff --git a/src/GCodes/GCodes2.cpp b/src/GCodes/GCodes2.cpp
index 2601f7b9a2..b54d169101 100644
--- a/src/GCodes/GCodes2.cpp
+++ b/src/GCodes/GCodes2.cpp
@@ -2028,6 +2028,10 @@ bool GCodes::HandleMcode(GCodeBuffer& gb, const StringRef& reply) THROWS(GCodeEx
 							type = Aux2Message;
 							break;
 #endif
+
+						case 6:		// MQTT
+							type = MqttMessage;
+							break;
 						default:
 							reply.printf("Invalid message type: %" PRIi32, param);
 							result = GCodeResult::error;
@@ -3787,11 +3791,17 @@ bool GCodes::HandleMcode(GCodeBuffer& gb, const StringRef& reply) THROWS(GCodeEx
 							{
 								const int port = (gb.Seen('R')) ? gb.GetIValue() : -1;
 								const int secure = (gb.Seen('T')) ? gb.GetIValue() : -1;
+
 								IPAddress ip;
-								if (gb.Seen('H'))
+
+								if (protocol == MqttProtocol)
 								{
-									gb.GetIPAddress(ip);
+									gb.MustSee('H');
+									{
+										gb.GetIPAddress(ip);
+									}
 								}
+
 								result = reprap.GetNetwork().EnableProtocol(interface, protocol, port,
 																			ip.GetV4LittleEndian(), secure, reply);
 							}
diff --git a/src/Networking/ESP8266WiFi/WiFiInterface.cpp b/src/Networking/ESP8266WiFi/WiFiInterface.cpp
index 8b2b9a7721..763a6ff5db 100644
--- a/src/Networking/ESP8266WiFi/WiFiInterface.cpp
+++ b/src/Networking/ESP8266WiFi/WiFiInterface.cpp
@@ -16,6 +16,7 @@
 #include <Networking/HttpResponder.h>
 #include <Networking/FtpResponder.h>
 #include <Networking/TelnetResponder.h>
+#include <Networking/MqttClient.h>
 #include "WifiFirmwareUploader.h"
 #include <General/IP4String.h>
 #include "WiFiSocket.h"
@@ -444,6 +445,9 @@ void WiFiInterface::Connect(NetworkProtocol protocol) noexcept
 
 	switch(protocol)
 	{
+	case MqttProtocol:
+		SendConnectCommand(portNumbers[protocol], protocol, ipAddresses[protocol]);
+		break;
 
 	default:
 		break;
@@ -476,6 +480,10 @@ void WiFiInterface::ShutdownProtocol(NetworkProtocol protocol) noexcept
 		TerminateSockets(portNumbers[protocol]);
 		break;
 
+	case MqttProtocol:
+		TerminateSockets(portNumbers[protocol], false);
+		break;
+
 	default:
 		break;
 	}
diff --git a/src/Networking/MqttClient.cpp b/src/Networking/MqttClient.cpp
new file mode 100644
index 0000000000..2bfe73d65a
--- /dev/null
+++ b/src/Networking/MqttClient.cpp
@@ -0,0 +1,276 @@
+/*
+ * MqttClient.cpp
+ */
+#include "MqttClient.h"
+
+#if SUPPORT_MQTT
+
+#include <Platform/Platform.h>
+#include <Networking/Network.h>
+
+#include "Socket.h"
+
+#include "mqtt.h"
+
+MqttClient::MqttClient(NetworkResponder *n, NetworkClient *c) noexcept : NetworkClient(n, c),
+																		sendBuf(nullptr),
+																		recvBuf(nullptr),
+																		client(nullptr),
+																		username(nullptr),
+																		password(nullptr),
+																		clientId(nullptr),
+																		publishTopic(nullptr),
+																		willClient(nullptr),
+																		willTopic(nullptr),
+																		willMessage(nullptr),
+																		willMessageSize(0),
+																		keepAlive(DefaultKeepAlive),
+																		qos(QOS::QOS0),
+																		retain(false),
+																		dup(false)
+{
+	client = new mqtt_client;
+	publishMutex.Create("MqttPublishBuffer");
+
+	sendBuf = new uint8_t[SendBufferSize];
+	recvBuf = new uint8_t[ReceiveBufferSize];
+
+	mqtt_init(client, skt, sendBuf, SendBufferSize, recvBuf, ReceiveBufferSize, PublishCallback);
+}
+
+bool MqttClient::Subscribe(const char *topic, QOS maxQos)
+{
+	enum MQTTErrors err = mqtt_subscribe(client, topic, static_cast<int>(maxQos));
+	return err == MQTT_OK;
+}
+
+bool MqttClient::Unsubscribe(const char *topic)
+{
+	enum MQTTErrors err = mqtt_unsubscribe(client, topic);
+	return err == MQTT_OK;
+}
+
+bool MqttClient::Publish(const char *topic, const void *message, size_t messageSize, QOS qos, bool retain, bool dup)
+{
+	uint8_t flags = 0;
+
+	switch (qos)
+	{
+	case QOS::QOS0:
+		flags |= MQTT_PUBLISH_QOS_0;
+		break;
+
+	case QOS::QOS1:
+		flags |= MQTT_PUBLISH_QOS_1;
+		break;
+
+	case QOS::QOS2:
+		flags |= MQTT_PUBLISH_QOS_2;
+		break;
+
+	default:
+		break;
+	}
+
+	flags |= (retain) ? MQTT_PUBLISH_RETAIN : 0;
+	flags |= (dup) ? MQTT_PUBLISH_DUP : 0;
+
+	enum MQTTErrors err = mqtt_publish(client, topic, message, messageSize, flags);
+	return err == MQTT_OK;
+}
+
+bool MqttClient::Spin() noexcept
+{
+	bool res = false;
+
+	if (skt && skt->CanSend())
+	{
+		switch (responderState)
+		{
+		case ResponderState::connecting:
+			{
+				mqtt_sync(client);
+
+				if (client->typical_response_time >= 0.0f) // ack was received
+				{
+					if (client->error == MQTT_OK)
+					{
+						responderState = ResponderState::connected;
+					}
+					else
+					{
+						// Re-attempt MQTT connection
+						Connect();
+					}
+
+					// Since at this stage we only expect ack (or some error), only work is done
+					// when it is actually received and processed.
+					res = true;
+				}
+			}
+			break;
+
+		case ResponderState::connected:
+		 	{
+				// Save the current receive buffer, so it can be checked if
+				// it moved forward (received something).
+				uint8_t *curr = client->recv_buffer.curr;
+
+				// Process the socket receive buffer
+				mqtt_sync(client);
+
+				bool ok = true, sent = false;
+				{
+					MutexLocker lock(publishMutex);
+
+					while (outBuf && ok)
+					{
+						ok = Publish(publishTopic ? publishTopic : reprap.GetNetwork().GetHostname(),
+										outBuf->Data(),outBuf->DataLength(), qos, retain, dup);
+						if (ok)
+						{
+							outBuf = OutputBuffer::Release(outBuf);
+							sent = true;
+						}
+					}
+				}
+
+				if (sent)
+				{
+					// Buffer the messages in the socket for the next loop
+					mqtt_sync(client);
+				}
+
+				// Did work if processed or sent something
+				res = (client->recv_buffer.curr > curr || sent);
+			}
+			break;
+
+		default:
+			break;
+		}
+	}
+	else
+	{
+		if (responderState == ResponderState::connecting || responderState == ResponderState::connected)
+		{
+			// Lost the connection
+			debugPrintf("connection lost\n");
+			ConnectionLost();
+		}
+	}
+
+	return res;
+}
+
+void MqttClient::Connect()
+{
+	mqtt_reinit(client, skt, sendBuf, SendBufferSize, recvBuf, ReceiveBufferSize);
+	mqtt_connect(client, clientId, willTopic, willMessage, willMessageSize, username, password, MQTT_CONNECT_CLEAN_SESSION, keepAlive);
+	// Buffer the CONNECT message, so the next mqtt_sync can process CONNACK if available.
+	mqtt_sync(client);
+	responderState = ResponderState::connecting;
+}
+
+bool MqttClient::Accept(Socket *s, NetworkProtocol protocol) noexcept
+{
+	if (responderState == ResponderState::free && HandlesProtocol(protocol) && s->GetInterface() == interface)
+	{
+		skt = s;
+		Connect();
+		return true;
+	}
+
+	return false;
+}
+
+void MqttClient::Disable() noexcept
+{
+	MutexLocker lock(MqttClient::Instance()->publishMutex);
+	OutputBuffer::ReleaseAll(MqttClient::Instance()->outBuf);
+}
+
+bool MqttClient::HandlesProtocol(NetworkProtocol protocol) noexcept
+{
+	return protocol == MqttProtocol;
+}
+
+void MqttClient::Terminate(NetworkProtocol protocol, NetworkInterface *interface) noexcept
+{
+	if (responderState != ResponderState::free && (HandlesProtocol(protocol) || protocol == AnyProtocol) && skt != nullptr && GetInterface() == interface)
+	{
+		ConnectionLost();
+		return;
+	}
+}
+
+void MqttClient::Diagnostics(MessageType mt) const noexcept
+{
+	GetPlatform().MessageF(mt, " MQTT(%d)", (int)responderState);
+}
+
+/*static*/ void MqttClient::QueueMessage(const char *msg) noexcept
+{
+	if (strlen(msg) < OUTPUT_BUFFER_SIZE - 1)
+	{
+		MutexLocker lock(MqttClient::Instance()->publishMutex);
+		OutputBuffer *buf = nullptr;
+		if (OutputBuffer::Allocate(buf))
+		{
+			buf->copy(msg);
+
+			if (MqttClient::Instance()->outBuf)
+			{
+				MqttClient::Instance()->outBuf->Append(buf);
+			}
+			else
+			{
+				MqttClient::Instance()->outBuf = buf;
+			}
+		}
+	};
+}
+
+/*static*/ void MqttClient::PublishCallback(void** state, struct mqtt_response_publish *published)
+{
+	debugPrintf("Received message from topic '%s':\n %s \n",
+				(const char*) published->topic_name, (const char*) published->application_message);
+}
+
+/*static*/ MqttClient *MqttClient::instance;
+
+/*
+ * MQTT-C PAL layer implementation
+ */
+
+ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags)
+{
+	Socket* skt = static_cast<Socket*>(fd);
+
+	size_t res = 0;
+
+	if (len)
+	{
+		res = skt->Send((const uint8_t*)buf, len);
+	}
+
+	return res;
+}
+
+ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags)
+{
+	Socket* skt = static_cast<Socket*>(fd);
+
+	size_t read = 0;
+	char c = 0;
+
+	while (read < bufsz && skt->ReadChar(c))
+	{
+		((char*) buf)[read] = c;
+		read++;
+	}
+
+	return read;
+}
+
+#endif
diff --git a/src/Networking/MqttClient.h b/src/Networking/MqttClient.h
new file mode 100644
index 0000000000..6622312e34
--- /dev/null
+++ b/src/Networking/MqttClient.h
@@ -0,0 +1,171 @@
+/*
+ * MqttClient.h
+ *
+ */
+
+#ifndef SRC_NETWORKING_MQTTCLIENT_H_
+#define SRC_NETWORKING_MQTTCLIENT_H_
+
+#include <sys/types.h>
+#include <stdint.h>
+#include <string.h>
+#include <limits.h>
+#include <stdarg.h>
+
+#ifdef __cplusplus
+
+#include "NetworkClient.h"
+
+struct mqtt_client;
+
+class MqttClient : public NetworkClient
+{
+public:
+	enum class QOS : uint8_t
+	{
+		QOS0,
+		QOS1,
+		QOS2
+	};
+
+	static MqttClient* Instance(NetworkResponder *r, NetworkClient *c)
+	{
+		if (!instance)
+		{
+			instance = new MqttClient(r, c);
+		}
+
+		return instance;
+	}
+
+	static MqttClient* Instance()
+	{
+		return instance;
+	}
+
+	bool Spin() noexcept override;															// do some work, returning true if we did anything significant
+	bool Accept(Socket *s, NetworkProtocol protocol) noexcept override;						// ask the responder to accept this connection, returns true if it did
+	void Terminate(NetworkProtocol protocol, NetworkInterface *interface) noexcept override;	// terminate the responder if it is serving the specified protocol on the specified interface
+	void Diagnostics(MessageType mtype) const noexcept override;
+
+	bool Subscribe(const char *topic, QOS maxQos);
+	bool Unsubscribe(const char *topic);
+
+	static void Disable() noexcept;
+	static void QueueMessage(const char *msg) noexcept;
+
+	bool HandlesProtocol(NetworkProtocol p) noexcept override;
+
+private:
+	MqttClient(NetworkResponder *n, NetworkClient *c) noexcept;
+
+	void Connect();
+	bool Publish(const char *topic, const void *message, size_t messageSize, QOS qos, bool retain, bool dup);
+
+	static void PublishCallback(void** state, struct mqtt_response_publish *published);
+
+	static const int SendBufferSize = 2048;
+	static const int ReceiveBufferSize = 1024;
+	static const int DefaultKeepAlive = 400;
+
+	uint8_t *sendBuf;
+	uint8_t *recvBuf;
+
+	mqtt_client* client;
+
+	const char* username;
+	const char* password;
+
+	const char* clientId;
+	const char* publishTopic;			// Topic this MQTT client publishes to, defaults to hostname if not specified
+	const char* willClient;
+	const char* willTopic;
+	const char* willMessage;
+
+	uint16_t willMessageSize;
+	uint16_t keepAlive;
+
+	QOS qos;
+
+	bool retain;
+	bool dup;
+
+	Mutex publishMutex;
+
+	static MqttClient *instance;
+};
+
+#endif
+
+/*
+ * MQTT-C PAL layer declarations and definitions
+ */
+
+// Network socket - pointer to Socket object, given to MqttClient on acceptance.
+typedef void* mqtt_pal_socket_handle;
+
+// Timekeeping - must give time in seconds. Define time units in terms of time_t,
+// even though it is not referenced to epoch.
+typedef time_t mqtt_pal_time_t;
+#define MQTT_PAL_TIME() (millis() / 1000)
+
+// Mutex - define to nothing; MQTT-C is used in a single thread context.
+typedef int mqtt_pal_mutex_t;
+#define MQTT_PAL_MUTEX_LOCK(x)
+#define MQTT_PAL_MUTEX_INIT(x)
+#define MQTT_PAL_MUTEX_UNLOCK(x)
+
+// Byte order functions
+#define MQTT_PAL_HTONS(s) __builtin_bswap16(s)
+#define MQTT_PAL_NTOHS(s) __builtin_bswap16(s)
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// Socket read/write
+/**
+ * @brief Sends all the bytes in a buffer.
+ * @ingroup pal
+ *
+ * @param[in] fd The file-descriptor (or handle) of the socket.
+ * @param[in] buf A pointer to the first byte in the buffer to send.
+ * @param[in] len The number of bytes to send (starting at \p buf).
+ * @param[in] flags Flags which are passed to the underlying socket.
+ *
+ * @returns The number of bytes sent if successful, an \ref MQTTErrors otherwise.
+ *
+ * Note about the error handling:
+ * - On an error, if some bytes have been processed already,
+ *   this function should return the number of bytes successfully
+ *   processed. (partial success)
+ * - Otherwise, if the error is an equivalent of EAGAIN, return 0.
+ * - Otherwise, return MQTT_ERROR_SOCKET_ERROR.
+ */
+ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags);
+
+/**
+ * @brief Non-blocking receive all the byte available.
+ * @ingroup pal
+ *
+ * @param[in] fd The file-descriptor (or handle) of the socket.
+ * @param[in] buf A pointer to the receive buffer.
+ * @param[in] bufsz The max number of bytes that can be put into \p buf.
+ * @param[in] flags Flags which are passed to the underlying socket.
+ *
+ * @returns The number of bytes received if successful, an \ref MQTTErrors otherwise.
+ *
+ * Note about the error handling:
+ * - On an error, if some bytes have been processed already,
+ *   this function should return the number of bytes successfully
+ *   processed. (partial success)
+ * - Otherwise, if the error is an equivalent of EAGAIN, return 0.
+ * - Otherwise, return MQTT_ERROR_SOCKET_ERROR.
+ */
+ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* SRC_NETWORKING_MQTTCLIENT_H_ */
diff --git a/src/Networking/Network.cpp b/src/Networking/Network.cpp
index 68307b8eab..ecd31c64e2 100644
--- a/src/Networking/Network.cpp
+++ b/src/Networking/Network.cpp
@@ -37,6 +37,8 @@
 # include "RTOSPlusTCPEthernet/RTOSPlusTCPEthernetInterface.h"
 #endif
 
+#include "MqttClient.h"
+
 #if SUPPORT_HTTP
 # include "HttpResponder.h"
 #endif
@@ -46,6 +48,9 @@
 #if SUPPORT_TELNET
 # include "TelnetResponder.h"
 #endif
+#if SUPPORT_MQTT
+#include "MqttClient.h"
+#endif
 #if SUPPORT_MULTICAST_DISCOVERY
 # include "MulticastDiscovery/MulticastResponder.h"
 #endif
@@ -269,6 +274,12 @@ GCodeResult Network::DisableProtocol(unsigned int interface, NetworkProtocol pro
 				break;
 #endif
 
+#if SUPPORT_MQTT
+			case MqttProtocol:
+				MqttClient::Disable();
+				break;
+#endif
+
 #if SUPPORT_MULTICAST_DISCOVERY
 			case MulticastDiscoveryProtocol:
 				break;
@@ -335,6 +346,9 @@ GCodeResult Network::EnableInterface(unsigned int interface, int mode, const Str
 #if SUPPORT_TELNET
 			TelnetResponder::Disable();
 #endif
+#if SUPPORT_MQTT
+			MqttClient::Disable();
+#endif
 #endif // HAS_RESPONDERS
 		}
 		return ret;
@@ -463,6 +477,10 @@ void Network::Activate() noexcept
 	MulticastResponder::Init();
 #endif
 
+#if SUPPORT_MQTT
+	responders = clients = MqttClient::Instance(responders, clients);
+#endif
+
 	// Finally, create the network task
 	networkTask.Create(NetworkLoop, "NETWORK", nullptr, TaskPriority::SpinPriority);
 #endif
@@ -488,6 +506,9 @@ void Network::Exit() noexcept
 #if SUPPORT_TELNET
 	TelnetResponder::Disable();
 #endif
+#if SUPPORT_MQTT
+	MqttClient::Disable();
+#endif
 
 	if (TaskBase::GetCallerTaskHandle() != &networkTask)
 	{
@@ -783,6 +804,13 @@ void Network::HandleTelnetGCodeReply(const char *msg) noexcept
 #endif
 }
 
+#if SUPPORT_MQTT
+void Network::MqttPublish(const char *msg) noexcept
+{
+	MqttClient::Instance()->QueueMessage(msg);
+}
+#endif
+
 void Network::HandleHttpGCodeReply(OutputBuffer *buf) noexcept
 {
 #if SUPPORT_HTTP
diff --git a/src/Networking/Network.h b/src/Networking/Network.h
index b7c52f85f7..9556e3cf84 100644
--- a/src/Networking/Network.h
+++ b/src/Networking/Network.h
@@ -22,6 +22,8 @@ const size_t MaxNetworkInterfaces = 1;
 # error Wrong Network.h file included
 #endif
 
+const size_t NumMqttClients = 1;
+
 #if defined(__LPC17xx__)
 // Only 2 http responders as we are tight on memory.
 const size_t NumHttpResponders = 2;		// the number of concurrent HTTP requests we can process
@@ -122,6 +124,11 @@ class Network INHERIT_OBJECT_MODEL
 	void HandleTelnetGCodeReply(OutputBuffer *buf) noexcept;
 
 	NetworkClient *GetClients() noexcept { return clients; }
+
+#if SUPPORT_MQTT
+	void MqttPublish(const char *msg) noexcept;
+#endif
+
 	uint32_t GetHttpReplySeq() noexcept;
 
 protected:
diff --git a/src/Networking/NetworkClient.cpp b/src/Networking/NetworkClient.cpp
index 013fe74977..9c140b50cd 100644
--- a/src/Networking/NetworkClient.cpp
+++ b/src/Networking/NetworkClient.cpp
@@ -9,17 +9,21 @@
 #include "Socket.h"
 #include <Platform/Platform.h>
 
-static const uint32_t connecTimeout = ConnectTimeout + 1000;
+static const uint32_t connectTimeout = ConnectTimeout + 1000; // Make sure the socket, if any, is closed before attempting another connection.
 
 NetworkClient::NetworkClient(NetworkResponder *n, NetworkClient *c) noexcept : NetworkResponder(n), interface(nullptr), next(c), whenRequest(0)
 {
 }
 
+bool NetworkClient::RequestConnection() noexcept
+{
+	return true;
+}
+
 bool NetworkClient::RequestConnection(NetworkProtocol protocol, NetworkInterface *interface) noexcept
 {
 	// Provides an implementation that automatically requests for reconnection all the time.
-	// However, children classes might override this behavior if they need a more ephemeral
-	// connection; for example, only requesting a connection when there is data to be sent.
+	// However, child classes might override this behavior if they need custom logic.
 	if (HandlesProtocol(protocol))
 	{
 		if (!skt)
@@ -57,7 +61,7 @@ bool NetworkClient::RequestConnection(NetworkProtocol protocol, NetworkInterface
 				return true;
 			}
 		}
-	} 
+	}
 
 	// Client does not handle the protocol, or has a socket already; do not request a connection.
 	return false;
diff --git a/src/Networking/NetworkClient.h b/src/Networking/NetworkClient.h
index 8c8d379d53..7b93462639 100644
--- a/src/Networking/NetworkClient.h
+++ b/src/Networking/NetworkClient.h
@@ -29,10 +29,6 @@ class NetworkClient : public NetworkResponder
 	bool RequestConnection(NetworkProtocol protocol, NetworkInterface *interface) noexcept;
 
 	virtual bool HandlesProtocol(NetworkProtocol p) noexcept = 0;
-private:
-
-	NetworkClient *next;
-	uint32_t connectTime;
 
 protected:
 	NetworkInterface *interface; 	// interface this client is associated with
diff --git a/src/Networking/NetworkDefs.h b/src/Networking/NetworkDefs.h
index d1f17b9eb9..a033999b53 100644
--- a/src/Networking/NetworkDefs.h
+++ b/src/Networking/NetworkDefs.h
@@ -36,7 +36,7 @@ constexpr IPAddress DefaultIpAddress;				// will be initialised to 0 by construc
 constexpr IPAddress DefaultNetMask(0x00FFFFFF);		// equivalent to 255.255.255.0. Use constexpr constructor to avoid it being allocated in RAM.
 constexpr IPAddress DefaultGateway;					// will be initialised to 0 by constructor
 
-constexpr size_t NumTcpProtocols = 3;
+constexpr size_t NumTcpProtocols = 4;
 
 #if SUPPORT_MULTICAST_DISCOVERY
 constexpr size_t NumProtocols = NumTcpProtocols + 1;	// number of network protocols we support, not counting FtpDataProtocol, MdnsProtocol or AnyProtocol
@@ -44,11 +44,12 @@ constexpr size_t NumProtocols = NumTcpProtocols + 1;	// number of network protoc
 constexpr size_t NumProtocols = NumTcpProtocols;		// number of network protocols we support, not counting FtpDataProtocol, MdnsProtocol or AnyProtocol
 #endif
 
-constexpr NetworkProtocol HttpProtocol = 0, FtpProtocol = 1, TelnetProtocol = 2, MulticastDiscoveryProtocol = 3, FtpDataProtocol = 3, MdnsProtocol = 4, AnyProtocol = 255;
+constexpr NetworkProtocol HttpProtocol = 0, FtpProtocol = 1, TelnetProtocol = 2, MulticastDiscoveryProtocol = 3, FtpDataProtocol = 3, MqttProtocol = 3, MdnsProtocol = 4, AnyProtocol = 255;
 
 constexpr TcpPort DefaultHttpPort = 80;
 constexpr TcpPort DefaultFtpPort = 21;
 constexpr TcpPort DefaultTelnetPort = 23;
+constexpr TcpPort DefaultMqttPort = 1883;
 #if SUPPORT_MULTICAST_DISCOVERY
 constexpr TcpPort DefaultMulticastDiscoveryPort = 10002;	// this is actually a UDP port
 #endif
@@ -57,15 +58,17 @@ constexpr TcpPort DefaultPortNumbers[NumProtocols] =
 {
 	DefaultHttpPort, DefaultFtpPort, DefaultTelnetPort,
 #if SUPPORT_MULTICAST_DISCOVERY
-	DefaultMulticastDiscoveryPort
+	DefaultMulticastDiscoveryPort,
 #endif
+	DefaultMqttPort,
 };
 constexpr const char *_ecv_array ProtocolNames[NumProtocols] =
 {
 	"HTTP", "FTP", "TELNET",
 #if SUPPORT_MULTICAST_DISCOVERY
-	"Multicast Discovery"
+	"Multicast Discovery",
 #endif
+	"MQTT"
 };
 
 constexpr uint8_t MdnsMacAddress[6] = { 0x01, 0x00, 0x5E, 0x00, 0x00, 0xFB };
diff --git a/src/Networking/NetworkResponder.h b/src/Networking/NetworkResponder.h
index 2cfb3c9779..7420e4e877 100644
--- a/src/Networking/NetworkResponder.h
+++ b/src/Networking/NetworkResponder.h
@@ -53,7 +53,10 @@ class NetworkResponder
 
 		// Telnet responder additional states
 		justConnected,
-		authenticating
+		authenticating,
+
+		connecting,
+		connected,
 	};
 
 	NetworkResponder(NetworkResponder *n) noexcept;
diff --git a/src/Platform/MessageType.h b/src/Platform/MessageType.h
index d692329483..4382f43ff3 100644
--- a/src/Platform/MessageType.h
+++ b/src/Platform/MessageType.h
@@ -27,6 +27,7 @@ enum MessageType : uint32_t
 	DaemonMessage =				 0x200u,	// A message that is sent to the daemon processor
 	Aux2Message =				 0x400u,	// A message that is to be sent to the second aux device
 	AutoPauseMessage =			 0x800u,	// A message that is to be sent to an auto-pause processor
+	MqttMessage = 				0x1000u,	// A message that is to be sent to a Mqtt processor
 
 	// Special destinations (byte 3)
 	BlockingUsbMessage =	   0x10000u,	// A message that is to be sent to USB in blocking mode
diff --git a/src/Platform/Platform.cpp b/src/Platform/Platform.cpp
index 7e003bf6ce..2ceeedadfa 100644
--- a/src/Platform/Platform.cpp
+++ b/src/Platform/Platform.cpp
@@ -3286,6 +3286,11 @@ void Platform::RawMessage(MessageType type, const char *_ecv_array message) noex
 		reprap.GetNetwork().HandleTelnetGCodeReply(message);
 	}
 
+	if ((type & MqttMessage) != 0)
+	{
+		reprap.GetNetwork().MqttPublish(message);
+	}
+
 	if ((type & Aux2Message) != 0)
 	{
 		AppendAuxReply(1, message, message[0] == '{' || (type & RawMessageFlag) != 0);