10059975553

Committed 23 Jul 2024 01:33PM UTC coverage: 37.622% (+0.09%) from 37.537%

Build # 10059975553

Build Type

Pull #2357

github

Committed by

web-flow

Commit Message

Merge 0b6ca6e1b into f1aff89a6

Pull Request Pull Request #2357: core: Correctly close sockets

Run Details

43 of 46 new or added lines in 5 files covered. (93.48%)

17 existing lines in 2 files now uncovered.

11469 of 30485 relevant lines covered (37.62%)

260.35 hits per line

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

73.33

/src/mavsdk/core/tcp_client_connection.cpp

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

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

#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

#ifndef WINDOWS
#define GET_ERROR(_x) strerror(_x)
#else
#define GET_ERROR(_x) WSAGetLastError()
#endif

namespace mavsdk {

/* change to remote_ip and remote_port */
TcpClientConnection::TcpClientConnection(
    Connection::ReceiverCallback receiver_callback,
    std::string remote_ip,
    int remote_port,
    ForwardingOption forwarding_option) :
    Connection(std::move(receiver_callback), 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;
    }

    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: %d", WSAGetLastError();
        _is_ok = false;
        return ConnectionResult::SocketError;
    }
#endif

    _socket_fd.reset(socket(AF_INET, SOCK_STREAM, 0));

    if (_socket_fd.empty()) {
        LogErr() << "socket error" << GET_ERROR(errno);
        _is_ok = false;
        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";
        _is_ok = false;
        return ConnectionResult::SocketConnectionError;
    }

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

    if (connect(_socket_fd.get(), reinterpret_cast<sockaddr*>(&remote_addr), sizeof(struct sockaddr_in)) <
        0) {
        LogErr() << "connect error: " << GET_ERROR(errno);
        _is_ok = false;
        return ConnectionResult::SocketConnectionError;
    }

    _is_ok = true;
    return ConnectionResult::Success;
}

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

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

    _socket_fd.close();

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

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

bool TcpClientConnection::send_message(const mavlink_message_t& message)
{
    if (!_is_ok) {
        return false;
    }

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

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

    struct sockaddr_in dest_addr {};
    dest_addr.sin_family = AF_INET;

    inet_pton(AF_INET, _remote_ip.c_str(), &dest_addr.sin_addr.s_addr);

    dest_addr.sin_port = htons(_remote_port_number);

    uint8_t buffer[MAVLINK_MAX_PACKET_LEN];
    uint16_t buffer_len = mavlink_msg_to_send_buffer(buffer, &message);

    // TODO: remove this assert again
    assert(buffer_len <= MAVLINK_MAX_PACKET_LEN);

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

    const auto send_len = sendto(
        _socket_fd.get(),
        reinterpret_cast<char*>(buffer),
        buffer_len,
        flags,
        reinterpret_cast<const sockaddr*>(&dest_addr),
        sizeof(dest_addr));

    if (send_len != buffer_len) {
        LogErr() << "sendto failure: " << GET_ERROR(errno);
        _is_ok = false;
        return false;
    }
    return true;
}

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

    while (!_should_exit) {
        if (!_is_ok) {
            LogErr() << "TCP receive error, trying to reconnect...";
            std::this_thread::sleep_for(std::chrono::seconds(1));
            setup_port();
        }

        const auto recv_len = recv(_socket_fd.get(), buffer, sizeof(buffer), 0);

        if (recv_len == 0) {
            // This can happen when shutdown is called on the socket,
            // therefore we check _should_exit again.
            _is_ok = false;
            continue;
        }

        if (recv_len < 0) {
            // This happens on desctruction when close(_socket_fd.get()) is called,
            // therefore be quiet.
            // LogErr() << "recvfrom error: " << GET_ERROR(errno);
            // Something went wrong, we should try to re-connect in next iteration.
            _is_ok = false;
            continue;
        }

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

        // Parse all mavlink messages in one data packet. Once exhausted, we'll exit while.
        while (_mavlink_receiver->parse_message()) {
            receive_message(_mavlink_receiver->get_last_message(), this);
        }
    }
}

} // namespace mavsdk

1	#include "tcp_client_connection.h"
2	#include "log.h"
3
4	#include <cassert>
5	#include <utility>
6	#include <thread>
7
8	#ifdef WINDOWS
9	#ifndef MINGW
10	#pragma comment(lib, "Ws2_32.lib") // Without this, Ws2_32.lib is not included in static library.
11	#endif
12	#else
13	#include <netinet/in.h>
14	#include <sys/socket.h>
15	#include <arpa/inet.h>
16	#include <errno.h>
17	#include <netdb.h>
18	#endif
19
20	#ifndef WINDOWS
21	#define GET_ERROR(_x) strerror(_x)
22	#else
23	#define GET_ERROR(_x) WSAGetLastError()
24	#endif
25
26	namespace mavsdk {
27
28	/* change to remote_ip and remote_port */
29	TcpClientConnection::TcpClientConnection(	1✔
30	Connection::ReceiverCallback receiver_callback,
31	std::string remote_ip,
32	int remote_port,
33	ForwardingOption forwarding_option) :	1✔
34	Connection(std::move(receiver_callback), forwarding_option),	1✔
35	_remote_ip(std::move(remote_ip)),	1✔
36	_remote_port_number(remote_port),
37	_should_exit(false)	2✔
38	{}	1✔
39
40	TcpClientConnection::~TcpClientConnection()	1✔
41	{
42	// If no one explicitly called stop before, we should at least do it.
43	stop();	1✔
44	}	1✔
45
46	ConnectionResult TcpClientConnection::start()	1✔
47	{
48	if (!start_mavlink_receiver()) {	1✔
49	return ConnectionResult::ConnectionsExhausted;	×
50	}
51
52	ConnectionResult ret = setup_port();	1✔
53	if (ret != ConnectionResult::Success) {	1✔
54	return ret;	×
55	}
56
57	start_recv_thread();	1✔
58
59	return ConnectionResult::Success;	1✔
60	}
61
62	ConnectionResult TcpClientConnection::setup_port()	1✔
63	{
64	#ifdef WINDOWS
65	WSADATA wsa;
66	if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) {
67	LogErr() << "Error: Winsock failed, error: %d", WSAGetLastError();
68	_is_ok = false;
69	return ConnectionResult::SocketError;
70	}
71	#endif
72
73	_socket_fd.reset(socket(AF_INET, SOCK_STREAM, 0));	1✔
74
75	if (_socket_fd.empty()) {	1✔
76	LogErr() << "socket error" << GET_ERROR(errno);	×
77	_is_ok = false;	×
78	return ConnectionResult::SocketError;	×
79	}
80
81	struct sockaddr_in remote_addr {};	1✔
82	remote_addr.sin_family = AF_INET;	1✔
83	remote_addr.sin_port = htons(_remote_port_number);	1✔
84
85	struct hostent* hp;
86	hp = gethostbyname(_remote_ip.c_str());	1✔
87	if (hp == nullptr) {	1✔
88	LogErr() << "Could not get host by name";	×
89	_is_ok = false;	×
90	return ConnectionResult::SocketConnectionError;	×
91	}
92
93	memcpy(&remote_addr.sin_addr, hp->h_addr, hp->h_length);	1✔
94
95	if (connect(_socket_fd.get(), reinterpret_cast<sockaddr*>(&remote_addr), sizeof(struct sockaddr_in)) <	1✔
96	0) {
97	LogErr() << "connect error: " << GET_ERROR(errno);	×
98	_is_ok = false;	×
99	return ConnectionResult::SocketConnectionError;	×
100	}
101
102	_is_ok = true;	1✔
103	return ConnectionResult::Success;	1✔
104	}
105
106	void TcpClientConnection::start_recv_thread()	1✔
107	{
108	_recv_thread = std::make_unique<std::thread>(&TcpClientConnection::receive, this);	1✔
109	}	1✔
110
111	ConnectionResult TcpClientConnection::stop()	1✔
112	{
113	_should_exit = true;	1✔
114
115	_socket_fd.close();	1✔
116
117	if (_recv_thread) {	1✔
118	_recv_thread->join();	1✔
119	_recv_thread.reset();	1✔
120	}
121
122	// We need to stop this after stopping the receive thread, otherwise
123	// it can happen that we interfere with the parsing of a message.
124	stop_mavlink_receiver();	1✔
125
126	return ConnectionResult::Success;	1✔
127	}
128
129	bool TcpClientConnection::send_message(const mavlink_message_t& message)	5✔
130	{
131	if (!_is_ok) {	5✔
UNCOV 132	return false;	×
133	}
134
135	if (_remote_ip.empty()) {	5✔
UNCOV 136	LogErr() << "Remote IP unknown";	×
UNCOV 137	return false;	×
138	}
139
140	if (_remote_port_number == 0) {	5✔
UNCOV 141	LogErr() << "Remote port unknown";	×
UNCOV 142	return false;	×
143	}
144
145	struct sockaddr_in dest_addr {};	5✔
146	dest_addr.sin_family = AF_INET;	5✔
147
148	inet_pton(AF_INET, _remote_ip.c_str(), &dest_addr.sin_addr.s_addr);	5✔
149
150	dest_addr.sin_port = htons(_remote_port_number);	5✔
151
152	uint8_t buffer[MAVLINK_MAX_PACKET_LEN];	5✔
153	uint16_t buffer_len = mavlink_msg_to_send_buffer(buffer, &message);	5✔
154
155	// TODO: remove this assert again
156	assert(buffer_len <= MAVLINK_MAX_PACKET_LEN);	5✔
157
158	#if !defined(MSG_NOSIGNAL)
159	auto flags = 0;
160	#else
161	auto flags = MSG_NOSIGNAL;	5✔
162	#endif
163
164	const auto send_len = sendto(	5✔
165	_socket_fd.get(),
166	reinterpret_cast<char*>(buffer),
167	buffer_len,
168	flags,
169	reinterpret_cast<const sockaddr*>(&dest_addr),
170	sizeof(dest_addr));
171
172	if (send_len != buffer_len) {	5✔
UNCOV 173	LogErr() << "sendto failure: " << GET_ERROR(errno);	×
UNCOV 174	_is_ok = false;	×
175	return false;	×
176	}
177	return true;	5✔
178	}
179
180	void TcpClientConnection::receive()	1✔
181	{
182	// Enough for MTU 1500 bytes.
183	char buffer[2048];	1✔
184
185	while (!_should_exit) {	4✔
186	if (!_is_ok) {	3✔
UNCOV 187	LogErr() << "TCP receive error, trying to reconnect...";	×
UNCOV 188	std::this_thread::sleep_for(std::chrono::seconds(1));	×
189	setup_port();	×
190	}
191
192	const auto recv_len = recv(_socket_fd.get(), buffer, sizeof(buffer), 0);	3✔
193
194	if (recv_len == 0) {	3✔
195	// This can happen when shutdown is called on the socket,
196	// therefore we check _should_exit again.
UNCOV 197	_is_ok = false;	×
UNCOV 198	continue;	×
199	}
200
201	if (recv_len < 0) {	3✔
202	// This happens on desctruction when close(_socket_fd.get()) is called,
203	// therefore be quiet.
204	// LogErr() << "recvfrom error: " << GET_ERROR(errno);
205	// Something went wrong, we should try to re-connect in next iteration.
206	_is_ok = false;	1✔
207	continue;	1✔
208	}
209
210	_mavlink_receiver->set_new_datagram(buffer, static_cast<int>(recv_len));	2✔
211
212	// Parse all mavlink messages in one data packet. Once exhausted, we'll exit while.
213	while (_mavlink_receiver->parse_message()) {	4✔
214	receive_message(_mavlink_receiver->get_last_message(), this);	2✔
215	}
216	}
217	}	1✔
218
219	} // namespace mavsdk

mavlink / MAVSDK / 10059975553

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