29462275360

Committed 16 Jul 2026 12:44AM UTC coverage: 50.438%. First build

Build # 29462275360

Build Type

Pull #2936

github

Committed by

web-flow

Commit Message

Merge a1505e5d2 into 23fd28534

Pull Request Pull Request #2936: Backport various bugfixes to v3

Coverage Stats

161 of 239 new or added lines in 22 files covered. (67.36%)

19309 of 38283 relevant lines covered (50.44%)

671.19 hits per line

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

0.0

/src/mavsdk/plugins/log_streaming/log_streaming_impl.cpp

#include <future>

#include "log_streaming_impl.h"
#include "log_streaming_backend_px4.h"
#include "log_streaming_backend_ardupilot.h"
#include "plugins/log_streaming/log_streaming.h"
#include "callback_list.tpp"
#include "mavsdk_export.h"
#include "base64.h"

namespace mavsdk {

template class MAVSDK_TEMPL_INST CallbackList<LogStreaming::LogStreamingRaw>;

LogStreamingImpl::LogStreamingImpl(System& system) : PluginImplBase(system)
{
    _system_impl->register_plugin(this);
}

LogStreamingImpl::LogStreamingImpl(std::shared_ptr<System> system) :
    PluginImplBase(std::move(system))
{
    _system_impl->register_plugin(this);
}

LogStreamingImpl::~LogStreamingImpl()
{
    _system_impl->unregister_plugin(this);
}

void LogStreamingImpl::init()
{
    if (const char* env_p = std::getenv("MAVSDK_LOG_STREAMING_DEBUGGING")) {
        if (std::string(env_p) == "1") {
            LogDebug() << "Log streaming debugging is on.";
            _debugging = true;
        }
    }
}

void LogStreamingImpl::deinit()
{
    std::unique_ptr<LogStreamingBackend> backend;
    {
        std::lock_guard<std::mutex> lock(_mutex);

        // Cancel any pending autopilot type polling
        if (_check_autopilot_cookie) {
            _system_impl->remove_call_every(_check_autopilot_cookie);
            _check_autopilot_cookie = {};
        }
        _start_callback = nullptr;

        backend = std::move(_backend);
    }

    // Deinit (and destroy) the backend outside of _mutex: its blocking unregister can wait
    // for an in-flight message callback to finish, and that callback may be in
    // process_data() waiting for _mutex.
    if (backend) {
        backend->deinit();
    }
}

void LogStreamingImpl::enable() {}

void LogStreamingImpl::disable()
{
    std::unique_ptr<LogStreamingBackend> backend;
    {
        std::lock_guard<std::mutex> lock(_mutex);
        backend = std::move(_backend);
    }

    // Deinit (and destroy) the backend outside of _mutex, see deinit() above.
    if (backend) {
        backend->deinit();
    }
}

bool LogStreamingImpl::maybe_create_backend()
{
    // Already have a backend
    if (_backend) {
        return true;
    }

    auto autopilot = _system_impl->effective_autopilot();

    // Don't create backend yet if autopilot type is unknown
    if (autopilot == Autopilot::Unknown) {
        if (_debugging) {
            LogDebug() << "Autopilot type unknown, cannot create backend yet";
        }
        return false;
    }

    if (autopilot == Autopilot::ArduPilot) {
        if (_debugging) {
            LogDebug() << "Creating ArduPilot log streaming backend";
        }
        _backend = std::make_unique<LogStreamingBackendArdupilot>();
    } else {
        if (_debugging) {
            LogDebug() << "Creating PX4 log streaming backend";
        }
        _backend = std::make_unique<LogStreamingBackendPx4>();
    }

    _backend->set_debugging(_debugging);
    _backend->init(_system_impl.get());
    _backend->set_data_callback([this](const std::vector<uint8_t>& data) { process_data(data); });
    return true;
}

void LogStreamingImpl::process_data(const std::vector<uint8_t>& data)
{
    std::lock_guard<std::mutex> lock(_mutex);

    if (_debugging) {
        LogDebug() << "Processing log data with size " << data.size();
    }

    // Convert to base64
    LogStreaming::LogStreamingRaw part;
    std::vector<uint8_t> data_copy = data;
    part.data_base64 = base64_encode(data_copy);

    // Let's pass it to the user.
    if (!_subscription_callbacks.empty()) {
        _subscription_callbacks.queue(
            part, [this](const auto& func) { _system_impl->call_user_callback(func); });
    }
}

void LogStreamingImpl::start_log_streaming_async(const LogStreaming::ResultCallback& callback)
{
    std::lock_guard<std::mutex> lock(_mutex);

    // Try to create backend if not yet created
    if (maybe_create_backend()) {
        _backend->start_log_streaming_async(callback);
        return;
    }

    // Autopilot type unknown - wait for it with polling.
    // Use timeout_s() for the total wait time, polling every 0.1s.
    const float poll_interval_s = 0.1f;
    const unsigned max_polls = static_cast<unsigned>(_system_impl->timeout_s() / poll_interval_s);

    _start_callback = callback;
    _autopilot_poll_count = 0;

    _check_autopilot_cookie = _system_impl->add_call_every(
        [this, max_polls]() {
            std::lock_guard<std::mutex> lock2(_mutex);

            if (maybe_create_backend()) {
                // Success - stop polling and start streaming
                _system_impl->remove_call_every(_check_autopilot_cookie);
                _backend->start_log_streaming_async(_start_callback);
                _start_callback = nullptr;
                return;
            }

            _autopilot_poll_count++;
            if (_autopilot_poll_count >= max_polls) {
                // Timeout - stop polling and report failure
                _system_impl->remove_call_every(_check_autopilot_cookie);
                if (_start_callback) {
                    auto cb = _start_callback;
                    _start_callback = nullptr;
                    _system_impl->call_user_callback(
                        [cb]() { cb(LogStreaming::Result::NoSystem); });
                }
            }
        },
        poll_interval_s);
}

LogStreaming::Result LogStreamingImpl::start_log_streaming()
{
    auto prom = std::promise<LogStreaming::Result>{};
    auto fut = prom.get_future();

    start_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });

    return fut.get();
}

void LogStreamingImpl::stop_log_streaming_async(const LogStreaming::ResultCallback& callback)
{
    std::lock_guard<std::mutex> lock(_mutex);

    if (!_backend) {
        if (callback) {
            _system_impl->call_user_callback(
                [callback]() { callback(LogStreaming::Result::NoSystem); });
        }
        return;
    }

    _backend->stop_log_streaming_async(callback);
}

LogStreaming::Result LogStreamingImpl::stop_log_streaming()
{
    auto prom = std::promise<LogStreaming::Result>{};
    auto fut = prom.get_future();

    stop_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });

    return fut.get();
}

LogStreaming::LogStreamingRawHandle
LogStreamingImpl::subscribe_log_streaming_raw(const LogStreaming::LogStreamingRawCallback& callback)
{
    std::lock_guard<std::mutex> lock(_mutex);
    auto handle = _subscription_callbacks.subscribe(callback);

    return handle;
}

void LogStreamingImpl::unsubscribe_log_streaming_raw(LogStreaming::LogStreamingRawHandle handle)
{
    std::lock_guard<std::mutex> lock(_mutex);
    _subscription_callbacks.unsubscribe(handle);
}

} // namespace mavsdk

1	#include <future>
2
3	#include "log_streaming_impl.h"
4	#include "log_streaming_backend_px4.h"
5	#include "log_streaming_backend_ardupilot.h"
6	#include "plugins/log_streaming/log_streaming.h"
7	#include "callback_list.tpp"
8	#include "mavsdk_export.h"
9	#include "base64.h"
10
11	namespace mavsdk {
12
13	template class MAVSDK_TEMPL_INST CallbackList<LogStreaming::LogStreamingRaw>;
14
15	LogStreamingImpl::LogStreamingImpl(System& system) : PluginImplBase(system)	×
16	{
17	_system_impl->register_plugin(this);	×
18	}	×
19
20	LogStreamingImpl::LogStreamingImpl(std::shared_ptr<System> system) :	×
21	PluginImplBase(std::move(system))	×
22	{
23	_system_impl->register_plugin(this);	×
24	}	×
25
26	LogStreamingImpl::~LogStreamingImpl()	×
27	{
28	_system_impl->unregister_plugin(this);	×
29	}	×
30
31	void LogStreamingImpl::init()	×
32	{
33	if (const char* env_p = std::getenv("MAVSDK_LOG_STREAMING_DEBUGGING")) {	×
34	if (std::string(env_p) == "1") {	×
35	LogDebug() << "Log streaming debugging is on.";	×
36	_debugging = true;	×
37	}
38	}
39	}	×
40
41	void LogStreamingImpl::deinit()	×
42	{
NEW 43	std::unique_ptr<LogStreamingBackend> backend;	×
44	{
NEW 45	std::lock_guard<std::mutex> lock(_mutex);	×
46
47	// Cancel any pending autopilot type polling
NEW 48	if (_check_autopilot_cookie) {	×
NEW 49	_system_impl->remove_call_every(_check_autopilot_cookie);	×
NEW 50	_check_autopilot_cookie = {};	×
51	}
NEW 52	_start_callback = nullptr;	×
53
NEW 54	backend = std::move(_backend);	×
55	}	×
56
57	// Deinit (and destroy) the backend outside of _mutex: its blocking unregister can wait
58	// for an in-flight message callback to finish, and that callback may be in
59	// process_data() waiting for _mutex.
NEW 60	if (backend) {	×
NEW 61	backend->deinit();	×
62	}
63	}	×
64
65	void LogStreamingImpl::enable() {}	×
66
67	void LogStreamingImpl::disable()	×
68	{
NEW 69	std::unique_ptr<LogStreamingBackend> backend;	×
70	{
NEW 71	std::lock_guard<std::mutex> lock(_mutex);	×
NEW 72	backend = std::move(_backend);	×
NEW 73	}	×
74
75	// Deinit (and destroy) the backend outside of _mutex, see deinit() above.
NEW 76	if (backend) {	×
NEW 77	backend->deinit();	×
78	}
79	}	×
80
81	bool LogStreamingImpl::maybe_create_backend()	×
82	{
83	// Already have a backend
84	if (_backend) {	×
85	return true;	×
86	}
87
88	auto autopilot = _system_impl->effective_autopilot();	×
89
90	// Don't create backend yet if autopilot type is unknown
91	if (autopilot == Autopilot::Unknown) {	×
92	if (_debugging) {	×
93	LogDebug() << "Autopilot type unknown, cannot create backend yet";	×
94	}
95	return false;	×
96	}
97
98	if (autopilot == Autopilot::ArduPilot) {	×
99	if (_debugging) {	×
100	LogDebug() << "Creating ArduPilot log streaming backend";	×
101	}
102	_backend = std::make_unique<LogStreamingBackendArdupilot>();	×
103	} else {
104	if (_debugging) {	×
105	LogDebug() << "Creating PX4 log streaming backend";	×
106	}
107	_backend = std::make_unique<LogStreamingBackendPx4>();	×
108	}
109
110	_backend->set_debugging(_debugging);	×
111	_backend->init(_system_impl.get());	×
112	_backend->set_data_callback([this](const std::vector<uint8_t>& data) { process_data(data); });	×
113	return true;	×
114	}
115
116	void LogStreamingImpl::process_data(const std::vector<uint8_t>& data)	×
117	{
118	std::lock_guard<std::mutex> lock(_mutex);	×
119
120	if (_debugging) {	×
121	LogDebug() << "Processing log data with size " << data.size();	×
122	}
123
124	// Convert to base64
125	LogStreaming::LogStreamingRaw part;	×
126	std::vector<uint8_t> data_copy = data;	×
127	part.data_base64 = base64_encode(data_copy);	×
128
129	// Let's pass it to the user.
130	if (!_subscription_callbacks.empty()) {	×
131	_subscription_callbacks.queue(	×
132	part, [this](const auto& func) { _system_impl->call_user_callback(func); });	×
133	}
134	}	×
135
136	void LogStreamingImpl::start_log_streaming_async(const LogStreaming::ResultCallback& callback)	×
137	{
138	std::lock_guard<std::mutex> lock(_mutex);	×
139
140	// Try to create backend if not yet created
141	if (maybe_create_backend()) {	×
142	_backend->start_log_streaming_async(callback);	×
143	return;	×
144	}
145
146	// Autopilot type unknown - wait for it with polling.
147	// Use timeout_s() for the total wait time, polling every 0.1s.
148	const float poll_interval_s = 0.1f;	×
149	const unsigned max_polls = static_cast<unsigned>(_system_impl->timeout_s() / poll_interval_s);	×
150
151	_start_callback = callback;	×
152	_autopilot_poll_count = 0;	×
153
154	_check_autopilot_cookie = _system_impl->add_call_every(	×
155	[this, max_polls]() {	×
156	std::lock_guard<std::mutex> lock2(_mutex);	×
157
158	if (maybe_create_backend()) {	×
159	// Success - stop polling and start streaming
160	_system_impl->remove_call_every(_check_autopilot_cookie);	×
161	_backend->start_log_streaming_async(_start_callback);	×
162	_start_callback = nullptr;	×
163	return;	×
164	}
165
166	_autopilot_poll_count++;	×
167	if (_autopilot_poll_count >= max_polls) {	×
168	// Timeout - stop polling and report failure
169	_system_impl->remove_call_every(_check_autopilot_cookie);	×
170	if (_start_callback) {	×
171	auto cb = _start_callback;	×
172	_start_callback = nullptr;	×
173	_system_impl->call_user_callback(	×
174	[cb]() { cb(LogStreaming::Result::NoSystem); });
175	}	×
176	}
177	},	×
178	poll_interval_s);
179	}	×
180
181	LogStreaming::Result LogStreamingImpl::start_log_streaming()	×
182	{
183	auto prom = std::promise<LogStreaming::Result>{};	×
184	auto fut = prom.get_future();	×
185
186	start_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });	×
187
188	return fut.get();	×
189	}	×
190
191	void LogStreamingImpl::stop_log_streaming_async(const LogStreaming::ResultCallback& callback)	×
192	{
193	std::lock_guard<std::mutex> lock(_mutex);	×
194
195	if (!_backend) {	×
196	if (callback) {	×
197	_system_impl->call_user_callback(	×
198	[callback]() { callback(LogStreaming::Result::NoSystem); });
199	}
200	return;	×
201	}
202
203	_backend->stop_log_streaming_async(callback);	×
204	}	×
205
206	LogStreaming::Result LogStreamingImpl::stop_log_streaming()	×
207	{
208	auto prom = std::promise<LogStreaming::Result>{};	×
209	auto fut = prom.get_future();	×
210
211	stop_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });	×
212
213	return fut.get();	×
214	}	×
215
216	LogStreaming::LogStreamingRawHandle
217	LogStreamingImpl::subscribe_log_streaming_raw(const LogStreaming::LogStreamingRawCallback& callback)	×
218	{
219	std::lock_guard<std::mutex> lock(_mutex);	×
220	auto handle = _subscription_callbacks.subscribe(callback);	×
221
222	return handle;	×
223	}	×
224
225	void LogStreamingImpl::unsubscribe_log_streaming_raw(LogStreaming::LogStreamingRawHandle handle)	×
226	{
227	std::lock_guard<std::mutex> lock(_mutex);	×
228	_subscription_callbacks.unsubscribe(handle);	×
229	}	×
230
231	} // namespace mavsdk

mavlink / MAVSDK / 29462275360

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