feat(modem): Add support for multiple connection in AT based example

Author: david-cermak

components/esp_modem/examples/modem_tcp_client/README.md

@@ -22,3 +22,18 @@ To enable this mode, please set `EXAMPLE_CUSTOM_TCP_TRANSPORT=y`
 This configuration could be used with any network library, which is connecting to a localhost endpoint instead of remote one. This example creates a localhost listener which basically mimics the remote endpoint by forwarding the traffic between the library and the TCP/socket layer of the modem (which is already secure if the TLS is used in the network library)
 
 ![with localhost listener](at_client_localhost.png)
+
+### Multi-connection support
+
+This example supports opening multiple TCP connections concurrently when the modem firmware allows it.
+
+- ESP-AT: Multi-connection mode is enabled via `AT+CIPMUX=1`. The example assigns a unique link ID per DCE instance and includes the link ID in `CIPSTART/CIPSEND/CIPRECVDATA` commands.
+- BG96/SIM7600: The example uses module-specific multi-connection syntax (for example `QIOPEN/CIPOPEN` with a connection ID) and tracks link IDs internally.
+
+How it works:
+- The `sock_dce` layer creates multiple DCE instances over a shared DTE. A lightweight mutex coordinates access to the UART so only one DCE issues AT commands at a time.
+- Asynchronous URCs (for example `+IPD`, `+QIURC`, `+CIPRXGET: 1,<cid>`) wake the corresponding DCE which then performs receive operations for its link.
+
+Usage:
+- `app_main` starts two DCE tasks to demonstrate concurrent connections. Adjust the number of DCE instances as needed.
+- For ESP-AT, ensure your firmware supports `CIPMUX=1` and passive receive (`CIPRECVTYPE`).

components/esp_modem/examples/modem_tcp_client/main/Kconfig.projbuild

@@ -44,6 +44,18 @@ menu "Example Configuration"
         help
             Set APN (Access Point Name), a logical name to choose data network
 
+    config EXAMPLE_USE_TLS
+        bool "Use TLS for MQTT broker"
+        default n
+        help
+            Enable TLS for connection to the MQTT broker.
+
+    config EXAMPLE_BROKER_HOST
+        string "MQTT broker host"
+        default "test.mosquitto.org"
+        help
+            Hostname or IP address of the MQTT broker.
+
     menu "UART Configuration"
         config EXAMPLE_MODEM_UART_TX_PIN
             int "TXD Pin Number"

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

@@ -4,6 +4,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include <algorithm>
 #include <charconv>
 #include <sys/socket.h>
 #include "esp_vfs.h"
@@ -14,6 +15,29 @@
 namespace sock_dce {
 
 constexpr auto const *TAG = "sock_dce";
+constexpr auto WAIT_TO_IDLE_TIMEOUT = 5000;
+
+// Definition of the static member variables
+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)
+    : Module(std::move(dte_arg), config)
+{
+    dce_list.push_back(this);
+}
+
+// Destructor - remove this DCE instance from the static list
+DCE::~DCE()
+{
+    auto it = std::find(dce_list.begin(), dce_list.end(), this);
+    if (it != dce_list.end()) {
+        dce_list.erase(it);
+    }
+}
 
 
 bool DCE::perform_sock()
@@ -61,13 +85,26 @@ bool DCE::perform_sock()
 
 void DCE::perform_at(uint8_t *data, size_t len)
 {
-    ESP_LOG_BUFFER_HEXDUMP(TAG, data, len, ESP_LOG_VERBOSE);
+    if (state != status::RECEIVING) {
+        std::string_view resp_sv((char *)data, len);
+        at.check_urc(state, resp_sv);
+        if (state == status::IDLE) {
+            return;
+        }
+    }
+    // Trace incoming AT bytes when handling a response; use DEBUG level
+    ESP_LOG_BUFFER_HEXDUMP(TAG, data, len, ESP_LOG_DEBUG);
     switch (at.process_data(state, data, len)) {
     case Responder::ret::OK:
+        // Release DTE access for this link after processing data
+        ESP_LOGD(TAG, "GIVE data %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::IDLE;
         signal.set(IDLE);
         return;
     case Responder::ret::FAIL:
+        ESP_LOGD(TAG, "GIVE data %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::FAILED;
         signal.set(IDLE);
         return;
@@ -82,10 +119,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_LOGD(TAG, "GIVE command %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::IDLE;
         signal.set(IDLE);
         return;
     case Responder::ret::FAIL:
+        ESP_LOGD(TAG, "GIVE command %d", at.link_id);
+        xSemaphoreGive(at.s_dte_mutex);
         state = status::FAILED;
         signal.set(IDLE);
         return;
@@ -121,7 +162,7 @@ bool DCE::at_to_sock()
     uint64_t data;
     read(data_ready_fd, &data, sizeof(data));
     ESP_LOGD(TAG, "select read: modem data available %" PRIu64, data);
-    if (!signal.wait(IDLE, 1000)) {
+    if (!signal.wait(IDLE, WAIT_TO_IDLE_TIMEOUT)) {
         ESP_LOGE(TAG, "Failed to get idle");
         close_sock();
         return false;
@@ -131,6 +172,10 @@ bool DCE::at_to_sock()
         close_sock();
         return false;
     }
+    // Take DTE mutex before issuing receive on this link
+    ESP_LOGD(TAG, "TAKE RECV %d", at.link_id);
+    xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+    ESP_LOGD(TAG, "TAKEN RECV %d", at.link_id);
     state = status::RECEIVING;
     at.start_receiving(at.get_buf_len());
     return true;
@@ -139,7 +184,7 @@ bool DCE::at_to_sock()
 bool DCE::sock_to_at()
 {
     ESP_LOGD(TAG,  "socket read: data available");
-    if (!signal.wait(IDLE, 1000)) {
+    if (!signal.wait(IDLE, WAIT_TO_IDLE_TIMEOUT)) {
         ESP_LOGE(TAG,  "Failed to get idle");
         close_sock();
         return false;
@@ -149,6 +194,10 @@ bool DCE::sock_to_at()
         close_sock();
         return false;
     }
+    // Take DTE mutex before issuing send on this link
+    ESP_LOGD(TAG, "TAKE SEND %d", at.link_id);
+    xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+    ESP_LOGD(TAG, "TAKEN SEND %d", at.link_id);
     state = status::SENDING;
     int len = ::recv(sock, at.get_buf(), at.get_buf_len(), 0);
     if (len < 0) {
@@ -201,7 +250,7 @@ void DCE::start_listening(int port)
     }
     int opt = 1;
     setsockopt(listen_sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
-    ESP_LOGI(TAG, "Socket created");
+    ESP_LOGD(TAG, "Socket created");
     struct sockaddr_in addr = { };
     addr.sin_family = AF_INET;
     addr.sin_port = htons(port);
@@ -213,7 +262,7 @@ void DCE::start_listening(int port)
         ESP_LOGE(TAG, "Socket unable to bind: errno %d", errno);
         return;
     }
-    ESP_LOGI(TAG, "Socket bound, port %d", 1883);
+    ESP_LOGD(TAG, "Socket bound, port %d", 1883);
     err = listen(listen_sock, 1);
     if (err != 0) {
         ESP_LOGE(TAG, "Error occurred during listen: errno %d", errno);
@@ -224,12 +273,12 @@ void DCE::start_listening(int port)
 
 bool DCE::connect(std::string host, int port)
 {
-    dte->on_read(nullptr);
-    tcp_close();
-    dte->on_read([this](uint8_t *data, size_t len) {
-        this->perform_at(data, len);
-        return esp_modem::command_result::TIMEOUT;
-    });
+    data_ready_fd = eventfd(0, EFD_SUPPORT_ISR);
+    assert(data_ready_fd > 0);
+    // Take DTE mutex before starting connect for this link
+    ESP_LOGD(TAG, "TAKE CONNECT %d", at.link_id);
+    xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+    ESP_LOGD(TAG, "TAKEN CONNECT %d", at.link_id);
     if (!at.start_connecting(host, port)) {
         ESP_LOGE(TAG, "Unable to start connecting");
         dte->on_read(nullptr);
@@ -241,12 +290,15 @@ bool DCE::connect(std::string host, int port)
 
 bool DCE::init()
 {
+    if (network_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);
 
-    data_ready_fd = eventfd(0, EFD_SUPPORT_ISR);
-    assert(data_ready_fd > 0);
-
     dte->on_read(nullptr);
     const int retries = 5;
     int i = 0;
@@ -287,6 +339,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,19 @@ class Responder {
         return total_len;
     }
 
+    // Unique link identifier used to target multi-connection AT commands
+    int link_id{s_link_id++};
+    // Shared mutex guarding DTE access across concurrent DCE instances
+    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;
@@ -101,6 +108,8 @@ class Responder {
 class DCE : public Module {
     using Module::Module;
 public:
+    DCE(std::shared_ptr<esp_modem::DTE> dte_arg, const esp_modem_dce_config *config);
+    ~DCE();
 
     /**
      * @brief Opens network in AT command mode
@@ -163,6 +172,10 @@ class DCE : public Module {
             return 0;
         }
         at.clear_offsets();
+        // Take DTE mutex before issuing receive on this link
+        ESP_LOGD("TAG", "TAKE RECV %d", at.link_id);
+        xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+        ESP_LOGD("TAG", "TAKEN RECV %d", at.link_id);
         state = status::RECEIVING;
         uint64_t data;
         read(data_ready_fd, &data, sizeof(data));
@@ -184,6 +197,10 @@ class DCE : public Module {
         if (!wait_to_idle(timeout_ms)) {
             return -1;
         }
+        // Take DTE mutex before issuing send on this link
+        ESP_LOGD("TAG", "TAKE SEND %d", at.link_id);
+        xSemaphoreTake(at.s_dte_mutex, portMAX_DELAY);
+        ESP_LOGD("TAG", "TAKEN 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);
@@ -224,6 +241,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();