19215666663

Committed 09 Nov 2025 10:52PM UTC coverage: 48.274% (-0.06%) from 48.329%

Build # 19215666663

Build Type

Pull #2708

github

Committed by

web-flow

Commit Message

Merge 99e578bbf into 0ca475a61

Pull Request Pull Request #2708: Fix connection usage for mavsdk_server

Run Details

17623 of 36506 relevant lines covered (48.27%)

462.67 hits per line

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

73.39

/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;
    }

    // Get socket fd with mutex protection, then release mutex before blocking send
    SocketHolder::DescriptorType socket_fd;
    {
        std::lock_guard<std::mutex> lock(_mutex);
        if (_socket_fd.empty()) {
            result.first = false;
            result.second = "Not connected";
            return result;
        }
        socket_fd = _socket_fd.get();
    }

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

    const auto send_len = send(socket_fd, 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));

        while (_mavlink_receiver->parse_message()) {
            receive_message(_mavlink_receiver->get_last_message(), this);
        }

        // 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)	28✔
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);	28✔
154
155	return send_raw_bytes(reinterpret_cast<const char*>(buffer), buffer_len);	28✔
156	}
157
158	std::pair<bool, std::string> TcpClientConnection::send_raw_bytes(const char* bytes, size_t length)	28✔
159	{
160	std::pair<bool, std::string> result;	28✔
161
162	if (_remote_ip.empty()) {	28✔
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) {	28✔
170	result.first = false;	×
171	result.second = "Remote port unknown";	×
172	LogErr() << result.second;	×
173	return result;	×
174	}
175
176	// Get socket fd with mutex protection, then release mutex before blocking send
177	SocketHolder::DescriptorType socket_fd;
178	{
179	std::lock_guard<std::mutex> lock(_mutex);	28✔
180	if (_socket_fd.empty()) {	28✔
181	result.first = false;	×
182	result.second = "Not connected";	×
183	return result;	×
184	}
185	socket_fd = _socket_fd.get();	28✔
186	}	28✔
187
188	#if !defined(MSG_NOSIGNAL)
189	auto flags = 0;
190	#else
191	auto flags = MSG_NOSIGNAL;	28✔
192	#endif
193
194	const auto send_len = send(socket_fd, bytes, length, flags);	28✔
195
196	if (send_len != static_cast<std::remove_cv_t<decltype(send_len)>>(length)) {	28✔
197	std::stringstream ss;	×
198	ss << "Send failure: " << strerror(errno);	×
199	LogErr() << ss.str();	×
200	result.first = false;	×
201	result.second = ss.str();	×
202	return result;	×
203	}	×
204
205	result.first = true;	28✔
206	return result;	28✔
207	}
208
209	void TcpClientConnection::receive()	4✔
210	{
211	// Enough for MTU 1500 bytes.
212	char buffer[2048];
213
214	while (!_should_exit) {	51✔
215	// Get socket fd with mutex protection, then release mutex before blocking recv
216	SocketHolder::DescriptorType socket_fd;
217	{
218	std::lock_guard<std::mutex> lock(_mutex);	47✔
219	if (_socket_fd.empty()) {	47✔
220	// Socket not connected, need to reconnect
221	socket_fd = SocketHolder::invalid_socket_fd;	×
222	} else {
223	socket_fd = _socket_fd.get();	47✔
224	}
225	}	47✔
226
227	if (socket_fd == SocketHolder::invalid_socket_fd) {	47✔
228	std::this_thread::sleep_for(std::chrono::seconds(1));	×
229	setup_port();	×
230	continue;	×
231	}
232
233	const auto recv_len = recv(socket_fd, buffer, sizeof(buffer), 0);	47✔
234
235	if (recv_len == 0) {	47✔
236	// Connection closed, just try again.
237	LogInfo() << "TCP connection closed, trying to reconnect...";	3✔
238	std::this_thread::sleep_for(std::chrono::seconds(1));	3✔
239	setup_port();	3✔
240	continue;	3✔
241	}
242
243	if (recv_len < 0) {	44✔
244	#ifdef WINDOWS
245	int err = WSAGetLastError();
246	if (err == WSAEWOULDBLOCK \|\| err == WSAETIMEDOUT) {
247	// Timeout, just try again.
248	continue;
249	}
250	LogErr() << "TCP receive error: " << get_socket_error_string(err)
251	<< ", trying to reconnect...";
252	#else
253	if (errno == EAGAIN \|\| errno == ETIMEDOUT) {	14✔
254	// Timeout, just try again.
255	continue;	14✔
256	}
257	LogErr() << "TCP receive error: " << strerror(errno) << ", trying to reconnect...";	×
258	#endif
259	std::this_thread::sleep_for(std::chrono::seconds(1));	×
260	setup_port();	×
261	continue;	×
262	}
263
264	_mavlink_receiver->set_new_datagram(buffer, static_cast<int>(recv_len));	30✔
265
266	while (_mavlink_receiver->parse_message()) {	60✔
267	receive_message(_mavlink_receiver->get_last_message(), this);	30✔
268	}
269
270	// Also parse with libmav if available
271	if (_libmav_receiver) {	30✔
272	_libmav_receiver->set_new_datagram(buffer, static_cast<int>(recv_len));	30✔
273
274	while (_libmav_receiver->parse_message()) {	60✔
275	receive_libmav_message(_libmav_receiver->get_last_message(), this);	30✔
276	}
277	}
278	}
279	}	4✔
280
281	} // namespace mavsdk

mavlink / MAVSDK / 19215666663

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