20560534171

Committed 28 Dec 2025 10:39PM UTC coverage: 48.047% (-0.03%) from 48.078%

Build # 20560534171

Build Type

Pull #2746

github

Committed by

web-flow

Commit Message

Merge bb2347827 into 2460e1568

Pull Request Pull Request #2746: Configuration and component cleanup

Run Details

18 of 45 new or added lines in 4 files covered. (40.0%)

13 existing lines in 6 files now uncovered.

17741 of 36924 relevant lines covered (48.05%)

464.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.6

/src/mavsdk/core/tcp_client_connection.cpp

#include "tcp_client_connection.h"
#include "log.h"

#include <cassert>
#include <sstream>
#include <utility>
#include <thread>
#include <sstream>

#ifdef WINDOWS
#ifndef MINGW
#pragma comment(lib, "Ws2_32.lib") // Without this, Ws2_32.lib is not included in static library.
#endif
#else
#include <netinet/in.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <errno.h>
#include <netdb.h>
#endif

namespace mavsdk {

/* change to remote_ip and remote_port */
TcpClientConnection::TcpClientConnection(
    Connection::ReceiverCallback receiver_callback,
    Connection::LibmavReceiverCallback libmav_receiver_callback,
    MavsdkImpl& mavsdk_impl,
    std::string remote_ip,
    int remote_port,
    ForwardingOption forwarding_option) :
    Connection(
        std::move(receiver_callback),
        std::move(libmav_receiver_callback),
        mavsdk_impl,
        forwarding_option),
    _remote_ip(std::move(remote_ip)),
    _remote_port_number(remote_port),
    _should_exit(false)
{}

TcpClientConnection::~TcpClientConnection()
{
    // If no one explicitly called stop before, we should at least do it.
    stop();
}

ConnectionResult TcpClientConnection::start()
{
    if (!start_mavlink_receiver()) {
        return ConnectionResult::ConnectionsExhausted;
    }

    if (!start_libmav_receiver()) {
        return ConnectionResult::ConnectionsExhausted;
    }

    ConnectionResult ret = setup_port();
    if (ret != ConnectionResult::Success) {
        return ret;
    }

    start_recv_thread();

    return ConnectionResult::Success;
}

ConnectionResult TcpClientConnection::setup_port()
{
#ifdef WINDOWS
    WSADATA wsa;
    if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) {
        LogErr() << "Error: Winsock failed, error: " << get_socket_error_string(WSAGetLastError());
        return ConnectionResult::SocketError;
    }
#endif

    std::lock_guard<std::mutex> lock(_mutex);

    int new_fd = socket(AF_INET, SOCK_STREAM, 0);

    if (new_fd < 0) {
        LogErr() << "socket error" << strerror(errno);
        return ConnectionResult::SocketError;
    }

    struct sockaddr_in remote_addr{};
    remote_addr.sin_family = AF_INET;
    remote_addr.sin_port = htons(_remote_port_number);

    struct hostent* hp;
    hp = gethostbyname(_remote_ip.c_str());
    if (hp == nullptr) {
        LogErr() << "Could not get host by name";
        return ConnectionResult::SocketConnectionError;
    }

    memcpy(&remote_addr.sin_addr, hp->h_addr, hp->h_length);

    if (connect(new_fd, reinterpret_cast<sockaddr*>(&remote_addr), sizeof(struct sockaddr_in)) <
        0) {
        LogErr() << "Connect error: " << strerror(errno);
        return ConnectionResult::SocketConnectionError;
    }

    // Set receive timeout cross-platform
    const unsigned timeout_ms = 500;

#if defined(WINDOWS)
    setsockopt(new_fd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeout_ms, sizeof(timeout_ms));
#else
    struct timeval tv;
    tv.tv_sec = 0;
    tv.tv_usec = timeout_ms * 1000;
    setsockopt(new_fd, SOL_SOCKET, SO_RCVTIMEO, (const void*)&tv, sizeof(tv));
#endif

    _socket_fd.reset(new_fd);

    return ConnectionResult::Success;
}

void TcpClientConnection::start_recv_thread()
{
    _recv_thread = std::make_unique<std::thread>(&TcpClientConnection::receive, this);
}

ConnectionResult TcpClientConnection::stop()
{
    _should_exit = true;

    if (_recv_thread) {
        _recv_thread->join();
        _recv_thread.reset();
    }

    {
        std::lock_guard<std::mutex> lock(_mutex);
        _socket_fd.close();
    }

    // We need to stop this after stopping the receive thread, otherwise
    // it can happen that we interfere with the parsing of a message.
    stop_mavlink_receiver();

    return ConnectionResult::Success;
}

std::pair<bool, std::string> TcpClientConnection::send_message(const mavlink_message_t& message)
{
    // Convert message to raw bytes and use common send path
    uint8_t buffer[MAVLINK_MAX_PACKET_LEN];
    uint16_t buffer_len = mavlink_msg_to_send_buffer(buffer, &message);

    return send_raw_bytes(reinterpret_cast<const char*>(buffer), buffer_len);
}

std::pair<bool, std::string> TcpClientConnection::send_raw_bytes(const char* bytes, size_t length)
{
    std::pair<bool, std::string> result;

    if (_remote_ip.empty()) {
        result.first = false;
        result.second = "Remote IP unknown";
        LogErr() << result.second;
        return result;
    }

    if (_remote_port_number == 0) {
        result.first = false;
        result.second = "Remote port unknown";
        LogErr() << result.second;
        return result;
    }

    // Hold mutex during send to prevent socket from being closed mid-send
    std::lock_guard<std::mutex> lock(_mutex);

    if (_socket_fd.empty()) {
        result.first = false;
        result.second = "Not connected";
        return result;
    }

#if !defined(MSG_NOSIGNAL)
    auto flags = 0;
#else
    auto flags = MSG_NOSIGNAL;
#endif

    const auto send_len = send(_socket_fd.get(), bytes, length, flags);

    if (send_len != static_cast<std::remove_cv_t<decltype(send_len)>>(length)) {
        std::stringstream ss;
        ss << "Send failure: " << strerror(errno);
        LogErr() << ss.str();
        result.first = false;
        result.second = ss.str();
        return result;
    }

    result.first = true;
    return result;
}

void TcpClientConnection::receive()
{
    // Enough for MTU 1500 bytes.
    char buffer[2048];

    while (!_should_exit) {
        // Get socket fd with mutex protection, then release mutex before blocking recv
        SocketHolder::DescriptorType socket_fd;
        {
            std::lock_guard<std::mutex> lock(_mutex);
            if (_socket_fd.empty()) {
                // Socket not connected, need to reconnect
                socket_fd = SocketHolder::invalid_socket_fd;
            } else {
                socket_fd = _socket_fd.get();
            }
        }

        if (socket_fd == SocketHolder::invalid_socket_fd) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
            setup_port();
            continue;
        }

        const auto recv_len = recv(socket_fd, buffer, sizeof(buffer), 0);

        if (recv_len == 0) {
            // Connection closed, just try again.
            LogInfo() << "TCP connection closed, trying to reconnect...";
            std::this_thread::sleep_for(std::chrono::seconds(1));
            setup_port();
            continue;
        }

        if (recv_len < 0) {
#ifdef WINDOWS
            int err = WSAGetLastError();
            if (err == WSAEWOULDBLOCK || err == WSAETIMEDOUT) {
                // Timeout, just try again.
                continue;
            }
            LogErr() << "TCP receive error: " << get_socket_error_string(err)
                     << ", trying to reconnect...";
#else
            if (errno == EAGAIN || errno == ETIMEDOUT) {
                // Timeout, just try again.
                continue;
            }
            LogErr() << "TCP receive error: " << strerror(errno) << ", trying to reconnect...";
#endif
            std::this_thread::sleep_for(std::chrono::seconds(1));
            setup_port();
            continue;
        }

        _mavlink_receiver->set_new_datagram(buffer, static_cast<int>(recv_len));

        auto parse_result = _mavlink_receiver->parse_message();
        while (parse_result != MavlinkReceiver::ParseResult::NoneAvailable) {
            receive_message(parse_result, _mavlink_receiver->get_last_message(), this);
            parse_result = _mavlink_receiver->parse_message();
        }

        // Also parse with libmav if available
        if (_libmav_receiver) {
            _libmav_receiver->set_new_datagram(buffer, static_cast<int>(recv_len));

            while (_libmav_receiver->parse_message()) {
                receive_libmav_message(_libmav_receiver->get_last_message(), this);
            }
        }
    }
}

} // namespace mavsdk

1	#include "tcp_client_connection.h"
2	#include "log.h"
3
4	#include <cassert>
5	#include <sstream>
6	#include <utility>
7	#include <thread>
8	#include <sstream>
9
10	#ifdef WINDOWS
11	#ifndef MINGW
12	#pragma comment(lib, "Ws2_32.lib") // Without this, Ws2_32.lib is not included in static library.
13	#endif
14	#else
15	#include <netinet/in.h>
16	#include <sys/socket.h>
17	#include <arpa/inet.h>
18	#include <errno.h>
19	#include <netdb.h>
20	#endif
21
22	namespace mavsdk {
23
24	/* change to remote_ip and remote_port */
25	TcpClientConnection::TcpClientConnection(	4✔
26	Connection::ReceiverCallback receiver_callback,
27	Connection::LibmavReceiverCallback libmav_receiver_callback,
28	MavsdkImpl& mavsdk_impl,
29	std::string remote_ip,
30	int remote_port,
31	ForwardingOption forwarding_option) :	4✔
32	Connection(
33	std::move(receiver_callback),	4✔
34	std::move(libmav_receiver_callback),	4✔
35	mavsdk_impl,
36	forwarding_option),
37	_remote_ip(std::move(remote_ip)),	4✔
38	_remote_port_number(remote_port),	4✔
39	_should_exit(false)	12✔
40	{}	4✔
41
42	TcpClientConnection::~TcpClientConnection()	8✔
43	{
44	// If no one explicitly called stop before, we should at least do it.
45	stop();	4✔
46	}	8✔
47
48	ConnectionResult TcpClientConnection::start()	4✔
49	{
50	if (!start_mavlink_receiver()) {	4✔
51	return ConnectionResult::ConnectionsExhausted;	×
52	}
53
54	if (!start_libmav_receiver()) {	4✔
55	return ConnectionResult::ConnectionsExhausted;	×
56	}
57
58	ConnectionResult ret = setup_port();	4✔
59	if (ret != ConnectionResult::Success) {	4✔
60	return ret;	×
61	}
62
63	start_recv_thread();	4✔
64
65	return ConnectionResult::Success;	4✔
66	}
67
68	ConnectionResult TcpClientConnection::setup_port()	7✔
69	{
70	#ifdef WINDOWS
71	WSADATA wsa;
72	if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) {
73	LogErr() << "Error: Winsock failed, error: " << get_socket_error_string(WSAGetLastError());
74	return ConnectionResult::SocketError;
75	}
76	#endif
77
78	std::lock_guard<std::mutex> lock(_mutex);	7✔
79
80	int new_fd = socket(AF_INET, SOCK_STREAM, 0);	7✔
81
82	if (new_fd < 0) {	7✔
83	LogErr() << "socket error" << strerror(errno);	×
84	return ConnectionResult::SocketError;	×
85	}
86
87	struct sockaddr_in remote_addr{};	7✔
88	remote_addr.sin_family = AF_INET;	7✔
89	remote_addr.sin_port = htons(_remote_port_number);	7✔
90
91	struct hostent* hp;
92	hp = gethostbyname(_remote_ip.c_str());	7✔
93	if (hp == nullptr) {	7✔
94	LogErr() << "Could not get host by name";	×
95	return ConnectionResult::SocketConnectionError;	×
96	}
97
98	memcpy(&remote_addr.sin_addr, hp->h_addr, hp->h_length);	7✔
99
100	if (connect(new_fd, reinterpret_cast<sockaddr*>(&remote_addr), sizeof(struct sockaddr_in)) <	7✔
101	0) {
102	LogErr() << "Connect error: " << strerror(errno);	2✔
103	return ConnectionResult::SocketConnectionError;	2✔
104	}
105
106	// Set receive timeout cross-platform
107	const unsigned timeout_ms = 500;	5✔
108
109	#if defined(WINDOWS)
110	setsockopt(new_fd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeout_ms, sizeof(timeout_ms));
111	#else
112	struct timeval tv;
113	tv.tv_sec = 0;	5✔
114	tv.tv_usec = timeout_ms * 1000;	5✔
115	setsockopt(new_fd, SOL_SOCKET, SO_RCVTIMEO, (const void*)&tv, sizeof(tv));	5✔
116	#endif
117
118	_socket_fd.reset(new_fd);	5✔
119
120	return ConnectionResult::Success;	5✔
121	}	7✔
122
123	void TcpClientConnection::start_recv_thread()	4✔
124	{
125	_recv_thread = std::make_unique<std::thread>(&TcpClientConnection::receive, this);	4✔
126	}	4✔
127
128	ConnectionResult TcpClientConnection::stop()	4✔
129	{
130	_should_exit = true;	4✔
131
132	if (_recv_thread) {	4✔
133	_recv_thread->join();	4✔
134	_recv_thread.reset();	4✔
135	}
136
137	{
138	std::lock_guard<std::mutex> lock(_mutex);	4✔
139	_socket_fd.close();	4✔
140	}	4✔
141
142	// We need to stop this after stopping the receive thread, otherwise
143	// it can happen that we interfere with the parsing of a message.
144	stop_mavlink_receiver();	4✔
145
146	return ConnectionResult::Success;	4✔
147	}
148
149	std::pair<bool, std::string> TcpClientConnection::send_message(const mavlink_message_t& message)	33✔
150	{
151	// Convert message to raw bytes and use common send path
152	uint8_t buffer[MAVLINK_MAX_PACKET_LEN];
153	uint16_t buffer_len = mavlink_msg_to_send_buffer(buffer, &message);	33✔
154
155	return send_raw_bytes(reinterpret_cast<const char*>(buffer), buffer_len);	33✔
156	}
157
158	std::pair<bool, std::string> TcpClientConnection::send_raw_bytes(const char* bytes, size_t length)	33✔
159	{
160	std::pair<bool, std::string> result;	33✔
161
162	if (_remote_ip.empty()) {	33✔
163	result.first = false;	×
164	result.second = "Remote IP unknown";	×
165	LogErr() << result.second;	×
166	return result;	×
167	}
168
169	if (_remote_port_number == 0) {	33✔
170	result.first = false;	×
171	result.second = "Remote port unknown";	×
172	LogErr() << result.second;	×
173	return result;	×
174	}
175
176	// Hold mutex during send to prevent socket from being closed mid-send
177	std::lock_guard<std::mutex> lock(_mutex);	33✔
178
179	if (_socket_fd.empty()) {	33✔
180	result.first = false;	×
181	result.second = "Not connected";	×
182	return result;	×
183	}
184
185	#if !defined(MSG_NOSIGNAL)
186	auto flags = 0;
187	#else
188	auto flags = MSG_NOSIGNAL;	33✔
189	#endif
190
191	const auto send_len = send(_socket_fd.get(), bytes, length, flags);	33✔
192
193	if (send_len != static_cast<std::remove_cv_t<decltype(send_len)>>(length)) {	33✔
UNCOV 194	std::stringstream ss;	×
UNCOV 195	ss << "Send failure: " << strerror(errno);	×
UNCOV 196	LogErr() << ss.str();	×
UNCOV 197	result.first = false;	×
UNCOV 198	result.second = ss.str();	×
UNCOV 199	return result;	×
UNCOV 200	}	×
201
202	result.first = true;	33✔
203	return result;	33✔
204	}	33✔
205
206	void TcpClientConnection::receive()	4✔
207	{
208	// Enough for MTU 1500 bytes.
209	char buffer[2048];
210
211	while (!_should_exit) {	57✔
212	// Get socket fd with mutex protection, then release mutex before blocking recv
213	SocketHolder::DescriptorType socket_fd;
214	{
215	std::lock_guard<std::mutex> lock(_mutex);	53✔
216	if (_socket_fd.empty()) {	53✔
217	// Socket not connected, need to reconnect
218	socket_fd = SocketHolder::invalid_socket_fd;	×
219	} else {
220	socket_fd = _socket_fd.get();	53✔
221	}
222	}	53✔
223
224	if (socket_fd == SocketHolder::invalid_socket_fd) {	53✔
225	std::this_thread::sleep_for(std::chrono::seconds(1));	×
226	setup_port();	×
227	continue;	×
228	}
229
230	const auto recv_len = recv(socket_fd, buffer, sizeof(buffer), 0);	53✔
231
232	if (recv_len == 0) {	53✔
233	// Connection closed, just try again.
234	LogInfo() << "TCP connection closed, trying to reconnect...";	3✔
235	std::this_thread::sleep_for(std::chrono::seconds(1));	3✔
236	setup_port();	3✔
237	continue;	3✔
238	}
239
240	if (recv_len < 0) {	50✔
241	#ifdef WINDOWS
242	int err = WSAGetLastError();
243	if (err == WSAEWOULDBLOCK \|\| err == WSAETIMEDOUT) {
244	// Timeout, just try again.
245	continue;
246	}
247	LogErr() << "TCP receive error: " << get_socket_error_string(err)
248	<< ", trying to reconnect...";
249	#else
250	if (errno == EAGAIN \|\| errno == ETIMEDOUT) {	13✔
251	// Timeout, just try again.
252	continue;	13✔
253	}
254	LogErr() << "TCP receive error: " << strerror(errno) << ", trying to reconnect...";	×
255	#endif
256	std::this_thread::sleep_for(std::chrono::seconds(1));	×
257	setup_port();	×
258	continue;	×
259	}
260
261	_mavlink_receiver->set_new_datagram(buffer, static_cast<int>(recv_len));	37✔
262
263	auto parse_result = _mavlink_receiver->parse_message();	37✔
264	while (parse_result != MavlinkReceiver::ParseResult::NoneAvailable) {	74✔
265	receive_message(parse_result, _mavlink_receiver->get_last_message(), this);	37✔
266	parse_result = _mavlink_receiver->parse_message();	37✔
267	}
268
269	// Also parse with libmav if available
270	if (_libmav_receiver) {	37✔
271	_libmav_receiver->set_new_datagram(buffer, static_cast<int>(recv_len));	37✔
272
273	while (_libmav_receiver->parse_message()) {	74✔
274	receive_libmav_message(_libmav_receiver->get_last_message(), this);	37✔
275	}
276	}
277	}
278	}	4✔
279
280	} // namespace mavsdk

mavlink / MAVSDK / 20560534171

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous