From 9670fdbdf7aa4a406194231d6260d89399dfab7b Mon Sep 17 00:00:00 2001
From: proddy <paul@derbyshire.nl>
Date: Fri, 28 Aug 2020 10:00:05 +0200
Subject: [PATCH] added back Rx queue - (v2) occasional Rx incomplete telegrams
 #460

---
 src/emsesp.cpp   | 17 +++++++++++++++--
 src/telegram.cpp | 21 +++++++++++++++------
 src/telegram.h   | 26 +++++++++++++++++++++++---
 3 files changed, 53 insertions(+), 11 deletions(-)

diff --git a/src/emsesp.cpp b/src/emsesp.cpp
index 5bad769fd..337b49858 100644
--- a/src/emsesp.cpp
+++ b/src/emsesp.cpp
@@ -222,6 +222,19 @@ void EMSESP::show_ems(uuid::console::Shell & shell) {
 
     shell.println();
 
+    // Rx queue
+    auto rx_telegrams = rxservice_.queue();
+    if (rx_telegrams.empty()) {
+        shell.printfln(F("Rx Queue is empty"));
+    } else {
+        shell.printfln(F("Rx Queue (%ld telegram%s):"), rx_telegrams.size(), rx_telegrams.size() == 1 ? "" : "s");
+        for (const auto & it : rx_telegrams) {
+            shell.printfln(F(" [%02d] %s"), it.id_, pretty_telegram(it.telegram_).c_str());
+        }
+    }
+
+    shell.println();
+
     // Tx queue
     auto tx_telegrams = txservice_.queue();
     if (tx_telegrams.empty()) {
@@ -270,7 +283,7 @@ void EMSESP::show_sensor_values(uuid::console::Shell & shell) {
     }
 
     char valuestr[8] = {0}; // for formatting temp
-    shell.printfln(F("External temperature sensors:"));
+    shell.printfln(F("Dallas temperature sensors:"));
     for (const auto & device : sensor_devices()) {
         shell.printfln(F("  ID: %s, Temperature: %s°C"), device.to_string().c_str(), Helpers::render_value(valuestr, device.temperature_c, 2));
     }
@@ -774,7 +787,7 @@ void EMSESP::send_raw_telegram(const char * data) {
 // the services must be loaded in the correct order
 void EMSESP::start() {
     // Load our library of known devices. Names are stored in Flash mem.
-    device_library_.reserve(100);
+    device_library_.reserve(80);
     device_library_ = {
 #include "device_library.h"
     };
diff --git a/src/telegram.cpp b/src/telegram.cpp
index 2bff2698d..5b22c26ed 100644
--- a/src/telegram.cpp
+++ b/src/telegram.cpp
@@ -129,10 +129,11 @@ std::string Telegram::to_string_message() const {
 
 // checks if we have an Rx telegram that needs processing
 void RxService::loop() {
-    if (rx_telegram) {
-        EMSESP::process_telegram(rx_telegram);
-        rx_telegram = nullptr;      // telegram has been processed, reset
-        increment_telegram_count(); // increase rx count
+    while (!rx_telegrams_.empty()) {
+        auto telegram = rx_telegrams_.front().telegram_;
+        (void)EMSESP::process_telegram(telegram); // further process the telegram
+        increment_telegram_count();               // increase rx count
+        rx_telegrams_.pop_front();                // remove it from the queue
     }
 }
 
@@ -212,8 +213,16 @@ void RxService::add(uint8_t * data, uint8_t length) {
     // if we receive a hc2.. telegram from 0x19.. match it to master_thermostat if master is 0x18
     src = EMSESP::check_master_device(src, type_id, true);
 
-    // create the telegram
-    rx_telegram = std::make_shared<Telegram>(Telegram::Operation::RX, src, dest, type_id, offset, message_data, message_length);
+        // create the telegram
+    auto telegram = std::make_shared<Telegram>(Telegram::Operation::RX, src, dest, type_id, offset, message_data, message_length);
+
+    // check if queue is full, if so remove top item to make space
+    if (rx_telegrams_.size() >= MAX_RX_TELEGRAMS) {
+        rx_telegrams_.pop_front();
+    }
+
+    rx_telegrams_.emplace_back(rx_telegram_id_++, std::move(telegram)); // add to queue
+
 }
 
 //
diff --git a/src/telegram.h b/src/telegram.h
index c0c900446..1eca7bd82 100644
--- a/src/telegram.h
+++ b/src/telegram.h
@@ -192,12 +192,13 @@ class EMSbus {
 };
 
 class RxService : public EMSbus {
+    static constexpr size_t MAX_RX_TELEGRAMS = 10;
+
   public:
     RxService()  = default;
     ~RxService() = default;
 
     void loop();
-
     void add(uint8_t * data, uint8_t length);
 
     uint16_t telegram_count() const {
@@ -216,10 +217,29 @@ class RxService : public EMSbus {
         telegram_error_count_++;
     }
 
+    class QueuedRxTelegram {
+      public:
+        const uint16_t                        id_;
+        const std::shared_ptr<const Telegram> telegram_;
+
+        ~QueuedRxTelegram() = default;
+        QueuedRxTelegram(uint16_t id, std::shared_ptr<Telegram> && telegram)
+            : id_(id)
+            , telegram_(std::move(telegram)) {
+        }
+    };
+
+    const std::list<QueuedRxTelegram> queue() const {
+        return rx_telegrams_;
+    }
+
   private:
+    uint8_t                         rx_telegram_id_       = 0; // queue counter
     uint16_t                        telegram_count_       = 0; // # Rx received
     uint16_t                        telegram_error_count_ = 0; // # Rx CRC errors
     std::shared_ptr<const Telegram> rx_telegram;               // the incoming Rx telegram
+
+    std::list<QueuedRxTelegram> rx_telegrams_; // the Rx Queue
 };
 
 class TxService : public EMSbus {
@@ -341,9 +361,9 @@ class TxService : public EMSbus {
 
     std::shared_ptr<Telegram> telegram_last_;
     uint16_t                  telegram_last_post_send_query_; // which type ID to query after a successful send, to read back the values just written
-    uint8_t                   retry_count_    = 0;            // count for # Tx retries
+    uint8_t                   retry_count_ = 0;               // count for # Tx retries
 
-    uint8_t                   tx_telegram_id_ = 0;            // queue counter
+    uint8_t tx_telegram_id_ = 0; // queue counter
 
     void send_telegram(const QueuedTxTelegram & tx_telegram);
     void send_telegram(const uint8_t * data, const uint8_t length);