fix(modem): WIP for multiple conn

Author: david-cermak

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

@@ -19,6 +19,7 @@ constexpr auto const *TAG = "sock_dce";
 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)
@@ -82,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;
@@ -103,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;
@@ -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,12 +267,15 @@ 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);
@@ -268,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);
 
@@ -311,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/generate/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);
@@ -64,6 +65,7 @@ class Responder {
     }
 
     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;
@@ -72,6 +74,7 @@ class Responder {
     {
 #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;
@@ -144,6 +147,7 @@ esp_modem::return_type name(ESP_MODEM_COMMAND_PARAMS(__VA_ARGS__));
             return 0;
         }
         at.clear_offsets();
+        // TODO: MUTEX
         state = status::RECEIVING;
         uint64_t data;
         read(data_ready_fd, &data, sizeof(data));
@@ -166,6 +170,7 @@ esp_modem::return_type name(ESP_MODEM_COMMAND_PARAMS(__VA_ARGS__));
         if (!wait_to_idle(timeout_ms)) {
             return -1;
         }
+        // TODO: MUTEX
         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

@@ -23,14 +23,17 @@
 #include "tcp_transport_mbedtls.h"
 #include "tcp_transport_at.h"
 
-#define BROKER_URL "test.mosquitto.org"
+// #define BROKER_URL "test.mosquitto.org"
+#define BROKER_URL "192.168.0.39"
 #define BROKER_PORT 1883
 
 
 static const char *TAG = "modem_client";
 static EventGroupHandle_t event_group = NULL;
 static const int CONNECT_BIT = BIT0;
 static const int GOT_DATA_BIT = BIT2;
+static const int DCE0_DONE = BIT3;
+static const int DCE1_DONE = BIT4;
 
 static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data)
 {
@@ -75,9 +78,10 @@ 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)
 {
-
     /* Init and register system/core components */
     ESP_ERROR_CHECK(esp_netif_init());
     ESP_ERROR_CHECK(esp_event_loop_create_default());
@@ -111,12 +115,34 @@ extern "C" void app_main(void)
         return;
     }
 
+    xTaskCreate(perform, "perform", 4096, dce.get(), 4, nullptr);
+
+    // vTaskDelay(pdMS_TO_TICKS(15000));
+    /* create another DCE to serve a new connection */
+    auto dce1 = sock_dce::create(&dce_config, dte);
+    if (!dce1->init()) {
+        ESP_LOGE(TAG,  "Failed to setup network");
+        return;
+    }
+    xTaskCreate(perform, "perform", 4096, dce1.get(), 4, nullptr);
+
+    xEventGroupWaitBits(event_group, DCE0_DONE | DCE1_DONE, pdFALSE, pdTRUE, portMAX_DELAY);
+}
+
+static void perform(void* ctx)
+{
+    auto dce = static_cast<sock_dce::DCE*>(ctx);
+    char mqtt_client_id[] = "MQTT_CLIENT_0";
+    static int counter = 0;
+    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;
+    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);
+    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());
@@ -146,8 +172,6 @@ extern "C" void app_main(void)
             ESP_LOGI(TAG, "Network reinitialized, retrying");
         }
     }
-#else
-    vTaskDelay(portMAX_DELAY);
 #endif
-
+    xEventGroupSetBits(event_group, id ? DCE0_DONE : DCE1_DONE);
 }

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

@@ -67,11 +67,18 @@ command_result net_open(CommandableIf *t)
     ESP_LOGI(TAG, "Multiple connections mode enabled");
 
     // Set passive receive mode (1) for better control
-    ret = set_rx_mode(t, 1);
-    if (ret != command_result::OK) {
-        ESP_LOGE(TAG, "Failed to set preferred Rx mode");
-        return ret;
+    for (int i = 0; i < 2; i++) {
+        std::string cmd = "AT+CIPRECVTYPE=" + std::to_string(i) + ",1\r\n";
+        dce_commands::generic_command(t, cmd, "OK", "ERROR", 1000);
     }
+    // std::string cmd = "AT+CIPRECVTYPE=" + std::to_string(link_id) + "," + std::to_string(mode) + "\r\n";
+    // return dce_commands::generic_command(t, cmd, "OK", "ERROR", 1000);
+    //
+    // ret = set_rx_mode(t, 1);
+    // if (ret != command_result::OK) {
+    //     ESP_LOGE(TAG, "Failed to set preferred Rx mode");
+    //     return ret;
+    // }
     return command_result::OK;
 }
 
@@ -111,6 +118,7 @@ command_result net_close(CommandableIf *t)
 
 command_result tcp_close(CommandableIf *t)
 {
+    return command_result::OK;
     ESP_LOGV(TAG, "%s", __func__);
     // Use link ID 0 for closing connection
     const int link_id = 0;
@@ -192,6 +200,12 @@ Responder::ret Responder::recv(uint8_t *data, size_t len)
     auto *recv_data = (char *)data;
 
     if (data_to_recv == 0) {
+        // const std::string_view error_str = "ERROR";
+        // const std::string_view data_sv(recv_data, len);
+        // if (data_sv.find(error_str) == std::string_view::npos) {
+        //     // no data,
+        //     return ret::OK;
+        // }
         const std::string_view head = "+CIPRECVDATA:";
 
         // Find the response header
@@ -200,7 +214,7 @@ Responder::ret Responder::recv(uint8_t *data, size_t len)
         });
 
         if (head_pos == recv_data + len) {
-            return ret::FAIL;
+            return ret::IN_PROGRESS;
         }
 
         // Find the end of the length field
@@ -316,6 +330,17 @@ Responder::ret Responder::connect(std::string_view response)
     }
     return ret::IN_PROGRESS;
 }
+Responder::ret Responder::check_urc(status state, std::string_view &response)
+{
+    // Handle data notifications - in multiple connections mode, format is +IPD,<link ID>,<len>
+    std::string expected_urc = "+IPD," + std::to_string(link_id);
+    if (response.find(expected_urc) != std::string::npos) {
+        uint64_t data_ready = 1;
+        write(data_ready_fd, &data_ready, sizeof(data_ready));
+        ESP_LOGD(TAG, "Data available notification");
+    }
+    return ret::IN_PROGRESS;
+}
 
 Responder::ret Responder::check_async_replies(status state, std::string_view &response)
 {
@@ -336,13 +361,6 @@ Responder::ret Responder::check_async_replies(status state, std::string_view &re
         return ret::FAIL;
     }
 
-    // Handle data notifications - in multiple connections mode, format is +IPD,<link ID>,<len>
-    if (response.find("+IPD,") != std::string::npos) {
-        uint64_t data_ready = 1;
-        write(data_ready_fd, &data_ready, sizeof(data_ready));
-        ESP_LOGD(TAG, "Data available notification");
-    }
-
     if (state == status::SENDING) {
         return send(response);
     } else if (state == status::CONNECTING) {