p23/MeshTalos-Client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(tcpclient): connect to the tcp server on master and receive requests

Browse source
This commit is contained in:
pictures23333 2025-12-17 13:06:17 +08:00
parent 1428ec2d11
commit 38fe78427a
5 changed files with 128 additions and 182 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
main/main.cpp
View file

@ -118,7 +118,7 @@ extern "C" int command_push_image() {
}
}
extern "C" int tcp_server_callback(char *buf, size_t len) {
extern "C" int tcp_client_callback(char *buf, size_t len) {
request_t *req = parse_data(buf, len);
if (!req) { // error -> invalid data
return -2;
@ -167,8 +167,8 @@ extern "C" void app_main() {
mesh_main();
// tcp server
start_tcp_server();
// tcp client
//start_tcp_client();
// Arduino-like setup()

8
main/main.h
View file

@ -23,9 +23,9 @@ static const uint8_t MESH_ID[6] = { 0x77, 0x77, 0x77, 0x77, 0x77, 0x76 };
#define CONFIG_MESH_NON_MESH_AP_CONNECTIONS 0 // number of non-node devices
#define CONFIG_TCP_DEBUG 0
#define CONFIG_TCP_SERVER_BIND_ADDRESS "0.0.0.0"
#define CONFIG_TCP_SERVER_BIND_PORT "8888"
#define CONFIG_TCP_SERVER_RXBUFFER_SIZE 4096 // tcp raw data max size
#define CONFIG_TCP_SERVER_ADDRESS "10.0.0.1"
#define CONFIG_TCP_SERVER_PORT "3030"
#define CONFIG_TCP_RXBUFFER_SIZE 4096 // tcp raw data max size
#define CONFIG_IMAGE_BUF_SLICE_SIZE 2048 // map["data"] max size
#define CONFIG_IMAGE_BUF_SIZE 5624
@ -37,7 +37,7 @@ static const uint8_t MESH_ID[6] = { 0x77, 0x77, 0x77, 0x77, 0x77, 0x76 };
extern "C" {
#endif
int tcp_server_callback(char *buf, size_t len);
int tcp_client_callback(char *buf, size_t len);
#ifdef __cplusplus
}

4
main/mesh_main.c
View file

@ -19,6 +19,7 @@
#include "freertos/semphr.h"
#include "main.h"
#include "tcphelper.h"
/*******************************************************
* Macros
@ -276,6 +277,9 @@ void ip_event_handler(void *arg, esp_event_base_t event_base,
ESP_ERROR_CHECK(esp_netif_get_dns_info(netif, ESP_NETIF_DNS_MAIN, &dns));
mesh_netif_start_root_ap(esp_mesh_is_root(), dns.ip.u_addr.ip4.addr);
//#endif
// tcp client
start_tcp_client();
}

290
main/tcphelper.c
View file

@ -17,7 +17,6 @@
#include "esp_log.h"
#include "nvs_flash.h"
#include "tcphelper.h"
#include "main.h"
/**
@ -107,6 +106,120 @@ static int socket_send(const char *tag, const int sock, const char * data, const
return len;
}
static void tcp_client_task(void *pvParameters)
{
static const char *TAG = "nonblocking-socket-client";
static char rx_buffer[CONFIG_TCP_RXBUFFER_SIZE];
struct addrinfo hints = { .ai_socktype = SOCK_STREAM };
struct addrinfo *address_info;
int sock = INVALID_SOCK;
int res = getaddrinfo(CONFIG_TCP_SERVER_ADDRESS, CONFIG_TCP_SERVER_PORT, &hints, &address_info);
if (res != 0 || address_info == NULL) {
ESP_LOGE(TAG, "couldn't get hostname for `%s` "
"getaddrinfo() returns %d, addrinfo=%p", CONFIG_TCP_SERVER_ADDRESS, res, address_info);
goto error;
}
// Creating client's socket
sock = socket(address_info->ai_family, address_info->ai_socktype, address_info->ai_protocol);
if (sock < 0) {
log_socket_error(TAG, sock, errno, "Unable to create socket");
goto error;
}
ESP_LOGI(TAG, "Socket created, connecting to %s:%s", CONFIG_TCP_SERVER_ADDRESS, CONFIG_TCP_SERVER_PORT);
// Marking the socket as non-blocking
int flags = fcntl(sock, F_GETFL);
if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) == -1) {
log_socket_error(TAG, sock, errno, "Unable to set socket non blocking");
}
if (connect(sock, address_info->ai_addr, address_info->ai_addrlen) != 0) {
if (errno == EINPROGRESS) {
ESP_LOGD(TAG, "connection in progress");
fd_set fdset;
FD_ZERO(&fdset);
FD_SET(sock, &fdset);
// Connection in progress -> have to wait until the connecting socket is marked as writable, i.e. connection completes
res = select(sock+1, NULL, &fdset, NULL, NULL);
if (res < 0) {
log_socket_error(TAG, sock, errno, "Error during connection: select for socket to be writable");
goto error;
} else if (res == 0) {
log_socket_error(TAG, sock, errno, "Connection timeout: select for socket to be writable");
goto error;
} else {
int sockerr;
socklen_t len = (socklen_t)sizeof(int);
if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (void*)(&sockerr), &len) < 0) {
log_socket_error(TAG, sock, errno, "Error when getting socket error using getsockopt()");
goto error;
}
if (sockerr) {
log_socket_error(TAG, sock, sockerr, "Connection error");
goto error;
}
}
} else {
log_socket_error(TAG, sock, errno, "Socket is unable to connect");
goto error;
}
}
// Keep receiving message
do {
// receive
int len = 0;
memset(rx_buffer, 0, sizeof(rx_buffer));
do {
len = try_receive(TAG, sock, rx_buffer, sizeof(rx_buffer));
if (len < 0) {
ESP_LOGE(TAG, "Error occurred during try_receive");
goto error;
}
vTaskDelay(pdMS_TO_TICKS(YIELD_TO_ALL_MS));
} while (len == 0);
ESP_LOGI(TAG, "Received message (len=%d)", len);
// execute command
char tx_buffer[128] = {0};
int result = tcp_client_callback(rx_buffer, len);
switch (result) {
case 0:
strcpy(tx_buffer, "ok\n");
break;
case -2:
strcpy(tx_buffer, "invalid request\n");
break;
case -3:
strcpy(tx_buffer, "locked\n");
break;
default:
strcpy(tx_buffer, "unknown error\n");
break;
}
// send result
len = socket_send(TAG, sock, tx_buffer, strlen(tx_buffer));
if (len < 0) {
ESP_LOGE(TAG, "Error occurred during socket_send");
goto error;
}
ESP_LOGI(TAG, "Written: %.*s", len, tx_buffer);
} while (true);
error:
if (sock != INVALID_SOCK) {
close(sock);
}
free(address_info);
vTaskDelete(NULL);
}
/**
* @brief Returns the string representation of client's address (accepted on this server)
*/
@ -129,177 +242,6 @@ static inline char* get_clients_address(struct sockaddr_storage *source_addr)
return address_str;
}
static void tcp_server_task(void *pvParameters)
{
static char rx_buffer[CONFIG_TCP_SERVER_RXBUFFER_SIZE];
static const char *TAG = "nonblocking-socket-server";
SemaphoreHandle_t *server_ready = pvParameters;
struct addrinfo hints = { .ai_socktype = SOCK_STREAM };
struct addrinfo *address_info;
int listen_sock = INVALID_SOCK;
const size_t max_socks = CONFIG_LWIP_MAX_SOCKETS - 1;
static int sock[CONFIG_LWIP_MAX_SOCKETS - 1];
// Prepare a list of file descriptors to hold client's sockets, mark all of them as invalid, i.e. available
for (int i=0; i<max_socks; ++i) {
sock[i] = INVALID_SOCK;
}
// Translating the hostname or a string representation of an IP to address_info
int res = getaddrinfo(CONFIG_TCP_SERVER_BIND_ADDRESS, CONFIG_TCP_SERVER_BIND_PORT, &hints, &address_info);
if (res != 0 || address_info == NULL) {
ESP_LOGE(TAG, "couldn't get hostname for `%s` "
"getaddrinfo() returns %d, addrinfo=%p", CONFIG_TCP_SERVER_BIND_ADDRESS, res, address_info);
goto error;
}
// Creating a listener socket
listen_sock = socket(address_info->ai_family, address_info->ai_socktype, address_info->ai_protocol);
if (listen_sock < 0) {
log_socket_error(TAG, listen_sock, errno, "Unable to create socket");
goto error;
}
ESP_LOGI(TAG, "Listener socket created");
// Marking the socket as non-blocking
int flags = fcntl(listen_sock, F_GETFL);
if (fcntl(listen_sock, F_SETFL, flags | O_NONBLOCK) == -1) {
log_socket_error(TAG, listen_sock, errno, "Unable to set socket non blocking");
goto error;
}
ESP_LOGI(TAG, "Socket marked as non blocking");
// Binding socket to the given address
int err = bind(listen_sock, address_info->ai_addr, address_info->ai_addrlen);
if (err != 0) {
log_socket_error(TAG, listen_sock, errno, "Socket unable to bind");
goto error;
}
ESP_LOGI(TAG, "Socket bound on %s:%s", CONFIG_TCP_SERVER_BIND_ADDRESS, CONFIG_TCP_SERVER_BIND_PORT);
// Set queue (backlog) of pending connections to one (can be more)
err = listen(listen_sock, 1);
if (err != 0) {
log_socket_error(TAG, listen_sock, errno, "Error occurred during listen");
goto error;
}
ESP_LOGI(TAG, "Socket listening");
xSemaphoreGive(*server_ready);
// Main loop for accepting new connections and serving all connected clients
while (1) {
struct sockaddr_storage source_addr; // Large enough for both IPv4 or IPv6
socklen_t addr_len = sizeof(source_addr);
// Find a free socket
int new_sock_index = 0;
for (new_sock_index=0; new_sock_index<max_socks; ++new_sock_index) {
if (sock[new_sock_index] == INVALID_SOCK) {
break;
}
}
// We accept a new connection only if we have a free socket
if (new_sock_index < max_socks) {
// Try to accept a new connections
sock[new_sock_index] = accept(listen_sock, (struct sockaddr *)&source_addr, &addr_len);
if (sock[new_sock_index] < 0) {
if (errno == EWOULDBLOCK) { // The listener socket did not accepts any connection
// continue to serve open connections and try to accept again upon the next iteration
ESP_LOGV(TAG, "No pending connections...");
} else {
log_socket_error(TAG, listen_sock, errno, "Error when accepting connection");
goto error;
}
} else {
// We have a new client connected -> print it's address
ESP_LOGI(TAG, "[sock=%d]: Connection accepted from IP:%s", sock[new_sock_index], get_clients_address(&source_addr));
// ...and set the client's socket non-blocking
flags = fcntl(sock[new_sock_index], F_GETFL);
if (fcntl(sock[new_sock_index], F_SETFL, flags | O_NONBLOCK) == -1) {
log_socket_error(TAG, sock[new_sock_index], errno, "Unable to set socket non blocking");
goto error;
}
ESP_LOGI(TAG, "[sock=%d]: Socket marked as non blocking", sock[new_sock_index]);
}
}
// We serve all the connected clients in this loop
for (int i=0; i<max_socks; ++i) {
if (sock[i] != INVALID_SOCK) {
// This is an open socket -> try to serve it
memset(rx_buffer, 0, sizeof(rx_buffer));
int len = try_receive(TAG, sock[i], rx_buffer, sizeof(rx_buffer));
if (len < 0) {
// Error occurred within this client's socket -> close and mark invalid
ESP_LOGI(TAG, "[sock=%d]: try_receive() returned %d -> closing the socket", sock[i], len);
close(sock[i]);
sock[i] = INVALID_SOCK;
} else if (len > 0) {
// Received some data -> echo back
//ESP_LOGI(TAG, "[sock=%d]: Received %.*s", sock[i], len, rx_buffer);
ESP_LOGI(TAG, "[sock=%d]: Received message (len=%d)", sock[i], len);
char tx_buffer[128] = {0};
int result = tcp_server_callback(rx_buffer, len);
switch (result) {
case 0:
strcpy(tx_buffer, "ok\n");
break;
case -2:
strcpy(tx_buffer, "invalid request\n");
break;
case -3:
strcpy(tx_buffer, "locked\n");
break;
default:
strcpy(tx_buffer, "unknown error\n");
break;
}
len = socket_send(TAG, sock[i], tx_buffer, strlen(tx_buffer));
if (len < 0) {
// Error occurred on write to this socket -> close it and mark invalid
ESP_LOGI(TAG, "[sock=%d]: socket_send() returned %d -> closing the socket", sock[i], len);
close(sock[i]);
sock[i] = INVALID_SOCK;
} else {
// Successfully echoed to this socket
ESP_LOGI(TAG, "[sock=%d]: Written %.*s", sock[i], strlen(tx_buffer), tx_buffer);
}
}
} // one client's socket
} // for all sockets
// Yield to other tasks
vTaskDelay(pdMS_TO_TICKS(YIELD_TO_ALL_MS));
}
error:
if (listen_sock != INVALID_SOCK) {
close(listen_sock);
}
for (int i=0; i<max_socks; ++i) {
if (sock[i] != INVALID_SOCK) {
close(sock[i]);
}
}
free(address_info);
vTaskDelete(NULL);
}
void start_tcp_server(void) {
SemaphoreHandle_t server_ready = xSemaphoreCreateBinary();
assert(server_ready);
xTaskCreate(tcp_server_task, "tcp_server", 4096, &server_ready, 5, NULL);
xSemaphoreTake(server_ready, portMAX_DELAY);
vSemaphoreDelete(server_ready);
void start_tcp_client(void) {
xTaskCreate(tcp_client_task, "tcp_client", 4096, NULL, 5, NULL);
}

2
main/tcphelper.h
View file

@ -5,7 +5,7 @@ extern "C" {
#endif
// functions
void start_tcp_server(void);
void start_tcp_client(void);
#ifdef __cplusplus
}