fixup! Added Zephyr implementation for Mqtt client

Author: andreagilardoni

src/implementations/ZephyrMqttClient.cpp

@@ -41,7 +41,8 @@ static inline void __mqtt_utf8_allocation_helper(
 	}
 }
 
-ZephyrMqttClient::ZephyrMqttClient() {
+ZephyrMqttClient::ZephyrMqttClient()
+: last_message(nullptr) {
 	fds.fd = -1;
     mqtt_client_init(&client);
 }
@@ -198,21 +199,6 @@ error_t ZephyrMqttClient::publish(Topic t, uint8_t payload[], size_t size, MqttQ
 	return mqtt_publish(&client, &param); // error codes should be proxied
 }
 
-class ZephyrMqttOStream: public MqttOStream {
-public:
-	ZephyrMqttOStream(): MqttOStream(NotImplementedError) {}
-
-    size_t write(uint8_t) override { return 0; };
-    size_t write(const uint8_t *buffer, size_t size) override { return 0; };
-    int availableForWrite() override { return 0; };
-};
-
-MqttOStream&& ZephyrMqttClient::publish(
-        Topic t, MqttQos qos,
-        MqttPublishFlag flags) {
-	return std::move(ZephyrMqttOStream()); // TODO is this correct?
-}
-
 error_t ZephyrMqttClient::unsubscribe(Topic t) {
 	struct mqtt_topic topic;
 	struct mqtt_subscription_list unsub;
@@ -273,56 +259,43 @@ error_t ZephyrMqttClient::ping() {
 	return mqtt_ping(&client);
 }
 
-// this class should extend something like IStream, that doesn't exist in arduino apis
-class ZephyrMqttReadStream: public IStream {
-public:
-	ZephyrMqttReadStream(struct mqtt_client *client, const struct mqtt_publish_param *p)
-	: client(client), p(p) {}
-
-    ~ZephyrMqttReadStream() { end(); }
-
-    size_t readBytes(uint8_t* buf, size_t s) override {
-		return mqtt_read_publish_payload(client, buf, s);
-	}
-
-    int available() override {
-		return client->internal.remaining_payload;
-	}
+int ZephyrMqttClient::read() {
+	uint8_t a;
+	int res = read(&a, 1);
+	return res <= 0 ? res : a;
+}
 
-    int read() override {
-		uint8_t r;
-		int res = mqtt_read_publish_payload_blocking(client, &r, 1);
+int ZephyrMqttClient::read(uint8_t payload[], size_t len) {
+	int res = mqtt_read_publish_payload_blocking(&client, payload, len);
 
-		return res==1 ? r : res;
+	if(res < 0) {
+		return res;
 	}
 
-    void end() {
-		if(p == nullptr) {
-			return;
-		}
-		// consume all remaining payload
-		mqtt_readall_publish_payload(client, nullptr, available());
-
-		// acknowledgements should be sent when the entire packet is consumed, the maximum size for an mqtt
-		// message could be 4MB, so this should be treated like a stream.
-		if (p->message.topic.qos == MQTT_QOS_1_AT_LEAST_ONCE) {
+	if(available() == 0) {
+		if (messageQoS() == MqttQos1) {
 			const struct mqtt_puback_param ack_param = {
-				.message_id = p->message_id
+				.message_id = messageId()
 			};
-			mqtt_publish_qos1_ack(client, &ack_param);
-		} else if (p->message.topic.qos == MQTT_QOS_2_EXACTLY_ONCE) {
+			mqtt_publish_qos1_ack(&client, &ack_param);
+		} else if (messageQoS() == MqttQos2) {
 			const struct mqtt_pubrec_param rec_param = {
-				.message_id = p->message_id
+				.message_id = messageId()
 			};
-			mqtt_publish_qos2_receive(client, &rec_param);
+
+			mqtt_publish_qos2_receive(&client, &rec_param);
 		}
 
-		p = nullptr;
+		delete last_message;
+		last_message = nullptr;
 	}
-private:
-	struct mqtt_client *client;
-	const struct mqtt_publish_param *p;
-};
+
+	return res;
+}
+
+int ZephyrMqttClient::available() {
+	return client.internal.remaining_payload;
+}
 
 void ZephyrMqttClient::mqtt_event_handler(
     struct mqtt_client *const client, const struct mqtt_evt *evt) {
@@ -354,16 +327,18 @@ void ZephyrMqttClient::mqtt_event_handler(
 	}
 	case MQTT_EVT_PUBLISH: {
 		const struct mqtt_publish_param *p = &evt->param.publish;
-		ZephyrMqttReadStream stream(&this->client, p);
+		if(last_message == nullptr) {
+			last_message = new mqtt_publish_param;
+		}
+
+		memcpy(last_message, p, sizeof(*last_message));
 
 		// Pass the stream to the user callback
 		if(_cbk != nullptr) {
 			const char* topic = (const char*)p->message.topic.topic.utf8; // will this be null terminated string?
 
-			_cbk(topic, stream);
+			_cbk(topic);
 		}
-
-		stream.end();
 	}
 	default:
 		break;
@@ -396,5 +371,45 @@ void ZephyrMqttClient::setReceiveCallback(MqttReceiveCallback cbk) {
     this->_cbk = cbk;
 }
 
+String ZephyrMqttClient::messageTopic() const {
+	if(last_message != nullptr) {
+		return (const char*)last_message->message.topic.topic.utf8;
+	}
+
+	return "";
+}
+
+int ZephyrMqttClient::messageDup() const {
+	if(last_message != nullptr) {
+		return last_message->dup_flag;
+	}
+
+	return -1;
+}
+
+uint16_t ZephyrMqttClient::messageId() const {
+	if(last_message != nullptr) {
+		return last_message->message_id;
+	}
+
+	return 0;
+}
+
+MqttQos ZephyrMqttClient::messageQoS() const {
+	if(last_message != nullptr) {
+		return static_cast<MqttQos>(last_message->message.topic.qos);
+	}
+
+	return QosDefault;
+}
+
+int ZephyrMqttClient::messageRetain() const {
+	if(last_message != nullptr) {
+		return last_message->retain_flag;
+	}
+
+	return -1;
+}
+
 
 #endif // CONFIG_MQTT_LIB

src/implementations/ZephyrMqttClient.h

@@ -28,11 +28,6 @@ class ZephyrMqttClient: public MqttClientInterface {
         size_t size, MqttQos qos = QosDefault,
         MqttPublishFlag flags = MqttPublishFlags::None) override;
 
-    // TODO zephyr mqtt publish definition doesn't support streaming a message
-    MqttOStream&& publish(
-        Topic t, MqttQos qos = QosDefault,
-        MqttPublishFlag flags = MqttPublishFlags::None) override;
-
     error_t unsubscribe(Topic t) override;
     void poll() override;
     error_t ping() override;
@@ -51,6 +46,15 @@ class ZephyrMqttClient: public MqttClientInterface {
     // in zephyr we do not need to provide a client on which to rely on
     void setClient(arduino::Client*) override {};
 
+    int read() override;
+    int read(uint8_t payload[], size_t len) override;
+    int available() override;
+
+    String messageTopic() const override;
+    int messageDup()      const override;
+    uint16_t messageId()  const override;
+    MqttQos messageQoS()  const override;
+    int messageRetain()   const override;
 private:
     int connect(sockaddr_storage addr);
     void client_init();
@@ -63,4 +67,6 @@ class ZephyrMqttClient: public MqttClientInterface {
     uint8_t tx_buffer[CONFIG_MQTT_CLIENT_TX_BUFFER];
 
     MqttReceiveCallback _cbk;
+
+    struct mqtt_publish_param *last_message;
 };