Branch data Line data Source code
1 : : /*
2 : : * Copyright (C) 2026 by Thun Lu. All rights reserved.
3 : : * Author: Thun Lu <thun.lu@zohomail.cn>
4 : : * Repo: https://github.com/thun-res/vlink
5 : : * _ __ __ _ __
6 : : * | | / / / / (_) ____ / /__
7 : : * | | / / / / / / / __ \ / //_/
8 : : * | |/ / / /___ / / / / / / / ,<
9 : : * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10 : : *
11 : : * Licensed under the Apache License, Version 2.0 (the "License");
12 : : * you may not use this file except in compliance with the License.
13 : : * You may obtain a copy of the License at
14 : : *
15 : : * http://www.apache.org/licenses/LICENSE-2.0
16 : : *
17 : : * Unless required by applicable law or agreed to in writing, software
18 : : * distributed under the License is distributed on an "AS IS" BASIS,
19 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 : : * See the License for the specific language governing permissions and
21 : : * limitations under the License.
22 : : */
23 : :
24 : : #include "./extension/vcap_reader.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <cinttypes>
29 : : #include <filesystem>
30 : : #include <fstream>
31 : : #include <limits>
32 : : #include <memory>
33 : : #include <mutex>
34 : : #include <shared_mutex>
35 : : #include <string>
36 : : #include <string_view>
37 : : #include <unordered_map>
38 : : #include <utility>
39 : : #include <vector>
40 : :
41 : : #include "./base/condition_variable.h"
42 : : #include "./base/elapsed_timer.h"
43 : : #include "./base/helpers.h"
44 : : #include "./base/logger.h"
45 : : #include "./version.h"
46 : :
47 : : // json
48 : : #include <nlohmann/json.hpp>
49 : :
50 : : // mcap
51 : : #include "./private/mcap_import.h"
52 : :
53 : : namespace vlink {
54 : :
55 : : [[maybe_unused]] static constexpr size_t kMaxTaskSize = 50000U;
56 : :
57 : : // VCAPReader::Impl
58 : : struct VCAPReader::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
59 : : std::atomic<BagReader::Status> status{VCAPReader::kStopped};
60 : : std::atomic_bool stop_flag{false};
61 : : std::atomic_bool pause_flag{false};
62 : : std::atomic_bool pause_next_flag{false};
63 : : std::atomic_bool jump_flag{false};
64 : : std::atomic<int64_t> pause_elapsed{0};
65 : : std::atomic<int64_t> offset_elapsed{0};
66 : : std::atomic<int64_t> real_elapsed{0};
67 : : std::atomic<int64_t> extra_elapsed{0};
68 : : std::atomic<int64_t> begin_time{0};
69 : : std::atomic<double> rate{1.0};
70 : : std::atomic<int> times{1};
71 : : std::atomic_bool is_pending{false};
72 : : std::atomic<int> split_index{0};
73 : :
74 : : bool read_only{false};
75 : : bool try_to_fix{false};
76 : : bool enable_compress{false};
77 : :
78 : : std::string path;
79 : : BagReader::Info info;
80 : : std::vector<BagReader::Info::UrlMeta> raw_url_metas;
81 : : std::mutex mtx;
82 : : ConditionVariable cv;
83 : :
84 : : BagReader::Config config;
85 : : std::mutex config_mtx;
86 : : std::shared_mutex time_mtx;
87 : :
88 : : ElapsedTimer elapsed_timer{ElapsedTimer::kMicro};
89 : : ElapsedTimer pause_elapsed_timer{ElapsedTimer::kMicro};
90 : : ElapsedTimer offset_timer{ElapsedTimer::kMicro};
91 : : ElapsedTimer real_timer{ElapsedTimer::kMicro};
92 : :
93 : : BagReader::StatusCallback status_callback;
94 : : BagReader::ReadyCallback ready_callback;
95 : : BagReader::FinishCallback finish_callback;
96 : :
97 : : int64_t total_start_timestamp_ns{-1};
98 : : bool total_has_completed{false};
99 : :
100 : : // mcap
101 : :
102 : : // WrapperFile
103 : : struct WrapperFile final {
104 : : std::string path;
105 : : std::unique_ptr<mcap::McapReader> reader;
106 : : std::unique_ptr<mcap::LinearMessageView> msg_view;
107 : : std::optional<mcap::LinearMessageView::Iterator> msg_view_begin;
108 : : std::optional<mcap::LinearMessageView::Iterator> msg_view_end;
109 : : int index{0};
110 : : int64_t start_timestamp_ns{0};
111 : : int64_t begin{0};
112 : : int64_t end{0};
113 : : std::unordered_map<std::string, int> url_to_id_map;
114 : : std::unordered_map<int, std::string> id_to_url_map;
115 : : std::unordered_map<int, ActionType> channel_action_map;
116 : : bool has_idx_elapsed{false};
117 : : bool has_idx_url{false};
118 : : bool has_schema{false};
119 : : bool has_completed{false};
120 : : bool is_channel_broken{false};
121 : :
122 : 88 : WrapperFile() {
123 [ + - ]: 88 : url_to_id_map.reserve(128);
124 [ + - ]: 88 : id_to_url_map.reserve(128);
125 [ + - ]: 88 : channel_action_map.reserve(128);
126 : 88 : }
127 : : };
128 : :
129 : : std::vector<WrapperFile> file_list;
130 : :
131 : : std::unique_ptr<mcap::LinearMessageView> cursor_msg_view;
132 : : std::optional<mcap::LinearMessageView::Iterator> cursor_iter;
133 : : std::optional<mcap::LinearMessageView::Iterator> cursor_iter_end;
134 : : int cursor_file_index{0};
135 : : int64_t cursor_begin_us{0};
136 : : int64_t cursor_end_us{0};
137 : : bool cursor_need_advance{false};
138 : : bool cursor_read_error{false};
139 : : BagReader::Config cursor_config;
140 : : };
141 : :
142 : : // VCAPReader
143 : 67 : VCAPReader::VCAPReader(const std::string& path, bool read_only, bool try_to_fix)
144 [ + - ]: 67 : : BagReader(path, read_only, try_to_fix), impl_{std::make_unique<Impl>()} {
145 [ + - + - ]: 67 : set_name("VCAPReader");
146 : :
147 [ + - + - ]: 67 : url_ser_map().reserve(128);
148 [ + - + - ]: 67 : url_schema_type_map().reserve(128);
149 : :
150 : 67 : impl_->read_only = read_only;
151 : 67 : impl_->try_to_fix = try_to_fix;
152 : :
153 [ + + ]: 67 : open(path);
154 : 74 : }
155 : :
156 : 60 : VCAPReader::~VCAPReader() {
157 [ + + ]: 60 : if (!impl_->stop_flag.load(std::memory_order_relaxed)) {
158 : 59 : do_stop();
159 : : }
160 : :
161 : 60 : quit(true);
162 : :
163 : 60 : impl_->cv.notify_one();
164 : :
165 : 60 : wait_for_quit();
166 : :
167 : 60 : detach_plugin();
168 : :
169 : 60 : close();
170 : 60 : }
171 : :
172 : 3 : void VCAPReader::bind_plugin_interface(const std::shared_ptr<BagPluginInterface>& plugin_interface) {
173 : 3 : BagReader::bind_plugin_interface(plugin_interface);
174 : 3 : impl_->info.url_metas = impl_->raw_url_metas;
175 : 3 : process_url_metas(impl_->info.url_metas);
176 : 3 : rebuild_url_meta_lookup(impl_->info.url_metas);
177 : 3 : }
178 : :
179 : 2 : void VCAPReader::register_status_callback(StatusCallback&& status_callback) {
180 : 2 : impl_->status_callback = std::move(status_callback);
181 : 2 : }
182 : :
183 : 2 : void VCAPReader::register_ready_callback(ReadyCallback&& ready_callback) {
184 : 2 : impl_->ready_callback = std::move(ready_callback);
185 : 2 : }
186 : :
187 : 4 : void VCAPReader::register_finish_callback(FinishCallback&& finish_callback) {
188 : 4 : impl_->finish_callback = std::move(finish_callback);
189 : 4 : }
190 : :
191 : 5 : void VCAPReader::register_output_callback(OutputCallback&& output_callback) {
192 : 5 : BagReader::register_output_callback(std::move(output_callback));
193 : 5 : }
194 : :
195 : 9 : void VCAPReader::play(const Config& config) {
196 [ + - - + ]: 9 : if VUNLIKELY (is_busy()) {
197 [ # # # # ]: 0 : VLOG_W("VCAPReader: Is busy.");
198 : : // return;
199 : : }
200 : :
201 [ + + ]: 9 : if (config.skip_blank) {
202 : 1 : impl_->begin_time.store(std::max(config.begin_time, impl_->info.blank_duration), std::memory_order_relaxed);
203 : : } else {
204 : 8 : impl_->begin_time.store(config.begin_time, std::memory_order_relaxed);
205 : : }
206 : :
207 [ + + ]: 9 : if (config.rate <= 0) {
208 : 1 : impl_->rate.store(1, std::memory_order_relaxed);
209 : : } else {
210 : 8 : impl_->rate.store(config.rate, std::memory_order_relaxed);
211 : : }
212 : :
213 : 9 : impl_->times.store(config.times, std::memory_order_relaxed);
214 : :
215 : 18 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
216 : 9 : impl_->is_pending.store(true, std::memory_order_relaxed);
217 : :
218 : : {
219 [ + - ]: 9 : std::unique_lock lock(impl_->mtx);
220 : 9 : impl_->stop_flag.store(false, std::memory_order_relaxed);
221 : 9 : impl_->pause_flag.store(false, std::memory_order_relaxed);
222 : 9 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
223 : 9 : impl_->jump_flag.store(false, std::memory_order_relaxed);
224 : 9 : }
225 : :
226 : 9 : Config config_snapshot;
227 : :
228 : : {
229 [ + - ]: 9 : std::unique_lock lock(impl_->config_mtx);
230 [ + - ]: 9 : impl_->config = config;
231 [ + - ]: 9 : config_snapshot = impl_->config;
232 : 9 : }
233 : :
234 [ + - + - : 18 : post_task([this, config_snapshot]() { read(config_snapshot); });
+ - ]
235 : 9 : }
236 : :
237 : 2 : void VCAPReader::stop() { do_stop(); }
238 : :
239 : 2 : void VCAPReader::pause() {
240 : : {
241 [ + - ]: 2 : std::unique_lock lock(impl_->mtx);
242 : 2 : impl_->pause_flag.store(true, std::memory_order_relaxed);
243 : 2 : }
244 : :
245 : 2 : impl_->cv.notify_one();
246 : 2 : }
247 : :
248 : 2 : void VCAPReader::resume() {
249 : : {
250 [ + - ]: 2 : std::unique_lock lock(impl_->mtx);
251 : 2 : impl_->pause_flag.store(false, std::memory_order_relaxed);
252 : 2 : }
253 : :
254 : 2 : impl_->cv.notify_one();
255 : 2 : }
256 : :
257 : 4 : void VCAPReader::pause_to_next() {
258 : : {
259 [ + - ]: 4 : std::unique_lock lock(impl_->mtx);
260 : :
261 [ + + ]: 4 : if (!impl_->pause_flag.load(std::memory_order_relaxed)) {
262 : 1 : return;
263 : : }
264 : :
265 : 3 : impl_->pause_next_flag.store(true, std::memory_order_relaxed);
266 [ + + ]: 4 : }
267 : :
268 : 3 : impl_->cv.notify_one();
269 : : }
270 : :
271 : 4 : void VCAPReader::jump(int64_t begin_time, double rate, int times, bool force_to_play) {
272 [ + + ]: 4 : if (begin_time < 0) {
273 : 1 : begin_time = 0;
274 [ + + ]: 3 : } else if (begin_time > impl_->info.total_duration) {
275 : 1 : begin_time = std::max<int64_t>(0, impl_->info.total_duration - 100);
276 : : }
277 : :
278 : 4 : impl_->real_elapsed.store(begin_time * 1000U, std::memory_order_relaxed);
279 : 4 : impl_->is_pending.store(true, std::memory_order_relaxed);
280 : :
281 : 4 : bool last_pause_flag = impl_->pause_flag.load(std::memory_order_relaxed);
282 : :
283 : : {
284 [ + - ]: 4 : std::unique_lock lock(impl_->mtx);
285 : 4 : impl_->stop_flag.store(false, std::memory_order_relaxed);
286 : 4 : impl_->pause_flag.store(false, std::memory_order_relaxed);
287 : 4 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
288 : 4 : impl_->jump_flag.store(true, std::memory_order_relaxed);
289 : 4 : }
290 : :
291 : 4 : impl_->cv.notify_one();
292 : :
293 [ + - ]: 4 : wait_for_idle();
294 : :
295 : 4 : impl_->begin_time.store(begin_time, std::memory_order_relaxed);
296 : :
297 [ + + ]: 4 : if (rate <= 0) {
298 : 3 : impl_->rate.store(1, std::memory_order_relaxed);
299 : : } else {
300 : 1 : impl_->rate.store(rate, std::memory_order_relaxed);
301 : : }
302 : :
303 : 4 : impl_->times.store(times, std::memory_order_relaxed);
304 : :
305 : : {
306 [ + - ]: 4 : std::unique_lock lock(impl_->mtx);
307 : 4 : impl_->stop_flag.store(false, std::memory_order_relaxed);
308 [ + + ]: 5 : impl_->pause_flag.store(force_to_play ? false : last_pause_flag, std::memory_order_relaxed);
309 : 4 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
310 : 4 : impl_->jump_flag.store(false, std::memory_order_relaxed);
311 : 4 : }
312 : :
313 : 4 : Config config_snapshot;
314 : :
315 : : {
316 [ + - ]: 4 : std::unique_lock lock(impl_->config_mtx);
317 [ + - ]: 4 : config_snapshot = impl_->config;
318 : 4 : }
319 : :
320 [ + - + - : 8 : post_task([this, config_snapshot]() { read(config_snapshot); });
+ - ]
321 : 5 : }
322 : :
323 : 32 : std::future<bool> VCAPReader::check() {
324 [ + + ]: 32 : if VUNLIKELY (is_busy()) {
325 [ + - + - ]: 2 : VLOG_W("VCAPReader: Is busy.");
326 : : // return std::future<bool>();
327 : : }
328 : :
329 : 378 : return invoke_task([this]() {
330 [ + + ]: 32 : if (!impl_->total_has_completed) {
331 [ + - + - ]: 2 : VLOG_W("VCAPReader: Incomplete data detected.");
332 : 1 : return false;
333 : : }
334 : :
335 : 31 : mcap::Status status;
336 : :
337 : : // LCOV_EXCL_START GCOVR_EXCL_START
338 : : auto status_function = [](const mcap::Status& status) {
339 : : if (!status.ok()) {
340 : : CLOG_W("VCAPReader: Failed to check summary, error = %s.", status.message.c_str());
341 : : }
342 : : };
343 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
344 : :
345 [ + + ]: 81 : for (auto& wrapper_file : impl_->file_list) {
346 [ - + ]: 50 : if VUNLIKELY (!wrapper_file.reader) {
347 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
348 : : }
349 : :
350 [ + - ]: 50 : status = wrapper_file.reader->readSummary(mcap::ReadSummaryMethod::ForceScan, status_function);
351 : :
352 [ - + ]: 50 : if VUNLIKELY (!status.ok()) {
353 : : CLOG_W("VCAPReader: Failed to check whole summary, error = %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
354 : : status.message.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
355 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
356 : : }
357 : : }
358 : :
359 : 31 : bool is_ok = true;
360 : :
361 [ + + ]: 31 : if VUNLIKELY (impl_->info.total_duration < impl_->info.blank_duration) {
362 [ + - + - ]: 2 : CLOG_W("VCAPReader: Invalid duration, blank=%" PRId64 " total=%" PRId64 ".",
363 : : static_cast<int64_t>(impl_->info.blank_duration), static_cast<int64_t>(impl_->info.total_duration));
364 : 1 : is_ok = false;
365 : : }
366 : :
367 [ + + + + : 31 : if VUNLIKELY (impl_->info.message_count > 0 && impl_->info.url_metas.empty()) {
+ + ]
368 [ + - + - ]: 2 : CLOG_W("VCAPReader: Message count is %" PRId64 " but url meta list is empty.",
369 : : static_cast<int64_t>(impl_->info.message_count));
370 : 1 : is_ok = false;
371 : : }
372 : :
373 : 31 : size_t total_count = 0;
374 : 31 : size_t total_raw_size = 0;
375 : :
376 [ + + ]: 78 : for (const auto& url_meta : impl_->info.url_metas) {
377 : 47 : total_count += url_meta.count;
378 : 47 : total_raw_size += url_meta.size;
379 : :
380 [ - + ]: 47 : if VUNLIKELY (!url_meta.valid) {
381 : : CLOG_W("VCAPReader: Invalid url meta detected at index=%d.", url_meta.index); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
382 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
383 : : }
384 : :
385 [ + + ]: 47 : if VUNLIKELY (url_meta.url.empty()) {
386 [ + - + - ]: 2 : CLOG_W("VCAPReader: Empty url detected at index=%d.", url_meta.index);
387 : 1 : is_ok = false;
388 : : }
389 : :
390 [ + + ]: 47 : if VUNLIKELY (url_meta.url_type.empty()) {
391 [ + - + - ]: 2 : CLOG_W("VCAPReader: Empty url_type detected for url=%s.", url_meta.url.c_str());
392 : 1 : is_ok = false;
393 : : }
394 : :
395 [ + - + + : 47 : if VUNLIKELY (url_meta.count > 0 && url_meta.ser_type.empty()) {
+ + ]
396 [ + - + - ]: 4 : CLOG_W("VCAPReader: Empty ser_type detected for url=%s.", url_meta.url.c_str());
397 : 2 : is_ok = false;
398 : : }
399 : :
400 [ - + ]: 47 : if VUNLIKELY (!SchemaData::is_valid_type(url_meta.schema_type)) {
401 : : CLOG_W("VCAPReader: Invalid schema_type=%d detected for url=%s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
402 : : static_cast<int>(url_meta.schema_type), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
403 : : url_meta.url.c_str());
404 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
405 : : }
406 : :
407 : 47 : auto inferred_schema_type = SchemaData::infer_ser_type(url_meta.ser_type);
408 : :
409 [ + + + + : 47 : if VUNLIKELY (url_meta.schema_type == SchemaType::kUnknown && inferred_schema_type != SchemaType::kUnknown) {
+ + ]
410 : 1 : const auto schema_label = SchemaData::convert_type(inferred_schema_type);
411 [ + - + - ]: 2 : CLOG_W("VCAPReader: Missing schema_type for url=%s, inferred=%.*s.", url_meta.url.c_str(),
412 : : static_cast<int>(schema_label.size()), schema_label.data());
413 : 1 : is_ok = false;
414 : : }
415 : :
416 [ + + + + : 47 : if VUNLIKELY (url_meta.loss < 0.0 || url_meta.loss > 1.0) {
+ + ]
417 [ + - + - ]: 6 : CLOG_W("VCAPReader: Invalid loss=%f detected for url=%s.", url_meta.loss, url_meta.url.c_str());
418 : 3 : is_ok = false;
419 : : }
420 : :
421 [ + + ]: 47 : if VUNLIKELY (url_meta.freq < 0.0) {
422 [ + - + - ]: 6 : CLOG_W("VCAPReader: Invalid freq=%f detected for url=%s.", url_meta.freq, url_meta.url.c_str());
423 : 3 : is_ok = false;
424 : : }
425 : : }
426 : :
427 [ + + + + : 61 : if ((!impl_->info.url_metas.empty() || impl_->info.message_count != 0) &&
+ + ]
428 [ + + ]: 30 : total_count != static_cast<size_t>(impl_->info.message_count)) {
429 [ + - + - ]: 4 : CLOG_W("VCAPReader: Message count mismatch, header=%" PRId64 " metas=%zu.",
430 : : static_cast<int64_t>(impl_->info.message_count), total_count);
431 : 2 : is_ok = false;
432 : : }
433 : :
434 [ + + - + : 60 : if ((!impl_->info.url_metas.empty() || impl_->info.total_raw_size != 0) &&
- + ]
435 [ - + ]: 29 : total_raw_size != static_cast<size_t>(impl_->info.total_raw_size)) {
436 : : CLOG_W("VCAPReader: Raw size mismatch, header=%" PRId64 " metas=%zu.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
437 : : static_cast<int64_t>(impl_->info.total_raw_size), total_raw_size);
438 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
439 : : }
440 : :
441 [ + - + + ]: 38 : for (const auto& schema_data : detect_schema()) {
442 [ - + ]: 7 : if VUNLIKELY (schema_data.name.empty()) {
443 : : CLOG_W("VCAPReader: Empty schema name detected."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
444 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
445 : : }
446 : :
447 [ - + ]: 7 : if VUNLIKELY (schema_data.encoding.empty()) {
448 : : CLOG_W("VCAPReader: Empty schema encoding detected for name=%s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
449 : : schema_data.name.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
450 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
451 : : }
452 : :
453 [ + - + + : 7 : if VUNLIKELY (!SchemaData::is_valid_type(schema_data.schema_type) ||
+ + ]
454 : : schema_data.schema_type == SchemaType::kUnknown) {
455 [ + - + - ]: 2 : CLOG_W("VCAPReader: Invalid schema_type=%d detected for schema=%s.", static_cast<int>(schema_data.schema_type),
456 : : schema_data.name.c_str());
457 : 1 : is_ok = false;
458 : : }
459 : 31 : }
460 : :
461 : 31 : return is_ok;
462 [ + - ]: 63 : });
463 : : }
464 : :
465 : 3 : std::future<bool> VCAPReader::reindex() {
466 [ + + ]: 3 : if VUNLIKELY (is_busy()) {
467 [ + - + - ]: 2 : VLOG_W("VCAPReader: Is busy.");
468 : : // return std::future<bool>();
469 : : }
470 : :
471 : 3 : return invoke_task([]() {
472 [ + - + - ]: 6 : VLOG_W("VCAPReader: Reindex is not supported for vcap.");
473 : :
474 : 3 : return false;
475 [ + - ]: 3 : });
476 : : }
477 : :
478 : 3 : std::future<bool> VCAPReader::fix(bool rebuild) {
479 [ + + ]: 3 : if VUNLIKELY (is_busy()) {
480 [ + - + - ]: 2 : VLOG_W("VCAPReader: Is busy.");
481 : : // return std::future<bool>();
482 : : }
483 : :
484 : 3 : return invoke_task([rebuild]() {
485 : : (void)rebuild;
486 : :
487 [ + - + - ]: 6 : VLOG_W("VCAPReader: Fix is not supported for vcap.");
488 : :
489 : 3 : return false;
490 [ + - ]: 3 : });
491 : : }
492 : :
493 : 8 : void VCAPReader::tag(const std::string& tag_name) {
494 [ + + ]: 8 : if VUNLIKELY (is_busy()) {
495 [ + - + - ]: 2 : VLOG_W("VCAPReader: Is busy.");
496 : : // return;
497 : : }
498 : :
499 [ + - + - ]: 8 : post_task([this, tag_name]() {
500 : : try {
501 : : #ifdef _WIN32
502 : : std::filesystem::path file_path(Helpers::string_to_wstring(impl_->path));
503 : : std::string suffix = Helpers::path_to_string(file_path.extension());
504 : : #else
505 [ + - ]: 8 : std::filesystem::path file_path(impl_->path);
506 [ + - + - ]: 8 : std::string suffix = file_path.extension().string();
507 : : #endif
508 : :
509 : 52 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
510 : :
511 [ + + ]: 8 : if (suffix == ".vcapx") {
512 : : try {
513 : 4 : nlohmann::ordered_json root_json;
514 : 4 : nlohmann::ordered_json header_json;
515 : :
516 : : {
517 [ + - ]: 4 : std::ifstream file(file_path);
518 : :
519 [ + + ]: 4 : file >> root_json;
520 : :
521 [ + - ]: 3 : file.close();
522 : 4 : }
523 : :
524 [ + - + - ]: 3 : header_json = root_json["VLinkHeader"];
525 : :
526 [ + - + - ]: 3 : header_json["tag"] = tag_name;
527 : :
528 [ + - ]: 3 : root_json["VLinkHeader"] = std::move(header_json);
529 : :
530 : : {
531 [ + - ]: 3 : std::ofstream filex(impl_->path, std::ios::out | std::ios::trunc);
532 : :
533 [ + - + - ]: 3 : if VLIKELY (filex.is_open()) {
534 [ + - + - ]: 3 : filex << root_json.dump(4);
535 [ + - ]: 3 : filex.close();
536 : : }
537 : 3 : }
538 [ - + ]: 6 : } catch (nlohmann::json::exception& e) {
539 [ + - + - ]: 2 : VLOG_W("VCAPReader: JSON parse error, ", e.what(), ".");
540 : 1 : }
541 : : } else {
542 [ + - + - ]: 8 : VLOG_W("VCAPReader: Tag is not supported for single vcap.");
543 : : }
544 [ - - ]: 8 : } catch (std::filesystem::filesystem_error& e) {
545 : : VLOG_F("VCAPReader: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
546 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
547 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
548 : : });
549 : 8 : }
550 : :
551 : 5 : int64_t VCAPReader::get_timestamp() const {
552 [ + - ]: 5 : std::shared_lock time_lock(impl_->time_mtx);
553 : :
554 [ + + ]: 5 : if (impl_->status.load(std::memory_order_relaxed) == kPlaying) {
555 [ - + ]: 1 : if (impl_->is_pending.load(std::memory_order_relaxed)) {
556 : : return impl_->real_elapsed.load(std::memory_order_relaxed) / 1000U; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
557 : : } else {
558 : 1 : return (impl_->real_elapsed.load(std::memory_order_relaxed) +
559 : 1 : (impl_->real_timer.get() * impl_->rate.load(std::memory_order_relaxed))) /
560 : 1 : 1000U;
561 : : }
562 [ + + ]: 4 : } else if (impl_->status.load(std::memory_order_relaxed) == kPaused) {
563 : 2 : return (impl_->real_elapsed.load(std::memory_order_relaxed) +
564 : 2 : ((impl_->real_timer.get() - impl_->pause_elapsed_timer.get() -
565 : 2 : impl_->extra_elapsed.load(std::memory_order_relaxed)) *
566 : 2 : impl_->rate.load(std::memory_order_relaxed))) /
567 : 2 : 1000U;
568 : : } else {
569 : 2 : return 0;
570 : : }
571 : 5 : }
572 : :
573 : 5 : int64_t VCAPReader::get_real_timestamp() const {
574 [ + + + + : 9 : if (impl_->status.load(std::memory_order_relaxed) == kPlaying ||
+ + ]
575 : 4 : impl_->status.load(std::memory_order_relaxed) == kPaused) {
576 : 6 : return impl_->real_elapsed.load(std::memory_order_relaxed) / 1000U;
577 : : } else {
578 : 2 : return 0;
579 : : }
580 : : }
581 : :
582 : 10 : BagReader::Status VCAPReader::get_status() const { return impl_->status.load(std::memory_order_relaxed); }
583 : :
584 : 28 : const BagReader::Info& VCAPReader::get_info() const { return impl_->info; }
585 : :
586 : 35 : std::vector<SchemaData> VCAPReader::detect_schema() {
587 : 35 : std::vector<SchemaData> schema_list;
588 : 35 : std::unordered_map<std::string, size_t> schema_index_map;
589 : :
590 [ + + ]: 35 : if (!impl_->info.has_schema) {
591 : 24 : return schema_list;
592 : : }
593 : :
594 [ + - ]: 11 : schema_index_map.reserve(impl_->info.url_metas.size());
595 : :
596 [ + + ]: 24 : for (auto& wrapper_file : impl_->file_list) {
597 [ + - + + ]: 25 : for (const auto& [schema_id, schema_ptr] : wrapper_file.reader->schemas()) {
598 : : (void)schema_id;
599 : :
600 : 12 : SchemaData schema;
601 [ + - ]: 12 : schema.name = schema_ptr->name;
602 [ + - ]: 12 : schema.encoding = schema_ptr->encoding;
603 : 12 : schema.schema_type = SchemaData::resolve_type(SchemaType::kUnknown, schema.name, schema.encoding);
604 : :
605 [ + - + - : 12 : if (!schema.name.empty() && !schema_ptr->data.empty()) {
+ - ]
606 [ + - ]: 12 : std::string schema_key = schema.name;
607 [ + - ]: 12 : schema_key.push_back('\x1F');
608 [ + - ]: 12 : schema_key.append(SchemaData::convert_type(schema.schema_type));
609 [ + - ]: 12 : auto schema_index_iter = schema_index_map.find(schema_key);
610 : :
611 [ + + ]: 12 : if (schema_index_iter == schema_index_map.end()) {
612 : : schema.data =
613 : 10 : Bytes::deep_copy(reinterpret_cast<const uint8_t*>(schema_ptr->data.data()), schema_ptr->data.size());
614 [ + - ]: 10 : schema_index_map.emplace(schema_key, schema_list.size());
615 [ + - ]: 10 : schema_list.emplace_back(std::move(schema));
616 : : } else {
617 : 2 : auto& current_schema = schema_list[schema_index_iter->second];
618 : :
619 [ - + - - : 2 : if (current_schema.encoding.empty() && !schema.encoding.empty()) {
- + ]
620 : : current_schema.encoding = schema.encoding; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
621 : : }
622 : :
623 [ - + ]: 2 : if (current_schema.data.empty()) {
624 : 0 : current_schema.data = Bytes::deep_copy(
625 : : reinterpret_cast<const uint8_t*>(schema_ptr->data.data()), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
626 : : schema_ptr->data.size()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
627 : : }
628 : : }
629 : 12 : }
630 : 25 : }
631 : : }
632 : :
633 : 11 : return schema_list;
634 : 35 : }
635 : :
636 : 9 : bool VCAPReader::is_split_mode() const { return impl_->info.split_count > 0; }
637 : :
638 : 4 : int VCAPReader::get_split_index() const { return impl_->split_index.load(std::memory_order_relaxed); }
639 : :
640 : 4 : bool VCAPReader::is_jumping() const { return impl_->jump_flag.load(std::memory_order_relaxed); }
641 : :
642 : 59 : size_t VCAPReader::get_max_task_count() const { return kMaxTaskSize; }
643 : :
644 : 56 : void VCAPReader::on_begin() { MessageLoop::on_begin(); }
645 : :
646 : 56 : void VCAPReader::on_end() { MessageLoop::on_end(); }
647 : :
648 : 30 : void VCAPReader::update_status(Status status) {
649 : 30 : bool has_changed = false;
650 : :
651 [ + + ]: 30 : if (status == kStopped) {
652 [ + + ]: 11 : if (impl_->status.load(std::memory_order_relaxed) != kStopped) {
653 : 10 : impl_->status.store(kStopped, std::memory_order_relaxed);
654 : 10 : has_changed = true;
655 : : }
656 [ + + ]: 19 : } else if (status == kPaused) {
657 [ + + ]: 5 : if (impl_->status.load(std::memory_order_relaxed) != kPaused) {
658 : 4 : impl_->status.store(kPaused, std::memory_order_relaxed);
659 : 4 : has_changed = true;
660 : : }
661 [ + - ]: 14 : } else if (status == kPlaying) {
662 [ + + ]: 14 : if (impl_->status.load(std::memory_order_relaxed) != kPlaying) {
663 : 13 : impl_->status.store(kPlaying, std::memory_order_relaxed);
664 : 13 : has_changed = true;
665 : : }
666 : : }
667 : :
668 [ + + ]: 30 : if (has_changed) {
669 [ + + ]: 27 : if VLIKELY (impl_->status_callback) {
670 : 17 : impl_->status_callback(impl_->status.load(std::memory_order_relaxed));
671 : : }
672 : : }
673 : 30 : }
674 : :
675 : 61 : void VCAPReader::do_stop() {
676 : : {
677 [ + - ]: 61 : std::unique_lock lock(impl_->mtx);
678 : 61 : impl_->stop_flag.store(true, std::memory_order_relaxed);
679 : 61 : impl_->pause_flag.store(false, std::memory_order_relaxed);
680 : 61 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
681 : 61 : impl_->jump_flag.store(false, std::memory_order_relaxed);
682 : 61 : }
683 : :
684 : 61 : impl_->cv.notify_one();
685 : 61 : }
686 : :
687 : 4 : void VCAPReader::do_pause() {
688 [ + - ]: 4 : std::unique_lock lock(impl_->mtx);
689 : :
690 [ + - ]: 4 : while (impl_->pause_flag.load(std::memory_order_relaxed)) {
691 : 4 : impl_->pause_elapsed_timer.restart();
692 [ + - ]: 4 : update_status(kPaused);
693 : :
694 : 4 : impl_->cv.wait(lock, [this]() -> bool {
695 [ + + + + ]: 19 : return impl_->stop_flag.load(std::memory_order_relaxed) || !impl_->pause_flag.load(std::memory_order_relaxed) ||
696 [ + - ]: 8 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
697 [ + - - + ]: 17 : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
698 : : });
699 : :
700 : 4 : impl_->pause_elapsed.fetch_add(impl_->pause_elapsed_timer.get(), std::memory_order_relaxed);
701 : :
702 : : {
703 [ + - ]: 4 : std::lock_guard time_lock(impl_->time_mtx);
704 : 4 : impl_->real_timer.restart();
705 : :
706 [ - + ]: 8 : if (impl_->offset_elapsed.load(std::memory_order_relaxed) > 0) {
707 : : impl_->real_elapsed.fetch_add( // LCOV_EXCL_LINE GCOVR_EXCL_LINE
708 : : (impl_->offset_timer.get() - impl_->pause_elapsed_timer.get()) * // LCOV_EXCL_LINE GCOVR_EXCL_LINE
709 : : impl_->rate.load(std::memory_order_relaxed), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
710 : : std::memory_order_relaxed);
711 : : }
712 : 4 : }
713 : :
714 [ + - ]: 4 : update_status(kPlaying);
715 : :
716 [ + - ]: 4 : if (impl_->pause_next_flag.load(std::memory_order_relaxed)) {
717 : 8 : impl_->pause_elapsed.fetch_sub(impl_->offset_elapsed.load(std::memory_order_relaxed), std::memory_order_relaxed);
718 : 4 : break;
719 : : // LCOV_EXCL_START GCOVR_EXCL_START
720 : : } else if (impl_->offset_elapsed.load(std::memory_order_relaxed) > 0) {
721 : : impl_->offset_timer.restart();
722 : :
723 : : impl_->cv.wait_for(lock, std::chrono::microseconds(impl_->offset_elapsed.load(std::memory_order_relaxed)),
724 : : [this]() -> bool {
725 : : return impl_->stop_flag.load(std::memory_order_relaxed) ||
726 : : impl_->pause_flag.load(std::memory_order_relaxed) ||
727 : : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
728 : : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
729 : : });
730 : :
731 : : if VUNLIKELY (impl_->pause_flag.load(std::memory_order_relaxed)) {
732 : : impl_->offset_elapsed.fetch_sub(impl_->offset_timer.get(), std::memory_order_relaxed);
733 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
734 : : } else {
735 : : impl_->offset_elapsed.store(0, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
736 : : }
737 : : }
738 : : }
739 : 4 : }
740 : :
741 : 87 : bool VCAPReader::prepare_file(void* file) {
742 : 87 : auto* wrapper_file = static_cast<Impl::WrapperFile*>(file);
743 : :
744 : 87 : wrapper_file->has_completed = true;
745 : :
746 : 87 : auto& reader = wrapper_file->reader;
747 : :
748 [ - + ]: 87 : if VUNLIKELY (!reader) {
749 : : wrapper_file->has_completed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
750 : : CLOG_F("VCAPReader: Mcap [%s] reader is nullptr.", wrapper_file->path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
751 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
752 : : }
753 : :
754 [ + - - + ]: 87 : if VUNLIKELY (!reader->header()) {
755 : : // LCOV_EXCL_START GCOVR_EXCL_START
756 : : wrapper_file->has_completed = false;
757 : : CLOG_F("VCAPReader: Mcap [%s] reader header is nullptr.", wrapper_file->path.c_str());
758 : : return false;
759 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
760 : : }
761 : :
762 [ + - - + ]: 87 : if VUNLIKELY (reader->header()->profile != "vlink") {
763 : : // LCOV_EXCL_START GCOVR_EXCL_START
764 : : wrapper_file->has_completed = false;
765 : : CLOG_F("VCAPReader: Mcap [%s] profile is %s, not valid.", wrapper_file->path.c_str(),
766 : : reader->header()->profile.c_str());
767 : : return false;
768 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
769 : : }
770 : :
771 : 87 : mcap::Status status;
772 : :
773 [ + - ]: 87 : status = reader->readSummary(mcap::ReadSummaryMethod::NoFallbackScan);
774 : :
775 [ - + ]: 87 : if VUNLIKELY (!status.ok()) {
776 : : // LCOV_EXCL_START GCOVR_EXCL_START
777 : : wrapper_file->has_completed = false;
778 : :
779 : : if (impl_->try_to_fix) {
780 : : CLOG_E("VCAPReader: Failed to read summary, error = %s. Trying to fix.", status.message.c_str());
781 : : status = reader->readSummary(mcap::ReadSummaryMethod::AllowFallbackScan);
782 : :
783 : : if VUNLIKELY (!status.ok()) {
784 : : CLOG_F("VCAPReader: Failed to read summary, error = %s.", status.message.c_str());
785 : : return false;
786 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
787 : : }
788 : :
789 : : } else {
790 : : CLOG_F("VCAPReader: Failed to read summary, error = %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
791 : : status.message.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
792 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
793 : : }
794 : : }
795 : :
796 [ + - ]: 87 : const auto& meta_index = reader->metadataIndexes();
797 [ + - ]: 87 : const auto& statistics = reader->statistics();
798 : :
799 [ - + ]: 87 : if (!statistics.has_value()) {
800 : : // LCOV_EXCL_START GCOVR_EXCL_START
801 : : wrapper_file->has_completed = false;
802 : : CLOG_F("VCAPReader: Mcap [%s] cannot find statistics.", wrapper_file->path.c_str());
803 : : return false;
804 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
805 : : }
806 : :
807 [ + - + + ]: 87 : for (const auto& [schema_id, schema_ptr] : reader->schemas()) {
808 : : (void)schema_id;
809 : :
810 [ + - + - : 18 : if (schema_ptr && !schema_ptr->data.empty()) {
+ - ]
811 : 18 : wrapper_file->has_schema = true;
812 : 18 : impl_->info.has_schema = true;
813 : 18 : break;
814 : : }
815 : 87 : }
816 : :
817 : : // read header
818 : : {
819 [ + - + - ]: 87 : auto header_iter = meta_index.find("VLinkHeader");
820 : :
821 [ - + ]: 87 : if VUNLIKELY (header_iter == meta_index.end()) {
822 : : // LCOV_EXCL_START GCOVR_EXCL_START
823 : : wrapper_file->has_completed = false;
824 : : CLOG_F("VCAPReader: Mcap [%s] cannot find header.", wrapper_file->path.c_str());
825 : : return false;
826 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
827 : : }
828 : :
829 : 87 : const auto& header_index = header_iter->second;
830 : :
831 : : mcap::Record header_record;
832 : :
833 [ + - + - ]: 87 : status = mcap::McapReader::ReadRecord(*reader->dataSource(), header_index.offset, &header_record);
834 : :
835 [ - + ]: 87 : if VUNLIKELY (!status.ok()) {
836 : : // LCOV_EXCL_START GCOVR_EXCL_START
837 : : wrapper_file->has_completed = false;
838 : : CLOG_F("VCAPReader: Failed to read header record for index, error = %s.", status.message.c_str());
839 : : return false;
840 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
841 : : }
842 : :
843 : 87 : mcap::Metadata header_meta_data;
844 : :
845 [ + - ]: 87 : status = mcap::McapReader::ParseMetadata(header_record, &header_meta_data);
846 : :
847 [ - + ]: 87 : if VUNLIKELY (!status.ok()) {
848 : : // LCOV_EXCL_START GCOVR_EXCL_START
849 : : wrapper_file->has_completed = false;
850 : : CLOG_F("VCAPReader: Failed to parse header meta data, error = %s.", status.message.c_str());
851 : : return false;
852 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
853 : : }
854 : :
855 [ + - + - ]: 87 : auto& tag_str = header_meta_data.metadata["tag"];
856 [ + - + - ]: 87 : auto& version_str = header_meta_data.metadata["version"];
857 [ + - + - ]: 87 : auto& compress_str = header_meta_data.metadata["compress"];
858 [ + - + - ]: 87 : auto& process_str = header_meta_data.metadata["process"];
859 [ + - + - ]: 87 : auto& date_str = header_meta_data.metadata["date"];
860 [ + - + - ]: 87 : auto& start_timestamp_str = header_meta_data.metadata["start_timestamp"];
861 [ + - + - ]: 87 : auto& timezone_str = header_meta_data.metadata["timezone"];
862 : :
863 [ - + ]: 87 : if VUNLIKELY (version_str.empty()) {
864 : : // LCOV_EXCL_START GCOVR_EXCL_START
865 : : wrapper_file->has_completed = false;
866 : :
867 : : if (impl_->read_only) {
868 : : CLOG_E("VCAPReader: Mcap [%s] cannot find version in header.", wrapper_file->path.c_str());
869 : : }
870 : :
871 : : version_str = "0.0.0";
872 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
873 : : } else {
874 : 87 : auto version = Version::from_string(version_str);
875 : :
876 [ + + ]: 87 : if VUNLIKELY (!version.is_valid()) {
877 : 1 : wrapper_file->has_completed = false;
878 : :
879 [ - + ]: 1 : if (impl_->read_only) {
880 : : CLOG_E("VCAPReader: Mcap [%s] header version is invalid.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
881 : : wrapper_file->path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
882 : : }
883 : : } else {
884 [ + + ]: 86 : if VUNLIKELY (version.major != VLINK_VERSION_MAJOR) {
885 : 1 : wrapper_file->has_completed = false;
886 [ + - - + ]: 2 : VLOG_F("VCAPReader: Mcap version is incompatible.");
887 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
888 : : }
889 : : }
890 : : }
891 : :
892 [ - + ]: 86 : if VUNLIKELY (compress_str.empty()) {
893 : : // LCOV_EXCL_START GCOVR_EXCL_START
894 : : wrapper_file->has_completed = false;
895 : :
896 : : if (impl_->read_only) {
897 : : CLOG_E("VCAPReader: Mcap [%s] cannot find compress in header.", wrapper_file->path.c_str());
898 : : }
899 : :
900 : : compress_str = "Unknown";
901 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
902 : : }
903 : :
904 [ - + ]: 86 : if VUNLIKELY (process_str.empty()) {
905 : : // LCOV_EXCL_START GCOVR_EXCL_START
906 : : wrapper_file->has_completed = false;
907 : :
908 : : if (impl_->read_only) {
909 : : CLOG_E("VCAPReader: Mcap [%s] cannot find process in header.", wrapper_file->path.c_str());
910 : : }
911 : :
912 : : process_str = "Unknown";
913 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
914 : : }
915 : :
916 [ - + ]: 86 : if VUNLIKELY (date_str.empty()) {
917 : : // LCOV_EXCL_START GCOVR_EXCL_START
918 : : wrapper_file->has_completed = false;
919 : :
920 : : if (impl_->read_only) {
921 : : CLOG_E("VCAPReader: Mcap [%s] cannot find date in header.", wrapper_file->path.c_str());
922 : : }
923 : :
924 : : date_str = "Unknown";
925 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
926 : : }
927 : :
928 [ - + ]: 86 : if VUNLIKELY (timezone_str.empty()) {
929 : : // LCOV_EXCL_START GCOVR_EXCL_START
930 : : wrapper_file->has_completed = false;
931 : :
932 : : if (impl_->read_only) {
933 : : CLOG_E("VCAPReader: Mcap [%s] cannot find timezone in header.", wrapper_file->path.c_str());
934 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
935 : : }
936 : : }
937 : :
938 [ - + ]: 86 : if (tag_str.empty()) {
939 : : tag_str = "Empty"; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
940 : : }
941 : :
942 [ + - ]: 86 : impl_->info.tag_name = tag_str;
943 [ + - ]: 86 : impl_->info.version = version_str;
944 [ + - ]: 86 : impl_->info.storage_type = "vcap";
945 : 86 : impl_->info.message_count = statistics->messageCount;
946 [ + - ]: 86 : impl_->info.time_accuracy = "MicroSecond";
947 [ + - ]: 86 : impl_->info.compression_type = compress_str;
948 [ + - ]: 86 : impl_->info.process_name = process_str;
949 [ + - ]: 86 : impl_->info.date_time = date_str;
950 : :
951 : : try {
952 [ + + ]: 86 : impl_->info.start_timestamp = std::stoll(start_timestamp_str);
953 [ - + ]: 1 : } catch (std::exception&) {
954 : 1 : impl_->info.start_timestamp = Helpers::convert_date_to_timestamp(impl_->info.date_time) / 1000'000;
955 : 1 : }
956 : :
957 [ + + ]: 86 : if VUNLIKELY (impl_->info.start_timestamp < 0) {
958 : 1 : impl_->info.start_timestamp = 0;
959 : :
960 [ - + ]: 1 : if (impl_->read_only) {
961 : : VLOG_E("VCAPReader: Invalid start_timestamp_ns."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
962 : : }
963 : : }
964 : :
965 : 86 : int64_t timestamp_diff = 0;
966 : :
967 : 86 : timestamp_diff = static_cast<int64_t>(statistics->messageStartTime / 1000'000) - impl_->info.start_timestamp;
968 : :
969 [ + + ]: 86 : if (timestamp_diff < 0) {
970 : 3 : timestamp_diff = 0;
971 : 3 : impl_->info.start_timestamp = static_cast<int64_t>(statistics->messageStartTime / 1000'000);
972 : : }
973 : :
974 : 86 : impl_->info.blank_duration = timestamp_diff;
975 : :
976 : 86 : timestamp_diff = static_cast<int64_t>(statistics->messageEndTime / 1000'000) - impl_->info.start_timestamp;
977 : :
978 [ - + ]: 86 : if (timestamp_diff < 0) {
979 : : timestamp_diff = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
980 : : }
981 : :
982 : 86 : impl_->info.total_duration = timestamp_diff;
983 : :
984 : : try {
985 [ + - ]: 86 : impl_->info.timezone = std::stoi(timezone_str);
986 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
987 : : impl_->info.timezone = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
988 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
989 [ + - ]: 87 : }
990 : :
991 : : // read channel
992 : : {
993 : 86 : int channel = 0;
994 : 86 : std::string channel_str;
995 : : mcap::Record channel_record;
996 : :
997 [ + - ]: 86 : url_ser_map().clear();
998 [ + - ]: 86 : url_schema_type_map().clear();
999 : 86 : impl_->info.url_metas.clear();
1000 : 86 : impl_->raw_url_metas.clear();
1001 : :
1002 : 86 : impl_->info.total_raw_size = 0;
1003 : :
1004 [ + + ]: 286 : for (const auto& [name, index] : meta_index) {
1005 [ + + ]: 200 : if (!Helpers::has_startwith(name, "VLinkChannel_")) {
1006 : 86 : continue;
1007 : : }
1008 : :
1009 [ + - ]: 114 : channel_str = name;
1010 [ + - + - ]: 114 : Helpers::replace_string(channel_str, "VLinkChannel_", "");
1011 : :
1012 : : try {
1013 [ + - ]: 114 : channel = std::stoi(channel_str);
1014 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1015 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1016 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1017 : :
1018 [ + - ]: 114 : auto channel_count_iter = statistics->channelMessageCounts.find(channel);
1019 : :
1020 [ - + ]: 114 : if VUNLIKELY (channel_count_iter == statistics->channelMessageCounts.end()) {
1021 : : // LCOV_EXCL_START GCOVR_EXCL_START
1022 : : wrapper_file->has_completed = false;
1023 : :
1024 : : if (impl_->read_only) {
1025 : : CLOG_E("VCAPReader: Mcap [%s] cannot read statistics in channel.", wrapper_file->path.c_str());
1026 : : }
1027 : :
1028 : : continue;
1029 : : }
1030 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1031 : :
1032 : 114 : auto channel_msg_count = channel_count_iter->second;
1033 : :
1034 [ + - ]: 114 : const auto& channel_ptr = reader->channel(channel);
1035 : :
1036 [ - + ]: 114 : if VUNLIKELY (!channel_ptr) {
1037 : : // LCOV_EXCL_START GCOVR_EXCL_START
1038 : : wrapper_file->has_completed = false;
1039 : :
1040 : : if (impl_->read_only) {
1041 : : CLOG_E("VCAPReader: Mcap [%s] cannot find ptr in channel.", wrapper_file->path.c_str());
1042 : : }
1043 : :
1044 : : continue;
1045 : : }
1046 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1047 : :
1048 [ + - + - ]: 114 : status = mcap::McapReader::ReadRecord(*reader->dataSource(), index.offset, &channel_record);
1049 : :
1050 [ - + ]: 114 : if VUNLIKELY (!status.ok()) {
1051 : : // LCOV_EXCL_START GCOVR_EXCL_START
1052 : : wrapper_file->has_completed = false;
1053 : :
1054 : : if (impl_->read_only) {
1055 : : CLOG_E("VCAPReader: Failed to read channel record for index, error = %s.", status.message.c_str());
1056 : : }
1057 : :
1058 : : continue;
1059 : : }
1060 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1061 : :
1062 : 114 : mcap::Metadata channel_meta_data;
1063 : :
1064 [ + - ]: 114 : status = mcap::McapReader::ParseMetadata(channel_record, &channel_meta_data);
1065 : :
1066 [ - + ]: 114 : if VUNLIKELY (!status.ok()) {
1067 : : // LCOV_EXCL_START GCOVR_EXCL_START
1068 : : wrapper_file->has_completed = false;
1069 : :
1070 : : if (impl_->read_only) {
1071 : : CLOG_E("VCAPReader: Failed to parse channel meta data, error = %s.", status.message.c_str());
1072 : : }
1073 : :
1074 : : continue;
1075 : : }
1076 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1077 : :
1078 : 114 : Info::UrlMeta url_meta;
1079 : :
1080 [ + - + - ]: 114 : auto& index_str = channel_meta_data.metadata["index"];
1081 [ + - + - ]: 114 : auto& type_str = channel_meta_data.metadata["type"];
1082 [ + - + - ]: 114 : auto& count_str = channel_meta_data.metadata["count"];
1083 [ + - + - ]: 114 : auto& size_str = channel_meta_data.metadata["size"];
1084 [ + - + - ]: 114 : auto& loss_str = channel_meta_data.metadata["loss"];
1085 [ + - + - ]: 114 : auto& freq_str = channel_meta_data.metadata["freq"];
1086 [ + - + - ]: 114 : auto ser_iter = channel_meta_data.metadata.find("ser");
1087 [ + - + - ]: 114 : auto encoding_iter = channel_meta_data.metadata.find("encoding");
1088 [ + - + - ]: 114 : auto action_iter = channel_meta_data.metadata.find("action");
1089 [ + - ]: 114 : const auto& schema_ptr = reader->schema(channel_ptr->schemaId);
1090 : :
1091 : 114 : int pindex = -1;
1092 : :
1093 : : try {
1094 [ + - ]: 114 : pindex = std::stoi(index_str);
1095 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1096 : : pindex = -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1097 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1098 : :
1099 [ - + ]: 114 : if VUNLIKELY (pindex != channel - 1) {
1100 : : // LCOV_EXCL_START GCOVR_EXCL_START
1101 : : wrapper_file->has_completed = false;
1102 : :
1103 : : if (impl_->read_only) {
1104 : : CLOG_E("VCAPReader: Mcap [%s] channel index error.", wrapper_file->path.c_str());
1105 : : }
1106 : :
1107 : : continue;
1108 : : }
1109 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1110 : :
1111 : 114 : url_meta.valid = true;
1112 : 114 : url_meta.index = pindex;
1113 [ + - ]: 114 : url_meta.url = channel_ptr->topic;
1114 [ + - ]: 114 : url_meta.url_type = type_str;
1115 : 114 : url_meta.schema_type = SchemaType::kUnknown;
1116 : :
1117 [ + - ]: 114 : if (ser_iter != channel_meta_data.metadata.end()) {
1118 [ + - ]: 114 : url_meta.ser_type = ser_iter->second;
1119 : : } else if (schema_ptr) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1120 : : url_meta.ser_type = schema_ptr->name; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1121 : : }
1122 : :
1123 [ + - ]: 114 : if (encoding_iter != channel_meta_data.metadata.end()) {
1124 : 114 : url_meta.schema_type =
1125 [ + + ]: 132 : SchemaData::resolve_type(SchemaData::convert_encoding(encoding_iter->second), url_meta.ser_type,
1126 : 132 : schema_ptr ? std::string_view(schema_ptr->encoding) : std::string_view{});
1127 : : // LCOV_EXCL_START GCOVR_EXCL_START
1128 : : } else if (schema_ptr) {
1129 : : url_meta.schema_type =
1130 : : SchemaData::resolve_type(SchemaData::convert_encoding(schema_ptr->encoding), url_meta.ser_type);
1131 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1132 : : }
1133 : :
1134 [ + + + - : 114 : if (schema_ptr && !schema_ptr->data.empty()) {
+ + ]
1135 : 18 : wrapper_file->has_schema = true;
1136 : 18 : impl_->info.has_schema = true;
1137 : : }
1138 : :
1139 : : try {
1140 [ + - ]: 114 : url_meta.count = static_cast<size_t>(std::stoull(count_str));
1141 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1142 : : url_meta.count = channel_msg_count; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1143 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1144 : :
1145 : : try {
1146 [ + - ]: 114 : url_meta.loss = std::stod(loss_str);
1147 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1148 : : url_meta.loss = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1149 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1150 : :
1151 : : try {
1152 [ + - ]: 114 : url_meta.size = static_cast<size_t>(std::stoull(size_str));
1153 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1154 : : url_meta.size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1155 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1156 : :
1157 : : try {
1158 [ + - ]: 114 : url_meta.freq = std::stod(freq_str);
1159 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1160 : : url_meta.freq = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1161 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1162 : :
1163 : 114 : impl_->info.total_raw_size += url_meta.size;
1164 : :
1165 [ + - ]: 114 : wrapper_file->id_to_url_map.emplace(url_meta.index, url_meta.url);
1166 [ + - ]: 114 : wrapper_file->url_to_id_map.emplace(url_meta.url, url_meta.index);
1167 : :
1168 [ + - + - ]: 114 : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
1169 [ + - + - ]: 114 : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
1170 : :
1171 [ + - ]: 114 : if (action_iter != channel_meta_data.metadata.end()) {
1172 [ + - ]: 114 : url_meta.action_type = convert_action(action_iter->second);
1173 [ + - ]: 114 : wrapper_file->channel_action_map.emplace(channel, url_meta.action_type);
1174 : : // LCOV_EXCL_START GCOVR_EXCL_START
1175 : : } else if (auto channel_action_iter = channel_ptr->metadata.find("action");
1176 : : channel_action_iter != channel_ptr->metadata.end()) {
1177 : : url_meta.action_type = convert_action(channel_action_iter->second);
1178 : : wrapper_file->channel_action_map.emplace(channel, url_meta.action_type);
1179 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1180 : : } else {
1181 : : wrapper_file->channel_action_map.emplace(channel, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1182 : : ActionType::kUnknownAction); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1183 : : }
1184 : :
1185 [ + - ]: 114 : impl_->info.url_metas.emplace_back(std::move(url_meta));
1186 [ + - + - : 114 : }
+ - + - ]
1187 : :
1188 : 86 : wrapper_file->is_channel_broken = false;
1189 : :
1190 [ + + ]: 86 : if (impl_->info.url_metas.empty()) {
1191 [ + - - + ]: 2 : for (const auto& [id, pchannel] : reader->channels()) {
1192 : : // LCOV_EXCL_START GCOVR_EXCL_START
1193 : : wrapper_file->is_channel_broken = true;
1194 : :
1195 : : auto pschema_ptr = reader->schema(pchannel->schemaId);
1196 : : auto channel_count_iter = statistics->channelMessageCounts.find(id);
1197 : :
1198 : : Info::UrlMeta url_meta;
1199 : :
1200 : : url_meta.valid = true;
1201 : : url_meta.index = id - 1;
1202 : : url_meta.url = pchannel->topic;
1203 : : url_meta.url_type = "Event";
1204 : :
1205 : : if (pschema_ptr) {
1206 : : url_meta.ser_type = pschema_ptr->name;
1207 : : url_meta.schema_type =
1208 : : SchemaData::resolve_type(SchemaData::convert_encoding(pschema_ptr->encoding), url_meta.ser_type);
1209 : :
1210 : : if (!pschema_ptr->data.empty()) {
1211 : : wrapper_file->has_schema = true;
1212 : : impl_->info.has_schema = true;
1213 : : }
1214 : : }
1215 : :
1216 : : if (channel_count_iter != statistics->channelMessageCounts.end()) {
1217 : : url_meta.count = channel_count_iter->second;
1218 : : }
1219 : :
1220 : : wrapper_file->id_to_url_map.emplace(url_meta.index, url_meta.url);
1221 : : wrapper_file->url_to_id_map.emplace(url_meta.url, url_meta.index);
1222 : :
1223 : : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
1224 : : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
1225 : :
1226 : : if (auto action_iter = pchannel->metadata.find("action"); action_iter != pchannel->metadata.end()) {
1227 : : url_meta.action_type = convert_action(action_iter->second);
1228 : : wrapper_file->channel_action_map.emplace(id, url_meta.action_type);
1229 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1230 : : } else {
1231 : : wrapper_file->channel_action_map.emplace(id, ActionType::kUnknownAction); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1232 : : }
1233 : :
1234 : : impl_->info.url_metas.emplace_back(std::move(url_meta)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1235 : 2 : }
1236 : : }
1237 : :
1238 [ + - ]: 86 : std::sort(impl_->info.url_metas.begin(), impl_->info.url_metas.end());
1239 [ + - ]: 86 : impl_->raw_url_metas = impl_->info.url_metas;
1240 [ + - ]: 86 : rebuild_url_meta_lookup(impl_->info.url_metas);
1241 : 86 : }
1242 : :
1243 : 86 : impl_->info.has_idx_elapsed = false;
1244 : 86 : impl_->info.has_idx_url = false;
1245 : :
1246 : 86 : impl_->info.has_completed = wrapper_file->has_completed;
1247 : :
1248 : 86 : return true;
1249 : 87 : }
1250 : :
1251 : 67 : void VCAPReader::open(const std::string& path) {
1252 [ + - ]: 67 : close();
1253 : :
1254 : 175 : auto to_open = [this](Impl::WrapperFile& wrapper_file) -> bool {
1255 : 88 : mcap::Status status;
1256 : :
1257 [ + - ]: 88 : wrapper_file.reader = std::make_unique<mcap::McapReader>();
1258 : :
1259 [ + - ]: 88 : status = wrapper_file.reader->open(wrapper_file.path);
1260 : :
1261 [ + + ]: 88 : if VUNLIKELY (!status.ok()) {
1262 [ + - - + ]: 2 : CLOG_F("VCAPReader: Failed to open vcap, error = %s.", status.message.c_str());
1263 : : wrapper_file.reader.reset(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1264 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1265 : : }
1266 : :
1267 [ + + - + ]: 87 : if VUNLIKELY (!prepare_file(&wrapper_file)) {
1268 : : wrapper_file.reader->close(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1269 : : wrapper_file.reader.reset(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1270 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1271 : : }
1272 : :
1273 : 86 : return true;
1274 : 88 : };
1275 : :
1276 [ + - ]: 67 : impl_->path = path;
1277 : :
1278 : 67 : impl_->total_start_timestamp_ns = -1;
1279 : :
1280 : 67 : impl_->total_has_completed = true;
1281 : :
1282 : : try {
1283 : : #ifdef _WIN32
1284 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
1285 : :
1286 : : impl_->info.file_name = Helpers::path_to_string(file_path.filename());
1287 : :
1288 : : impl_->info.file_size = 0;
1289 : :
1290 : : std::string suffix = Helpers::path_to_string(file_path.extension());
1291 : : #else
1292 [ + - ]: 67 : std::filesystem::path file_path(path);
1293 : :
1294 [ + - + - ]: 67 : impl_->info.file_name = file_path.filename().string();
1295 : :
1296 : 67 : impl_->info.file_size = 0;
1297 : :
1298 [ + - + - ]: 67 : std::string suffix = file_path.extension().string();
1299 : : #endif
1300 : :
1301 : 67 : std::error_code exists_ec;
1302 : :
1303 [ + + ]: 67 : if VUNLIKELY (!std::filesystem::exists(file_path, exists_ec)) {
1304 [ + - - + ]: 2 : CLOG_F("VCAPReader: Mcap [%s] does not exist.", path.c_str());
1305 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1306 : : }
1307 : :
1308 : 66 : std::filesystem::path parent_path;
1309 : :
1310 : : try {
1311 [ + - ]: 66 : parent_path = file_path.parent_path();
1312 : : } catch (std::filesystem::filesystem_error&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1313 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1314 : :
1315 : 422 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
1316 : :
1317 [ + + ]: 66 : if (suffix == ".vcapx") {
1318 : : try {
1319 : 26 : int64_t blank_duration = -1;
1320 : :
1321 : 26 : nlohmann::json root_json;
1322 : :
1323 : : {
1324 [ + - ]: 26 : std::ifstream filex(file_path);
1325 : :
1326 [ + + ]: 26 : filex >> root_json;
1327 : :
1328 [ + - ]: 25 : filex.close();
1329 : 26 : }
1330 : :
1331 [ + - + - ]: 25 : nlohmann::json header_json = root_json["VLinkHeader"];
1332 [ + - + - ]: 25 : nlohmann::json urls_json = root_json["VLinkUrls"];
1333 [ + - + - ]: 25 : nlohmann::json files_json = root_json["VLinkFiles"];
1334 : :
1335 : 25 : impl_->info.file_size = 0;
1336 : :
1337 : 25 : int file_index = 0;
1338 : :
1339 [ + + ]: 25 : if (!files_json.empty()) {
1340 [ + - ]: 24 : impl_->file_list.reserve(files_json.size());
1341 : :
1342 : 24 : impl_->info.has_idx_elapsed = true;
1343 : 24 : impl_->info.has_idx_url = true;
1344 : 24 : impl_->info.has_schema = false;
1345 : : } else {
1346 : 1 : impl_->info.has_idx_elapsed = false;
1347 : 1 : impl_->info.has_idx_url = false;
1348 : 1 : impl_->info.has_schema = false;
1349 : : }
1350 : :
1351 : 25 : std::filesystem::path file_db;
1352 : 25 : std::string file_db_str;
1353 : :
1354 [ + - + - : 73 : for (const auto& file_info : files_json) {
+ - + + ]
1355 : : #ifdef _WIN32
1356 : :
1357 : : if (parent_path.empty()) {
1358 : : file_db = std::filesystem::path(Helpers::string_to_wstring(file_info));
1359 : : } else {
1360 : : file_db = parent_path / std::filesystem::path(Helpers::string_to_wstring(file_info));
1361 : : }
1362 : : #else
1363 : :
1364 [ + + ]: 49 : if (parent_path.empty()) {
1365 [ + - ]: 4 : file_db = std::filesystem::path(file_info);
1366 : : } else {
1367 [ + - + - ]: 45 : file_db = parent_path / std::filesystem::path(file_info);
1368 : : }
1369 : : #endif
1370 : :
1371 [ + - ]: 49 : file_db_str = file_db.string();
1372 : :
1373 : 49 : std::error_code db_exists_ec;
1374 : :
1375 [ + + ]: 49 : if VUNLIKELY (!std::filesystem::exists(file_db, db_exists_ec)) {
1376 [ + - - + ]: 2 : CLOG_F("VCAPReader: Mcap [%s] does not exist.", file_db_str.c_str());
1377 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1378 : : }
1379 : :
1380 [ + - ]: 48 : Impl::WrapperFile wrapper_file;
1381 [ + - ]: 48 : wrapper_file.path = file_db_str;
1382 : 48 : wrapper_file.index = file_index;
1383 : :
1384 [ + - - + ]: 48 : if VUNLIKELY (!to_open(wrapper_file)) {
1385 : : CLOG_W("VCAPReader: Skipping invalid vcap [%s].", file_db_str.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1386 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1387 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1388 : :
1389 [ + - ]: 48 : if (!wrapper_file.has_idx_elapsed) {
1390 : 48 : impl_->info.has_idx_elapsed = false;
1391 : : }
1392 : :
1393 [ + - ]: 48 : if (!wrapper_file.has_idx_url) {
1394 : 48 : impl_->info.has_idx_url = false;
1395 : : }
1396 : :
1397 [ + + ]: 48 : if (wrapper_file.has_schema) {
1398 : 2 : impl_->info.has_schema = true;
1399 : : }
1400 : :
1401 : 48 : std::error_code db_size_ec;
1402 : 48 : std::uintmax_t file_size = std::filesystem::file_size(file_db, db_size_ec);
1403 : :
1404 [ - + ]: 48 : if VUNLIKELY (db_size_ec) {
1405 : : CLOG_W("VCAPReader: file_size failed for [%s]: %s.", file_db_str.c_str(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1406 : : db_size_ec.message().c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1407 : : file_size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1408 : : }
1409 : :
1410 : 48 : impl_->info.file_size += file_size;
1411 : :
1412 : 48 : wrapper_file.start_timestamp_ns = impl_->info.start_timestamp * 1000'000;
1413 : 48 : wrapper_file.begin = impl_->info.blank_duration;
1414 : 48 : wrapper_file.end = impl_->info.total_duration;
1415 : :
1416 [ + + ]: 48 : if (impl_->total_start_timestamp_ns < 0) {
1417 : 23 : impl_->total_start_timestamp_ns = wrapper_file.start_timestamp_ns;
1418 : : }
1419 : :
1420 [ - + ]: 48 : if (!wrapper_file.has_completed) {
1421 : : impl_->total_has_completed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1422 : : }
1423 : :
1424 [ + + ]: 48 : if (blank_duration < 0) {
1425 : 23 : blank_duration = impl_->info.blank_duration;
1426 : : }
1427 : :
1428 [ + - ]: 48 : impl_->file_list.emplace_back(std::move(wrapper_file));
1429 : 48 : ++file_index;
1430 [ + - ]: 48 : }
1431 : :
1432 : 24 : impl_->info.split_count = impl_->file_list.size();
1433 : :
1434 [ + + ]: 24 : if VUNLIKELY (impl_->file_list.empty()) {
1435 [ + - - + ]: 2 : VLOG_F("VCAPReader: DB list is empty.");
1436 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1437 : : }
1438 : :
1439 [ + - + - ]: 23 : int version_major = header_json["major"];
1440 [ + - + - ]: 23 : int version_minor = header_json["minor"];
1441 [ + - + - ]: 23 : int version_patch = header_json["patch"];
1442 : :
1443 : 23 : impl_->info.version =
1444 [ + - + - : 46 : std::to_string(version_major) + "." + std::to_string(version_minor) + "." + std::to_string(version_patch);
+ - + - +
- + - +
- ]
1445 [ + - ]: 23 : impl_->info.storage_type = "vcap";
1446 [ + - + - ]: 23 : impl_->info.message_count = header_json["count"];
1447 [ + - + - ]: 23 : impl_->info.total_duration = header_json["duration"];
1448 : 23 : impl_->info.total_duration /= 1000U;
1449 [ + - + - ]: 23 : impl_->info.time_accuracy = header_json["accuracy"];
1450 [ + - + - ]: 23 : impl_->info.compression_type = header_json["compress"];
1451 [ + - + - ]: 23 : impl_->info.process_name = header_json["process"];
1452 [ + - + - ]: 23 : impl_->info.date_time = header_json["date"];
1453 : :
1454 [ + - + + ]: 23 : if (header_json.contains("start_timestamp")) {
1455 [ + - + - ]: 22 : impl_->info.start_timestamp = header_json["start_timestamp"];
1456 : : } else {
1457 : 1 : impl_->info.start_timestamp = Helpers::convert_date_to_timestamp(impl_->info.date_time) / 1000'000;
1458 : : }
1459 : :
1460 [ + + ]: 23 : if VUNLIKELY (impl_->info.start_timestamp < 0) {
1461 : 1 : impl_->info.start_timestamp = 0;
1462 : :
1463 [ + - ]: 1 : if (impl_->read_only) {
1464 [ + - + - ]: 2 : VLOG_E("VCAPReader: Invalid start_timestamp.");
1465 : : }
1466 : : }
1467 : :
1468 [ + - + + ]: 23 : if (header_json.contains("tag")) {
1469 [ + - + - ]: 22 : impl_->info.tag_name = header_json["tag"];
1470 : : } else {
1471 [ + - ]: 1 : impl_->info.tag_name = "Empty";
1472 : : }
1473 : :
1474 [ + - + + ]: 23 : if (header_json.contains("complete")) {
1475 [ + - + - ]: 22 : impl_->info.has_completed = header_json["complete"];
1476 : : } else {
1477 : 1 : impl_->info.has_completed = true;
1478 : : }
1479 : :
1480 [ + - + + ]: 23 : if (header_json.contains("timezone")) {
1481 [ + - + - ]: 22 : impl_->info.timezone = header_json["timezone"];
1482 : : } else {
1483 : 1 : impl_->info.timezone = 480;
1484 : : }
1485 : :
1486 [ + - + + ]: 23 : if (header_json.contains("split_by_size")) {
1487 [ + - + - ]: 22 : impl_->info.split_by_size = header_json["split_by_size"];
1488 : : }
1489 : :
1490 [ + - + + ]: 23 : if (header_json.contains("split_by_time")) {
1491 [ + - + - ]: 22 : impl_->info.split_by_time = header_json["split_by_time"];
1492 : : }
1493 : :
1494 : 23 : impl_->info.blank_duration = blank_duration;
1495 : :
1496 [ + + + + ]: 49 : if (impl_->info.compression_type.empty() || impl_->info.compression_type == "None" ||
1497 [ + - + + : 49 : impl_->info.compression_type == "NONE" || impl_->info.compression_type == "none") {
+ + ]
1498 : 22 : impl_->enable_compress = false;
1499 : : } else {
1500 : 1 : impl_->enable_compress = true;
1501 : : }
1502 : :
1503 [ + + ]: 23 : if VUNLIKELY (impl_->info.time_accuracy != "MicroSecond") {
1504 [ + - - + ]: 2 : VLOG_F("VCAPReader: MCAP accuracy is not supported.");
1505 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1506 : : }
1507 : :
1508 [ + - ]: 22 : url_ser_map().clear();
1509 [ + - ]: 22 : url_schema_type_map().clear();
1510 : 22 : impl_->info.url_metas.clear();
1511 : 22 : impl_->raw_url_metas.clear();
1512 : :
1513 [ + - ]: 22 : impl_->info.url_metas.reserve(urls_json.size());
1514 : :
1515 : 22 : impl_->info.total_raw_size = 0;
1516 : :
1517 [ + - + - : 44 : for (const auto& url_info : urls_json) {
+ - + + ]
1518 : 22 : Info::UrlMeta url_meta;
1519 : :
1520 : 22 : url_meta.valid = true;
1521 [ + - + - ]: 22 : url_meta.index = url_info["index"];
1522 [ + - + - ]: 22 : url_meta.url = url_info["url"];
1523 [ + - + - ]: 22 : url_meta.url_type = url_info["type"];
1524 : :
1525 [ + - + + ]: 22 : if (url_info.contains("action")) {
1526 [ + - + - : 21 : url_meta.action_type = convert_action(url_info["action"].get<std::string>());
+ - ]
1527 : : }
1528 : :
1529 [ + - + - ]: 22 : url_meta.ser_type = url_info["ser"];
1530 : :
1531 [ + - + + ]: 22 : if (url_info.contains("encoding")) {
1532 : 42 : url_meta.schema_type = SchemaData::resolve_type(
1533 [ + - + - ]: 42 : SchemaData::convert_encoding(url_info["encoding"].get<std::string>()), url_meta.ser_type);
1534 : : } else {
1535 : 1 : url_meta.schema_type = SchemaType::kUnknown;
1536 : : }
1537 : :
1538 [ + - + - ]: 22 : url_meta.count = url_info["count"];
1539 [ + - + - ]: 22 : url_meta.loss = url_info["loss"];
1540 : :
1541 [ + - + + ]: 22 : if (url_info.contains("size")) {
1542 [ + - + - ]: 21 : url_meta.size = url_info["size"];
1543 : : }
1544 : :
1545 [ + - + + ]: 22 : if (url_info.contains("freq")) {
1546 [ + - + - ]: 21 : url_meta.freq = url_info["freq"];
1547 : : }
1548 : :
1549 : 22 : impl_->info.total_raw_size += url_meta.size;
1550 : :
1551 [ + - + - ]: 22 : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
1552 [ + - + - ]: 22 : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
1553 [ + - ]: 22 : impl_->info.url_metas.emplace_back(std::move(url_meta));
1554 : 22 : }
1555 : :
1556 [ + - ]: 22 : std::sort(impl_->info.url_metas.begin(), impl_->info.url_metas.end());
1557 [ + - ]: 22 : impl_->raw_url_metas = impl_->info.url_metas;
1558 [ + - ]: 22 : rebuild_url_meta_lookup(impl_->info.url_metas);
1559 [ + - + - : 45 : } catch (nlohmann::json::exception& e) {
+ - + - +
- + - +
+ ]
1560 [ + - - + ]: 2 : VLOG_F("VCAPReader: JSON parse error, ", e.what(), ".");
1561 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1562 : 1 : }
1563 : : } else {
1564 [ + - ]: 40 : Impl::WrapperFile wrapper_file;
1565 [ + - ]: 40 : wrapper_file.path = impl_->path;
1566 : :
1567 [ + + - + ]: 40 : if VUNLIKELY (!to_open(wrapper_file)) {
1568 : : CLOG_F("VCAPReader: Failed to prepare vcap [%s].", path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1569 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1570 : : }
1571 : :
1572 : 38 : impl_->info.file_size = 0;
1573 : :
1574 : 38 : std::error_code single_size_ec;
1575 : 38 : std::uintmax_t file_size = std::filesystem::file_size(file_path, single_size_ec);
1576 : :
1577 [ - + ]: 38 : if VUNLIKELY (single_size_ec) {
1578 : : CLOG_W("VCAPReader: file_size failed for [%s]: %s.", path.c_str(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1579 : : single_size_ec.message().c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1580 : : file_size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1581 : : }
1582 : :
1583 : 38 : impl_->info.file_size += file_size;
1584 : :
1585 : 38 : wrapper_file.start_timestamp_ns = impl_->info.start_timestamp * 1000'000;
1586 : 38 : wrapper_file.begin = impl_->info.blank_duration;
1587 : 38 : wrapper_file.end = impl_->info.total_duration;
1588 : :
1589 [ + - ]: 38 : if (impl_->total_start_timestamp_ns < 0) {
1590 : 38 : impl_->total_start_timestamp_ns = wrapper_file.start_timestamp_ns;
1591 : : }
1592 : :
1593 [ + + ]: 38 : if (!wrapper_file.has_completed) {
1594 : 1 : impl_->total_has_completed = false;
1595 : : }
1596 : :
1597 [ + - ]: 38 : impl_->file_list.emplace_back(std::move(wrapper_file));
1598 [ + - ]: 40 : }
1599 [ + - + - : 87 : } catch (std::filesystem::filesystem_error& e) {
+ - + - ]
1600 : : VLOG_F("VCAPReader: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1601 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1602 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1603 : : }
1604 : :
1605 : 127 : void VCAPReader::close() {
1606 : 127 : impl_->cursor_iter.reset();
1607 : 127 : impl_->cursor_iter_end.reset();
1608 : 127 : impl_->cursor_msg_view.reset();
1609 : :
1610 [ + + ]: 211 : for (auto& wrapper_file : impl_->file_list) {
1611 [ + - ]: 84 : if (wrapper_file.reader) {
1612 [ + - ]: 84 : wrapper_file.reader->close();
1613 : 84 : wrapper_file.reader.reset();
1614 : : }
1615 : : }
1616 : :
1617 : 127 : impl_->file_list.clear();
1618 : 127 : }
1619 : :
1620 : 14 : int VCAPReader::get_reset_index(const Config& config) {
1621 : 14 : impl_->is_pending.store(true, std::memory_order_relaxed);
1622 : :
1623 : : // LCOV_EXCL_START GCOVR_EXCL_START
1624 : : auto status_function = [](const mcap::Status& status) {
1625 : : if (!status.ok()) {
1626 : : CLOG_W("VCAPReader: Failed to read message, error = %s.", status.message.c_str());
1627 : : }
1628 : : };
1629 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1630 : :
1631 : 48 : auto filter_function = [this, &config](std::string_view url) -> bool {
1632 : 48 : return match_playback_url_filter(url, config.filter_urls);
1633 : 14 : };
1634 : :
1635 : 14 : int start_index = -1;
1636 : :
1637 : 14 : int64_t last_time = impl_->begin_time.load(std::memory_order_relaxed);
1638 : :
1639 : 14 : mcap::ReadMessageOptions read_options;
1640 : :
1641 [ + + ]: 28 : for (auto& wrapper_file : impl_->file_list) {
1642 [ + - + - : 42 : if (start_index < 0 && impl_->begin_time.load(std::memory_order_relaxed) >= last_time &&
+ - ]
1643 [ + - ]: 28 : impl_->begin_time.load(std::memory_order_relaxed) <= wrapper_file.end) {
1644 [ + + ]: 28 : if (impl_->begin_time.load(std::memory_order_relaxed) > 0) {
1645 : 2 : read_options.startTime =
1646 : 4 : impl_->begin_time.load(std::memory_order_relaxed) * 1000'000 + impl_->total_start_timestamp_ns;
1647 : 2 : read_options.endTime = mcap::MaxTime;
1648 : 2 : read_options.topicFilter = filter_function;
1649 : 2 : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
1650 : : } else {
1651 : 12 : read_options.startTime = 0;
1652 : 12 : read_options.endTime = mcap::MaxTime;
1653 : 12 : read_options.topicFilter = filter_function;
1654 : 12 : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
1655 : : }
1656 : :
1657 : 14 : start_index = wrapper_file.index;
1658 : : } else {
1659 : : // LCOV_EXCL_START GCOVR_EXCL_START
1660 : : read_options.startTime = 0;
1661 : : read_options.endTime = mcap::MaxTime;
1662 : : read_options.topicFilter = filter_function;
1663 : : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
1664 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1665 : : }
1666 : :
1667 : 14 : const auto [start_offset, end_offset] =
1668 [ + - ]: 14 : wrapper_file.reader->byteRange(read_options.startTime, read_options.endTime);
1669 : :
1670 : : // NOLINTNEXTLINE(readability-redundant-smartptr-get)
1671 [ + - ]: 28 : wrapper_file.msg_view = std::make_unique<mcap::LinearMessageView>(*wrapper_file.reader.get(), read_options,
1672 : 14 : start_offset, end_offset, status_function);
1673 : :
1674 [ - + ]: 14 : if (start_offset == end_offset) {
1675 : : wrapper_file.msg_view_begin.emplace(wrapper_file.msg_view->end()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1676 : : wrapper_file.msg_view_end.emplace(wrapper_file.msg_view->end()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1677 : : } else {
1678 [ + - ]: 14 : wrapper_file.msg_view_begin.emplace(wrapper_file.msg_view->begin());
1679 [ + - ]: 14 : wrapper_file.msg_view_end.emplace(wrapper_file.msg_view->end());
1680 : : }
1681 : :
1682 : 14 : last_time = wrapper_file.end;
1683 : : }
1684 : :
1685 : 14 : impl_->is_pending.store(false, std::memory_order_relaxed);
1686 : :
1687 : 14 : return start_index;
1688 : 14 : }
1689 : :
1690 : 51 : bool VCAPReader::prepare_cursor_view(int file_index) {
1691 : 51 : impl_->cursor_iter.reset();
1692 : 51 : impl_->cursor_iter_end.reset();
1693 : 51 : impl_->cursor_msg_view.reset();
1694 : 51 : impl_->cursor_need_advance = false;
1695 : :
1696 [ + - - + : 51 : if VUNLIKELY (file_index < 0 || file_index >= static_cast<int>(impl_->file_list.size())) {
- + ]
1697 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1698 : : }
1699 : :
1700 [ + - ]: 51 : auto& wrapper_file = impl_->file_list.at(file_index);
1701 : :
1702 [ - + ]: 51 : if VUNLIKELY (!wrapper_file.reader) {
1703 : : VLOG_W("VCAPReader: Cursor target vcap reader is empty."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1704 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1705 : : }
1706 : :
1707 : 0 : auto status_function = [this](const mcap::Status& status) {
1708 : : // LCOV_EXCL_START GCOVR_EXCL_START
1709 : : if (!status.ok()) {
1710 : : CLOG_W("VCAPReader: Failed to read cursor message, error = %s.", status.message.c_str());
1711 : : impl_->cursor_read_error = true;
1712 : : }
1713 : : };
1714 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1715 : :
1716 : 72 : auto filter_function = [this](std::string_view url) -> bool {
1717 : 72 : return match_playback_url_filter(url, impl_->cursor_config.filter_urls);
1718 : 51 : };
1719 : :
1720 : 51 : mcap::ReadMessageOptions read_options;
1721 : 51 : read_options.startTime =
1722 [ + + ]: 51 : impl_->cursor_begin_us > 0 ? impl_->cursor_begin_us * 1000 + impl_->total_start_timestamp_ns : 0;
1723 : 51 : read_options.endTime = mcap::MaxTime;
1724 : 51 : read_options.topicFilter = filter_function;
1725 : 51 : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
1726 : :
1727 [ + - ]: 51 : const auto [start_offset, end_offset] = wrapper_file.reader->byteRange(read_options.startTime, read_options.endTime);
1728 : :
1729 : : // NOLINTNEXTLINE(readability-redundant-smartptr-get)
1730 [ + - ]: 102 : impl_->cursor_msg_view = std::make_unique<mcap::LinearMessageView>(*wrapper_file.reader.get(), read_options,
1731 : 51 : start_offset, end_offset, status_function);
1732 : :
1733 [ + + ]: 51 : if (start_offset == end_offset) {
1734 [ + - ]: 1 : impl_->cursor_iter.emplace(impl_->cursor_msg_view->end());
1735 : : } else {
1736 [ + - ]: 50 : impl_->cursor_iter.emplace(impl_->cursor_msg_view->begin());
1737 : : }
1738 : :
1739 [ + - ]: 51 : impl_->cursor_iter_end.emplace(impl_->cursor_msg_view->end());
1740 : 51 : impl_->cursor_file_index = file_index;
1741 : :
1742 : 51 : return true;
1743 : 51 : }
1744 : :
1745 : 41 : bool VCAPReader::do_open_cursor(const Config& config) {
1746 : 41 : impl_->cursor_config = config;
1747 [ + + ]: 41 : impl_->cursor_begin_us = config.begin_time > 0 ? config.begin_time * 1000 : 0;
1748 [ + + ]: 41 : impl_->cursor_end_us = config.end_time > 0 ? config.end_time * 1000 : 0;
1749 : 41 : impl_->cursor_file_index = 0;
1750 : 41 : impl_->cursor_read_error = false;
1751 : :
1752 [ - + ]: 41 : if VUNLIKELY (impl_->file_list.empty()) {
1753 : : VLOG_W("VCAPReader: Cursor cannot find any data."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1754 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1755 : : }
1756 : :
1757 : 41 : return prepare_cursor_view(0);
1758 : : }
1759 : :
1760 : 95 : bool VCAPReader::do_read_next(Frame& out, bool& is_error) {
1761 : 95 : is_error = false;
1762 : :
1763 : : while (true) {
1764 [ + - - + : 105 : if VUNLIKELY (!impl_->cursor_iter.has_value() || !impl_->cursor_iter_end.has_value()) {
- + ]
1765 : 95 : return false;
1766 : : }
1767 : :
1768 [ + - ]: 105 : auto& iter = impl_->cursor_iter.value();
1769 [ + - ]: 105 : auto& iter_end = impl_->cursor_iter_end.value();
1770 : :
1771 [ + + ]: 105 : if (impl_->cursor_need_advance) {
1772 [ + - + - ]: 54 : if (iter != iter_end) {
1773 [ + - ]: 54 : iter++;
1774 : : }
1775 : :
1776 : 54 : impl_->cursor_need_advance = false;
1777 : : }
1778 : :
1779 [ + - + + ]: 105 : if (iter == iter_end) {
1780 [ - + ]: 47 : if VUNLIKELY (impl_->cursor_read_error) {
1781 : : is_error = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1782 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1783 : : }
1784 : :
1785 [ + + ]: 47 : if (impl_->cursor_file_index + 1 < static_cast<int>(impl_->file_list.size())) {
1786 [ + - - + ]: 10 : if VUNLIKELY (!prepare_cursor_view(impl_->cursor_file_index + 1)) {
1787 : : is_error = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1788 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1789 : : }
1790 : :
1791 : 10 : continue;
1792 : : }
1793 : :
1794 : 37 : return false;
1795 : : }
1796 : :
1797 [ + - ]: 58 : const int64_t timestamp = (iter->message.logTime - impl_->total_start_timestamp_ns) / 1000;
1798 : :
1799 [ + + - + : 58 : if (impl_->cursor_begin_us > 0 && timestamp < impl_->cursor_begin_us) {
- + ]
1800 : : iter++; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1801 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1802 : : }
1803 : :
1804 [ + + + + : 58 : if (impl_->cursor_end_us > 0 && timestamp > impl_->cursor_end_us) {
+ + ]
1805 : 4 : return false;
1806 : : }
1807 : :
1808 [ + - - + ]: 54 : if VUNLIKELY (iter->message.dataSize > static_cast<uint64_t>(std::numeric_limits<size_t>::max())) {
1809 : : // LCOV_EXCL_START GCOVR_EXCL_START
1810 : : CLOG_W("VCAPReader: Cursor message data size is too large to address, size = %" PRIu64 ".",
1811 : : static_cast<uint64_t>(iter->message.dataSize));
1812 : : iter++;
1813 : : continue;
1814 : : }
1815 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1816 : :
1817 : 54 : std::string output_url;
1818 : :
1819 [ + - + - : 54 : if VUNLIKELY (!convert_playback_url(iter->channel->topic, output_url)) {
- + ]
1820 : : iter++; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1821 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1822 : : }
1823 : :
1824 : 54 : ActionType action_type = ActionType::kUnknownAction;
1825 [ + - ]: 54 : auto& wrapper_file = impl_->file_list.at(impl_->cursor_file_index);
1826 : :
1827 [ + - + - ]: 54 : if (auto action_iter = wrapper_file.channel_action_map.find(iter->message.channelId);
1828 [ + - ]: 54 : action_iter != wrapper_file.channel_action_map.end()) {
1829 : 54 : action_type = action_iter->second;
1830 : : }
1831 : :
1832 [ + - ]: 54 : const auto* data = reinterpret_cast<const uint8_t*>(iter->message.data);
1833 [ + - ]: 54 : const auto size = static_cast<size_t>(iter->message.dataSize);
1834 : :
1835 : 54 : out.timestamp = timestamp;
1836 : 54 : out.url = std::move(output_url);
1837 : 54 : out.ser_type.clear();
1838 : 54 : out.schema_type = SchemaType::kUnknown;
1839 : 54 : out.action_type = action_type;
1840 : 54 : out.data = Bytes::shallow_copy(data, size);
1841 : :
1842 [ + - ]: 54 : fill_frame_meta(out);
1843 : :
1844 : 54 : impl_->cursor_need_advance = true;
1845 : :
1846 : 54 : return true;
1847 [ + - ]: 64 : }
1848 : : }
1849 : :
1850 : 13 : void VCAPReader::read(const Config& config) {
1851 : 13 : int loop_times = 0;
1852 : :
1853 [ + + ]: 13 : if (config.auto_pause) {
1854 : 1 : impl_->pause_flag.store(true, std::memory_order_relaxed);
1855 : : }
1856 : :
1857 : 13 : bool is_interrupted = false;
1858 : :
1859 : : do {
1860 : 14 : bool is_end = false;
1861 : :
1862 : : // prepare
1863 [ + - ]: 14 : int start_index = get_reset_index(config);
1864 : :
1865 [ + + ]: 14 : if (impl_->ready_callback) {
1866 [ + - ]: 8 : impl_->ready_callback();
1867 : : }
1868 : :
1869 [ + - - + : 14 : if VUNLIKELY (start_index < 0 || start_index > static_cast<int>(impl_->file_list.size()) - 1) {
- + ]
1870 : : // LCOV_EXCL_START GCOVR_EXCL_START
1871 : : VLOG_W("VCAPReader: Cannot find any data for play.");
1872 : :
1873 : : update_status(kStopped);
1874 : :
1875 : : if (config.auto_quit) {
1876 : : quit();
1877 : : }
1878 : :
1879 : : return;
1880 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1881 : : }
1882 : :
1883 : : {
1884 [ + - ]: 14 : std::lock_guard time_lock(impl_->time_mtx);
1885 : 14 : impl_->pause_elapsed.store(0, std::memory_order_relaxed);
1886 : 14 : impl_->offset_elapsed.store(0, std::memory_order_relaxed);
1887 : 28 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
1888 : :
1889 : 14 : impl_->elapsed_timer.restart();
1890 : 14 : impl_->pause_elapsed_timer.restart();
1891 : 14 : impl_->offset_timer.restart();
1892 : 14 : impl_->real_timer.restart();
1893 : 14 : }
1894 : :
1895 [ + + ]: 14 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
1896 : 1 : is_interrupted = true;
1897 [ + - ]: 1 : update_status(kStopped);
1898 : 5 : break;
1899 [ + + ]: 13 : } else if (impl_->jump_flag.load(std::memory_order_relaxed)) {
1900 : 2 : break;
1901 [ + + ]: 11 : } else if (impl_->pause_flag.load(std::memory_order_relaxed)) {
1902 : 1 : impl_->pause_elapsed_timer.restart();
1903 [ + - ]: 1 : update_status(kPaused);
1904 [ + - ]: 1 : do_pause();
1905 : : // impl_->pause_next_flag = false;
1906 : : {
1907 [ + - ]: 1 : std::lock_guard time_lock(impl_->time_mtx);
1908 : 1 : impl_->pause_elapsed.store(0, std::memory_order_relaxed);
1909 : 1 : impl_->offset_elapsed.store(0, std::memory_order_relaxed);
1910 : 2 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
1911 : :
1912 : 1 : impl_->elapsed_timer.restart();
1913 : 1 : impl_->pause_elapsed_timer.restart();
1914 : 1 : impl_->offset_timer.restart();
1915 : 1 : impl_->real_timer.restart();
1916 : 1 : }
1917 : :
1918 : : } else {
1919 [ + - ]: 10 : update_status(kPlaying);
1920 : : }
1921 : :
1922 : 11 : int64_t elapsed = 0;
1923 : 11 : int64_t timestamp = 0;
1924 : 11 : int64_t last_timestamp = 0;
1925 : 11 : const uint8_t* data = nullptr;
1926 : 11 : size_t size = 0;
1927 : :
1928 : : // process files
1929 [ + + ]: 20 : for (int index = start_index; index < static_cast<int>(impl_->file_list.size()); ++index) {
1930 : 11 : impl_->split_index.store(index, std::memory_order_relaxed);
1931 : :
1932 [ + - ]: 22 : auto& wrapper_file = impl_->file_list.at(impl_->split_index.load(std::memory_order_relaxed));
1933 : :
1934 [ - + ]: 11 : if VUNLIKELY (!wrapper_file.reader) {
1935 : : VLOG_W("VCAPReader: Target vcap reader is empty."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1936 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1937 : : }
1938 : :
1939 : : // process datas
1940 [ + - ]: 11 : auto iter = std::move(wrapper_file.msg_view_begin).value(); // NOLINT
1941 [ + - ]: 11 : auto iter_end = std::move(wrapper_file.msg_view_end).value(); // NOLINT
1942 : :
1943 [ + - + - : 46 : for (; iter != iter_end; iter++) {
+ + ]
1944 [ + - ]: 37 : timestamp = (iter->message.logTime - impl_->total_start_timestamp_ns) / 1000;
1945 : :
1946 [ - + ]: 37 : if VUNLIKELY (last_timestamp > timestamp + 10'000U) {
1947 : : VLOG_W("VCAPReader: The vcap timestamp is incorrect."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1948 : : }
1949 : :
1950 : 37 : last_timestamp = timestamp;
1951 : :
1952 [ - + ]: 74 : if (timestamp < impl_->begin_time.load(std::memory_order_relaxed) * 1000U) {
1953 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1954 : : }
1955 : :
1956 [ + + - + ]: 37 : if (config.end_time > 0 && timestamp > config.end_time * 1000U) {
1957 : : timestamp = config.end_time * 1000U; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1958 : : is_end = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1959 : : }
1960 : :
1961 [ + - ]: 37 : data = reinterpret_cast<const uint8_t*>(iter->message.data);
1962 : :
1963 [ + - - + ]: 37 : if VUNLIKELY (iter->message.dataSize > static_cast<uint64_t>(std::numeric_limits<size_t>::max())) {
1964 : : // LCOV_EXCL_START GCOVR_EXCL_START
1965 : : CLOG_W("VCAPReader: Message data size is too large to address, size = %" PRIu64 ".",
1966 : : static_cast<uint64_t>(iter->message.dataSize));
1967 : : continue;
1968 : : }
1969 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1970 : :
1971 [ + - ]: 37 : size = static_cast<size_t>(iter->message.dataSize);
1972 : :
1973 : 37 : elapsed =
1974 : 37 : (timestamp / impl_->rate.load(std::memory_order_relaxed)) -
1975 : 37 : (impl_->elapsed_timer.get() - impl_->pause_elapsed.load(std::memory_order_relaxed)) -
1976 : 74 : (impl_->begin_time.load(std::memory_order_relaxed) * 1000U / impl_->rate.load(std::memory_order_relaxed));
1977 : :
1978 : : {
1979 [ + - ]: 37 : std::unique_lock lock(impl_->mtx);
1980 : :
1981 [ + + ]: 37 : if (config.force_delay > 0) {
1982 : 25 : impl_->cv.wait_for(lock, std::chrono::milliseconds(config.force_delay), [this]() -> bool {
1983 [ + - ]: 101 : return impl_->stop_flag.load(std::memory_order_relaxed) ||
1984 [ + + ]: 100 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
1985 [ + + - + ]: 151 : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
1986 : : });
1987 [ + + + - ]: 12 : } else if (config.force_delay < 0 && elapsed > 0) {
1988 : 1 : impl_->offset_timer.restart();
1989 : :
1990 : 1 : impl_->cv.wait_for(lock, std::chrono::microseconds(elapsed), [this]() -> bool {
1991 [ + - ]: 4 : return impl_->stop_flag.load(std::memory_order_relaxed) ||
1992 [ + - ]: 4 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
1993 [ + - ]: 4 : impl_->jump_flag.load(std::memory_order_relaxed) ||
1994 [ + - - + ]: 6 : impl_->pause_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
1995 : : });
1996 : :
1997 [ - + ]: 1 : if VUNLIKELY (impl_->pause_flag.load(std::memory_order_relaxed)) {
1998 : : impl_->offset_elapsed.store(elapsed - impl_->offset_timer.get(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1999 : : std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2000 : : }
2001 : : }
2002 : 37 : }
2003 : :
2004 [ + + + + : 72 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
+ + ]
2005 [ + - - + ]: 35 : is_ready_to_quit()) {
2006 : 2 : is_interrupted = true;
2007 : 2 : break;
2008 [ + + ]: 35 : } else if (impl_->pause_flag.load(std::memory_order_relaxed)) {
2009 [ + - ]: 3 : do_pause();
2010 : 3 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
2011 : :
2012 [ + - + - : 6 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
- + ]
2013 [ + - - + ]: 3 : is_ready_to_quit()) {
2014 : : is_interrupted = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2015 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2016 : : }
2017 : : }
2018 : :
2019 : : {
2020 [ + - ]: 35 : std::lock_guard time_lock(impl_->time_mtx);
2021 : 35 : impl_->real_timer.restart();
2022 : 35 : impl_->real_elapsed.store(timestamp, std::memory_order_relaxed);
2023 : 35 : }
2024 : :
2025 [ - + ]: 35 : if (is_end) {
2026 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2027 : : }
2028 : :
2029 : 35 : ActionType action_type = ActionType::kUnknownAction;
2030 : :
2031 [ + - + - ]: 35 : if (auto action_iter = wrapper_file.channel_action_map.find(iter->message.channelId);
2032 [ + - ]: 35 : action_iter != wrapper_file.channel_action_map.end()) {
2033 : 35 : action_type = action_iter->second;
2034 : : }
2035 : :
2036 : 35 : Frame frame;
2037 : 35 : frame.timestamp = timestamp;
2038 [ + - + - ]: 35 : frame.url = iter->channel->topic;
2039 : 35 : frame.action_type = action_type;
2040 : 35 : frame.data = Bytes::shallow_copy(data, size);
2041 : :
2042 [ + - ]: 35 : BagReader::process_output(frame);
2043 : 35 : }
2044 : :
2045 [ + + - + ]: 11 : if (is_interrupted || is_end) {
2046 : : break;
2047 : : }
2048 [ + + + + ]: 13 : }
2049 : :
2050 [ + + ]: 11 : if (is_interrupted) {
2051 [ + + ]: 2 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
2052 [ + - ]: 1 : update_status(kStopped);
2053 : : }
2054 : :
2055 : 2 : break;
2056 : : }
2057 : :
2058 [ + - ]: 9 : if (!impl_->jump_flag.load(std::memory_order_relaxed)) {
2059 [ + - ]: 9 : update_status(kStopped);
2060 : :
2061 [ + + ]: 9 : if (config.skip_blank) {
2062 : 1 : impl_->begin_time.store(std::max(config.begin_time, impl_->info.blank_duration), std::memory_order_relaxed);
2063 : : } else {
2064 : 8 : impl_->begin_time.store(config.begin_time, std::memory_order_relaxed);
2065 : : }
2066 : : }
2067 [ + - + + ]: 27 : } while (impl_->times.load(std::memory_order_relaxed) <= 0 ||
2068 [ + - ]: 18 : (impl_->times.load(std::memory_order_relaxed) > 0 &&
2069 [ + + ]: 18 : ++loop_times < impl_->times.load(std::memory_order_relaxed)));
2070 : :
2071 [ + + ]: 13 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
2072 : 2 : is_interrupted = true;
2073 : : }
2074 : :
2075 [ + + + + : 13 : if (!impl_->jump_flag.load(std::memory_order_relaxed) && !is_interrupted) {
+ + ]
2076 : 8 : flush_plugin();
2077 : : }
2078 : :
2079 [ + + + + : 13 : if (!impl_->jump_flag.load(std::memory_order_relaxed) && impl_->finish_callback) {
+ + ]
2080 : 9 : impl_->finish_callback(is_interrupted);
2081 : : }
2082 : :
2083 [ + + + + : 13 : if (!impl_->jump_flag.load(std::memory_order_relaxed) && config.auto_quit) {
+ + ]
2084 : 1 : quit();
2085 : : }
2086 : :
2087 : : // clean msg_view
2088 [ + + ]: 26 : for (auto& wrapper_file : impl_->file_list) {
2089 : 13 : wrapper_file.msg_view.reset();
2090 : : }
2091 : : }
2092 : :
2093 : : } // namespace vlink
|
