fix(modem): Support for TCP-TRANSPORT mode

Author: david-cermak

components/esp_modem/examples/modem_tcp_client/main/command/sock_dce.cpp

@@ -6,7 +6,6 @@
 
 #include <charconv>
 #include <sys/socket.h>
-#include <algorithm>  // for std::find
 #include "esp_vfs.h"
 #include "esp_vfs_eventfd.h"
 
@@ -17,8 +16,10 @@ namespace sock_dce {
 constexpr auto const *TAG = "sock_dce";
 
 // Definition of the static member variables
-std::vector<DCE*> DCE::dce_list{};
+std::vector<DCE *> DCE::dce_list{};
 bool DCE::network_init = false;
+int Responder::s_link_id = 0;
+SemaphoreHandle_t Responder::s_dte_mutex{};
 
 // Constructor - add this DCE instance to the static list
 DCE::DCE(std::shared_ptr<esp_modem::DTE> dte_arg, const esp_modem_dce_config *config)
@@ -36,6 +37,7 @@ DCE::~DCE()
     }
 }
 
+
 bool DCE::perform_sock()
 {
     if (listen_sock == -1) {
@@ -81,13 +83,22 @@ bool DCE::perform_sock()
 
 void DCE::perform_at(uint8_t *data, size_t len)
 {
-    ESP_LOG_BUFFER_HEXDUMP(TAG, data, len, ESP_LOG_VERBOSE);
+    std::string_view resp_sv((char *)data, len);
+    at.check_urc(state, resp_sv);
+    if (state == status::IDLE) {
+        return;
+    }
+    ESP_LOG_BUFFER_HEXDUMP(TAG, data, len, ESP_LOG_INFO);
     switch (at.process_data(state, data, len)) {
     case Responder::ret::OK:
+        ESP_LOGW(TAG, "GIVE data %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::IDLE;
         signal.set(IDLE);
         return;
     case Responder::ret::FAIL:
+        ESP_LOGW(TAG, "GIVE data %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::FAILED;
         signal.set(IDLE);
         return;
@@ -102,10 +113,14 @@ void DCE::perform_at(uint8_t *data, size_t len)
     std::string_view response((char *)data, len);
     switch (at.check_async_replies(state, response)) {
     case Responder::ret::OK:
+        ESP_LOGW(TAG, "GIVE command %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::IDLE;
         signal.set(IDLE);
         return;
     case Responder::ret::FAIL:
+        ESP_LOGW(TAG, "GIVE command %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::FAILED;
         signal.set(IDLE);
         return;
@@ -124,7 +139,6 @@ void DCE::close_sock()
         close(sock);
         sock = -1;
     }
-    close(data_ready_fd);
     dte->on_read(nullptr);
     const int retries = 5;
     int i = 0;
@@ -152,6 +166,9 @@ bool DCE::at_to_sock()
         close_sock();
         return false;
     }
+    ESP_LOGI(TAG, "TAKE RECV %d", at.link_id);
+    xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+    ESP_LOGE(TAG, "TAKE RECV %d", at.link_id);
     state = status::RECEIVING;
     at.start_receiving(at.get_buf_len());
     return true;
@@ -160,8 +177,8 @@ bool DCE::at_to_sock()
 bool DCE::sock_to_at()
 {
     ESP_LOGD(TAG,  "socket read: data available");
-    if (!signal.wait(IDLE, 1000)) {
-        ESP_LOGE(TAG,  "Failed to get idle");
+    if (!signal.wait(IDLE, 5000)) {
+        ESP_LOGE(TAG,  "Failed to get idle 2");
         close_sock();
         return false;
     }
@@ -170,6 +187,9 @@ bool DCE::sock_to_at()
         close_sock();
         return false;
     }
+    ESP_LOGI(TAG, "TAKE SEND %d", at.link_id);
+    xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+    ESP_LOGE(TAG, "TAKE SEND %d", at.link_id);
     state = status::SENDING;
     int len = ::recv(sock, at.get_buf(), at.get_buf_len(), 0);
     if (len < 0) {
@@ -247,16 +267,18 @@ bool DCE::connect(std::string host, int port)
 {
     data_ready_fd = eventfd(0, EFD_SUPPORT_ISR);
     assert(data_ready_fd > 0);
-    dte->on_read(nullptr);
-    tcp_close();
-    dte->on_read([this](uint8_t *data, size_t len) {
-        read_callback(data, len);
-        return esp_modem::command_result::TIMEOUT;
-    });
+    // dte->on_read(nullptr);
+    // tcp_close();
+    // dte->on_read([](uint8_t *data, size_t len) {
+    //     read_callback(data, len);
+    //     return esp_modem::command_result::TIMEOUT;
+    // });
+    ESP_LOGI(TAG, "TAKE CONNECT %d", at.link_id);
+    xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+    ESP_LOGE(TAG, "TAKE CONNECT %d", at.link_id);
     if (!at.start_connecting(host, port)) {
         ESP_LOGE(TAG, "Unable to start connecting");
         dte->on_read(nullptr);
-        close(data_ready_fd);
         return false;
     }
     state = status::CONNECTING;
@@ -269,6 +291,8 @@ bool DCE::init()
         return true;
     }
     network_init = true;
+    Responder::s_dte_mutex = xSemaphoreCreateBinary();
+    xSemaphoreGive(at.s_dte_mutex);
     esp_vfs_eventfd_config_t config = ESP_VFS_EVENTD_CONFIG_DEFAULT();
     esp_vfs_eventfd_register(&config);
 
@@ -312,6 +336,10 @@ bool DCE::init()
         esp_modem::Task::Delay(5000);
     }
     ESP_LOGI(TAG, "Got IP %s", ip_addr.c_str());
+    dte->on_read([](uint8_t *data, size_t len) {
+        read_callback(data, len);
+        return esp_modem::command_result::TIMEOUT;
+    });
     return true;
 }
 

components/esp_modem/examples/modem_tcp_client/main/command/sock_dce.hpp

@@ -34,6 +34,7 @@ class Responder {
         sock(s), data_ready_fd(ready_fd), dte(dte_arg) {}
     ret process_data(status state, uint8_t *data, size_t len);
     ret check_async_replies(status state, std::string_view &response);
+    ret check_urc(status state, std::string_view &response);
 
     void start_sending(size_t len);
     void start_receiving(size_t len);
@@ -63,13 +64,17 @@ class Responder {
         return total_len;
     }
 
+    int link_id{s_link_id++};
+    static SemaphoreHandle_t s_dte_mutex;
 private:
+    static int s_link_id;
     static constexpr size_t buffer_size = 512;
 
     bool on_read(char *data, size_t len)
     {
 #ifndef CONFIG_EXAMPLE_CUSTOM_TCP_TRANSPORT
         ::send(sock, data, len, 0);
+        printf("sending %d\n", len);
 #else
         ::memcpy(&buffer[actual_read], data, len);
         actual_read += len;
@@ -100,11 +105,6 @@ class Responder {
 
 class DCE : public Module {
 public:
-    // Constructor and destructor for managing dce_list
-    // explicit GenericModule(std::shared_ptr<DTE> dte, std::unique_ptr<PdpContext> pdp):
-    //     dte(std::move(dte)), pdp(std::move(pdp)) {}
-    // explicit GenericModule(std::shared_ptr<DTE> dte, );
-    //
     DCE(std::shared_ptr<esp_modem::DTE> dte_arg, const esp_modem_dce_config *config);
     ~DCE();
 
@@ -169,6 +169,9 @@ class DCE : public Module {
             return 0;
         }
         at.clear_offsets();
+        ESP_LOGI("TAG", "TAKE RECV %d", at.link_id);
+        xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+        ESP_LOGE("TAG", "TAKE RECV %d", at.link_id);
         state = status::RECEIVING;
         uint64_t data;
         read(data_ready_fd, &data, sizeof(data));
@@ -190,6 +193,9 @@ class DCE : public Module {
         if (!wait_to_idle(timeout_ms)) {
             return -1;
         }
+        ESP_LOGI("TAG", "TAKE SEND %d", at.link_id);
+        xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+        ESP_LOGE("TAG", "TAKE SEND %d", at.link_id);
         state = status::SENDING;
         memcpy(at.get_buf(), buffer, len_to_send);
         ESP_LOG_BUFFER_HEXDUMP("dce", at.get_buf(), len, ESP_LOG_VERBOSE);
@@ -230,6 +236,14 @@ class DCE : public Module {
         }
         return -1;
     }
+    static std::vector<DCE *> dce_list;
+    static bool network_init;
+    static void read_callback(uint8_t *data, size_t len)
+    {
+        for (auto dce : dce_list) {
+            dce->perform_at(data, len);
+        }
+    }
 private:
     esp_modem::SignalGroup signal;
     void close_sock();
@@ -243,15 +257,6 @@ class DCE : public Module {
     int sock {-1};
     int listen_sock {-1};
     int data_ready_fd {-1};
-    static std::vector<DCE*> dce_list;
-    static bool network_init;
-    static void read_callback(uint8_t *data, size_t len)
-    {
-        for (auto dce : dce_list) {
-            dce->perform_at(data, len);
-        }
-    }
-
 };
 std::unique_ptr<DCE> create(const esp_modem::dce_config *config, std::shared_ptr<esp_modem::DTE> dte);
 }

components/esp_modem/examples/modem_tcp_client/main/generate/sock_dce.hpp

@@ -147,7 +147,9 @@ esp_modem::return_type name(ESP_MODEM_COMMAND_PARAMS(__VA_ARGS__));
             return 0;
         }
         at.clear_offsets();
-        // TODO: MUTEX
+        ESP_LOGI("TAG", "TAKE RECV %d", at.link_id);
+        xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+        ESP_LOGE("TAG", "TAKE RECV %d", at.link_id);
         state = status::RECEIVING;
         uint64_t data;
         read(data_ready_fd, &data, sizeof(data));
@@ -170,7 +172,9 @@ esp_modem::return_type name(ESP_MODEM_COMMAND_PARAMS(__VA_ARGS__));
         if (!wait_to_idle(timeout_ms)) {
             return -1;
         }
-        // TODO: MUTEX
+        ESP_LOGI("TAG", "TAKE SEND %d", at.link_id);
+        xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+        ESP_LOGE("TAG", "TAKE SEND %d", at.link_id);
         state = status::SENDING;
         memcpy(at.get_buf(), buffer, len_to_send);
         ESP_LOG_BUFFER_HEXDUMP("dce", at.get_buf(), len, ESP_LOG_VERBOSE);

components/esp_modem/examples/modem_tcp_client/main/modem_client.cpp

@@ -24,7 +24,7 @@
 #include "tcp_transport_at.h"
 
 // #define BROKER_URL "test.mosquitto.org"
-#define BROKER_URL "192.168.0.39"
+#define BROKER_URL "broker.emqx.io"
 #define BROKER_PORT 1883
 
 
@@ -76,8 +76,6 @@ static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_
     }
 }
 
-// sock_dce::DCE::dce_list{};
-
 static void perform(void* ctx);
 
 extern "C" void app_main(void)
@@ -117,7 +115,8 @@ extern "C" void app_main(void)
 
     xTaskCreate(perform, "perform", 4096, dce.get(), 4, nullptr);
 
-    // vTaskDelay(pdMS_TO_TICKS(15000));
+    // vTaskDelay(pdMS_TO_TICKS(5000));
+    // vTaskDelay(portMAX_DELAY);
     /* create another DCE to serve a new connection */
     auto dce1 = sock_dce::create(&dce_config, dte);
     if (!dce1->init()) {
@@ -127,6 +126,13 @@ extern "C" void app_main(void)
     xTaskCreate(perform, "perform", 4096, dce1.get(), 4, nullptr);
 
     xEventGroupWaitBits(event_group, DCE0_DONE | DCE1_DONE, pdFALSE, pdTRUE, portMAX_DELAY);
+#ifdef CONFIG_EXAMPLE_CUSTOM_TCP_TRANSPORT
+    // we release smart pointers, as this example does never exit
+    // and in tcp-transport option we don't need a task to run
+    // so we exit main and keep DCE's "running"
+    dce.release();
+    dce1.release();
+#endif
 }
 
 static void perform(void* ctx)
@@ -137,18 +143,18 @@ static void perform(void* ctx)
     const int id = counter++;
     mqtt_client_id[12] += id;    // assumes different client id per each thread
     esp_mqtt_client_config_t mqtt_config = {};
-    mqtt_config.broker.address.port = BROKER_PORT + id;
+    mqtt_config.broker.address.port = BROKER_PORT; // + id;
     mqtt_config.session.message_retransmit_timeout = 10000;
     mqtt_config.credentials.client_id = mqtt_client_id;
 #ifndef CONFIG_EXAMPLE_CUSTOM_TCP_TRANSPORT
     mqtt_config.broker.address.uri = "mqtt://127.0.0.1";
     dce->start_listening(BROKER_PORT + id);
 #else
     mqtt_config.broker.address.uri = "mqtt://" BROKER_URL;
-    esp_transport_handle_t at = esp_transport_at_init(dce.get());
-    esp_transport_handle_t ssl = esp_transport_tls_init(at);
+    esp_transport_handle_t at = esp_transport_at_init(dce);
+    // esp_transport_handle_t ssl = esp_transport_tls_init(at);
 
-    mqtt_config.network.transport = ssl;
+    mqtt_config.network.transport = at;
 #endif
     esp_mqtt_client_handle_t mqtt_client = esp_mqtt_client_init(&mqtt_config);
     esp_mqtt_client_register_event(mqtt_client, static_cast<esp_mqtt_event_id_t>(ESP_EVENT_ANY_ID), mqtt_event_handler, nullptr);
@@ -174,4 +180,5 @@ static void perform(void* ctx)
     }
 #endif
     xEventGroupSetBits(event_group, id ? DCE0_DONE : DCE1_DONE);
+    vTaskDelete(nullptr);
 }