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