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/bag_reader.h"
25 : :
26 : : #include <algorithm>
27 : : #include <memory>
28 : : #include <shared_mutex>
29 : : #include <string>
30 : : #include <unordered_map>
31 : : #include <unordered_set>
32 : : #include <utility>
33 : :
34 : : #include "./base/helpers.h"
35 : : #include "./base/logger.h"
36 : : #include "./extension/bag_plugin_interface.h"
37 : : #include "./extension/vcap_reader.h"
38 : : #include "./extension/vdb_reader.h"
39 : : #include "./impl/url.h"
40 : :
41 : : namespace vlink {
42 : :
43 : : // UrlMeta
44 : 209 : bool BagReader::Info::UrlMeta::operator<(const BagReader::Info::UrlMeta& target) const noexcept {
45 : 209 : int lindex = Url::get_sort_index(url);
46 : 209 : int rindex = Url::get_sort_index(target.url);
47 : :
48 [ + + ]: 209 : if (lindex < rindex) {
49 : 1 : return true;
50 [ + + ]: 208 : } else if (lindex > rindex) {
51 : 3 : return false;
52 [ + + ]: 205 : } else if (url < target.url) {
53 : 16 : return true;
54 [ + + ]: 189 : } else if (url > target.url) {
55 : 178 : return false;
56 : : }
57 : :
58 : 11 : return index < target.index;
59 : : }
60 : :
61 : : // BagReader::Impl
62 : : struct BagReader::Impl final {
63 : : BagReader::OutputCallback output_callback;
64 : : std::shared_ptr<BagPluginInterface> plugin_interface;
65 : : std::unordered_map<std::string, std::string> playback_url_remap;
66 : : std::unordered_set<std::string> excluded_playback_urls;
67 : : std::unordered_map<std::string, std::string> url_to_ser_map;
68 : : std::unordered_map<std::string, SchemaType> url_to_schema_type_map;
69 : : std::atomic_bool playback_url_rules_enabled{false};
70 : : mutable std::shared_mutex playback_state_mtx;
71 : : mutable std::shared_mutex output_callback_mtx;
72 : :
73 : : bool cursor_opened{false};
74 : : bool cursor_eof{false};
75 : : bool cursor_fail{false};
76 : : };
77 : :
78 : : // BagReader
79 : 188 : std::shared_ptr<BagReader> BagReader::create(const std::string& path, bool read_only, bool try_to_fix) {
80 [ + - ]: 188 : std::string suffix_check = path;
81 : :
82 : 188 : std::transform(suffix_check.begin(), suffix_check.end(), suffix_check.begin(),
83 : 14818 : [](unsigned char c) { return std::tolower(c); });
84 : :
85 [ + + + + : 188 : if (Helpers::has_endwith(suffix_check, ".vdb") || Helpers::has_endwith(suffix_check, ".vdbx")) {
+ + ]
86 [ + + ]: 109 : return std::make_shared<VDBReader>(path, read_only, try_to_fix);
87 [ + + + + : 79 : } else if (Helpers::has_endwith(suffix_check, ".vcap") || Helpers::has_endwith(suffix_check, ".vcapx")) {
+ + ]
88 [ + + ]: 78 : return std::make_shared<VCAPReader>(path, read_only, try_to_fix);
89 : : } else {
90 [ + - + - ]: 2 : CLOG_E("BagReader: Unknown bag suffix, path=%s", path.c_str());
91 : 1 : return nullptr;
92 : : }
93 : 188 : }
94 : :
95 [ + - ]: 206 : BagReader::BagReader(const std::string& path, bool read_only, bool try_to_fix) : impl_(std::make_unique<Impl>()) {
96 : : (void)path;
97 : : (void)read_only;
98 : : (void)try_to_fix;
99 : :
100 : 206 : Bytes::init_memory_pool();
101 : 206 : }
102 : :
103 : 206 : BagReader::~BagReader() {
104 : 206 : std::shared_ptr<BagPluginInterface> plugin_interface;
105 : :
106 : : {
107 : 206 : std::shared_lock state_lock(impl_->playback_state_mtx);
108 : 206 : plugin_interface = impl_->plugin_interface;
109 : 206 : }
110 : :
111 [ + + ]: 206 : if (plugin_interface) {
112 : 9 : plugin_interface->register_callback({});
113 : : }
114 : 206 : }
115 : :
116 : 170 : void BagReader::detach_plugin() {
117 : 170 : std::shared_ptr<BagPluginInterface> plugin_interface;
118 : :
119 : : {
120 [ + - ]: 170 : std::unique_lock state_lock(impl_->playback_state_mtx);
121 : 170 : plugin_interface = std::move(impl_->plugin_interface);
122 : 170 : }
123 : :
124 [ + + ]: 170 : if (plugin_interface) {
125 [ + - ]: 12 : plugin_interface->flush();
126 : 12 : plugin_interface->register_callback({});
127 : : }
128 : 170 : }
129 : :
130 : 44 : void BagReader::reset_plugin() {
131 : 44 : std::shared_ptr<BagPluginInterface> plugin_interface;
132 : :
133 : : {
134 [ + - ]: 44 : std::shared_lock state_lock(impl_->playback_state_mtx);
135 : 44 : plugin_interface = impl_->plugin_interface;
136 : 44 : }
137 : :
138 [ + + ]: 44 : if (plugin_interface) {
139 [ + - ]: 18 : plugin_interface->on_reset();
140 : : }
141 : 44 : }
142 : :
143 : 31 : void BagReader::flush_plugin() {
144 : 31 : std::shared_ptr<BagPluginInterface> plugin_interface;
145 : :
146 : : {
147 [ + - ]: 31 : std::shared_lock state_lock(impl_->playback_state_mtx);
148 : 31 : plugin_interface = impl_->plugin_interface;
149 : 31 : }
150 : :
151 [ + + ]: 31 : if (plugin_interface) {
152 [ + - ]: 13 : plugin_interface->flush();
153 : : }
154 : 31 : }
155 : :
156 : 26 : void BagReader::bind_bag_interface(const std::shared_ptr<BagPluginInterface>& bag_interface) {
157 : 26 : std::shared_ptr<BagPluginInterface> old_plugin_interface;
158 : :
159 : : {
160 [ + - ]: 26 : std::shared_lock state_lock(impl_->playback_state_mtx);
161 : 26 : old_plugin_interface = impl_->plugin_interface;
162 : 26 : }
163 : :
164 [ + + + - : 26 : if (old_plugin_interface && old_plugin_interface != bag_interface) {
+ + ]
165 [ + - ]: 3 : old_plugin_interface->flush();
166 : 3 : old_plugin_interface->register_callback({});
167 : : }
168 : :
169 [ + + ]: 26 : if VLIKELY (bag_interface) {
170 : 24 : bag_interface->bind_direction(BagPluginInterface::Direction::kRead);
171 : :
172 [ + - ]: 24 : bag_interface->register_callback([this](const Frame& frame) {
173 : 36 : std::string output_url;
174 : :
175 [ + - - + ]: 36 : if VUNLIKELY (!convert_playback_url(frame.url, output_url)) {
176 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
177 : : }
178 : :
179 [ + - ]: 36 : std::shared_lock callback_lock(impl_->output_callback_mtx);
180 : :
181 [ + - ]: 36 : if VLIKELY (impl_->output_callback) {
182 : 36 : Frame out;
183 : 36 : out.timestamp = frame.timestamp;
184 : 36 : out.url = std::move(output_url);
185 [ + - ]: 36 : out.ser_type = frame.ser_type;
186 : 36 : out.schema_type = frame.schema_type;
187 : 36 : out.action_type = frame.action_type;
188 : 36 : out.data = Bytes::shallow_copy(frame.data.data(), frame.data.size());
189 : :
190 [ + + - + : 36 : if (out.ser_type.empty() || out.schema_type == SchemaType::kUnknown) {
+ + ]
191 [ + - ]: 6 : fill_frame_meta(out);
192 : : }
193 : :
194 [ + - ]: 36 : impl_->output_callback(out);
195 : 36 : }
196 [ + - ]: 36 : });
197 : : }
198 : :
199 [ + - ]: 26 : std::unique_lock state_lock(impl_->playback_state_mtx);
200 : :
201 : 26 : impl_->plugin_interface = bag_interface;
202 : 26 : impl_->playback_url_remap.clear();
203 : 26 : impl_->excluded_playback_urls.clear();
204 : 26 : impl_->playback_url_rules_enabled.store(false, std::memory_order_release);
205 : 26 : }
206 : :
207 [ # # ]: 0 : void BagReader::clear_bag_interface() { bind_bag_interface(nullptr); }
208 : :
209 : 1 : void BagReader::register_status_callback(StatusCallback&& status_callback) { (void)status_callback; }
210 : :
211 : 1 : void BagReader::register_ready_callback(ReadyCallback&& ready_callback) { (void)ready_callback; }
212 : :
213 : 1 : void BagReader::register_finish_callback(FinishCallback&& finish_callback) { (void)finish_callback; }
214 : :
215 : 29 : void BagReader::register_output_callback(OutputCallback&& output_callback) {
216 [ + - ]: 29 : std::unique_lock lock(impl_->output_callback_mtx);
217 : :
218 : 29 : impl_->output_callback = std::move(output_callback);
219 : 29 : }
220 : :
221 : 111 : bool BagReader::open_cursor(const Config& config) {
222 : 111 : impl_->cursor_opened = false;
223 : 111 : impl_->cursor_eof = false;
224 : 111 : impl_->cursor_fail = false;
225 : :
226 [ + + ]: 111 : if VUNLIKELY (!do_open_cursor(config)) {
227 : 2 : impl_->cursor_fail = true;
228 : 2 : return false;
229 : : }
230 : :
231 : 109 : impl_->cursor_opened = true;
232 : 109 : return true;
233 : : }
234 : :
235 [ + - ]: 82 : bool BagReader::open_cursor() { return open_cursor(Config{}); }
236 : :
237 : 285 : bool BagReader::read_next(Frame& out) {
238 [ + + ]: 285 : if VUNLIKELY (impl_->cursor_fail) {
239 : 2 : return false;
240 : : }
241 : :
242 [ + + ]: 283 : if VUNLIKELY (impl_->cursor_eof) {
243 : 1 : return false;
244 : : }
245 : :
246 [ + + + - : 282 : if (!impl_->cursor_opened && !open_cursor(Config{})) {
- + + + -
+ - - ]
247 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
248 : : }
249 : :
250 : 282 : bool is_error = false;
251 : :
252 [ + - + + ]: 282 : if VLIKELY (do_read_next(out, is_error)) {
253 : 173 : return true;
254 : : }
255 : :
256 [ - + ]: 109 : if (is_error) {
257 : : impl_->cursor_fail = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
258 : : } else {
259 : 109 : impl_->cursor_eof = true;
260 : : }
261 : :
262 : 109 : return false;
263 : : }
264 : :
265 : 12 : BagReader& BagReader::operator>>(Frame& out) {
266 : 12 : read_next(out);
267 : :
268 : 12 : return *this;
269 : : }
270 : :
271 : 21 : bool BagReader::eof() const noexcept { return impl_->cursor_eof; }
272 : :
273 : 13 : bool BagReader::fail() const noexcept { return impl_->cursor_fail; }
274 : :
275 [ + + + + ]: 14 : BagReader::operator bool() const noexcept { return !impl_->cursor_eof && !impl_->cursor_fail; }
276 : :
277 : 1 : bool BagReader::do_open_cursor(const Config& config) {
278 : : (void)config;
279 : :
280 : 1 : return false;
281 : : }
282 : :
283 : : bool BagReader::do_read_next(Frame& out, bool& is_error) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
284 : : (void)out;
285 : :
286 : : is_error = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
287 : :
288 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
289 : : }
290 : :
291 : 121 : void BagReader::process_output(Frame& frame) {
292 : 121 : std::shared_ptr<BagPluginInterface> plugin_interface;
293 : :
294 : : {
295 [ + - ]: 121 : std::shared_lock state_lock(impl_->playback_state_mtx);
296 : :
297 : 121 : plugin_interface = impl_->plugin_interface;
298 : :
299 [ + + + - : 121 : if VUNLIKELY (plugin_interface && impl_->excluded_playback_urls.count(frame.url) != 0U) {
+ + + + ]
300 : 1 : return;
301 : : }
302 : :
303 : 120 : const std::string* meta_url = &frame.url;
304 : :
305 [ + + ]: 120 : if (plugin_interface) {
306 [ + - ]: 47 : auto remap_iter = impl_->playback_url_remap.find(frame.url);
307 : :
308 [ + + ]: 47 : if (remap_iter != impl_->playback_url_remap.end()) {
309 : 2 : meta_url = &remap_iter->second;
310 : : }
311 : : }
312 : :
313 [ + - ]: 120 : if (frame.ser_type.empty()) {
314 [ + - ]: 120 : auto ser_iter = impl_->url_to_ser_map.find(*meta_url);
315 : :
316 [ + + ]: 120 : if VLIKELY (ser_iter != impl_->url_to_ser_map.end()) {
317 [ + - ]: 114 : frame.ser_type = ser_iter->second;
318 : : }
319 : : }
320 : :
321 [ + - ]: 120 : if (frame.schema_type == SchemaType::kUnknown) {
322 [ + - ]: 120 : auto schema_iter = impl_->url_to_schema_type_map.find(*meta_url);
323 : :
324 [ + + ]: 120 : if VLIKELY (schema_iter != impl_->url_to_schema_type_map.end()) {
325 : 114 : frame.schema_type = schema_iter->second;
326 : : }
327 : : }
328 [ + + ]: 121 : }
329 : :
330 [ + + ]: 120 : if (plugin_interface) {
331 [ + - ]: 47 : plugin_interface->on_read(frame);
332 : : } else {
333 [ + - ]: 73 : std::shared_lock callback_lock(impl_->output_callback_mtx);
334 : :
335 [ + + ]: 73 : if VLIKELY (impl_->output_callback) {
336 [ + - ]: 72 : impl_->output_callback(frame);
337 : : }
338 : 73 : }
339 [ + + ]: 121 : }
340 : :
341 : 171 : void BagReader::fill_frame_meta(Frame& frame) const {
342 [ + - ]: 171 : std::shared_lock state_lock(impl_->playback_state_mtx);
343 : :
344 [ + - ]: 171 : if (frame.ser_type.empty()) {
345 [ + - ]: 171 : auto iter = impl_->url_to_ser_map.find(frame.url);
346 : :
347 [ + + ]: 171 : if VLIKELY (iter != impl_->url_to_ser_map.end()) {
348 [ + - ]: 166 : frame.ser_type = iter->second;
349 : : }
350 : : }
351 : :
352 [ + - ]: 171 : if (frame.schema_type == SchemaType::kUnknown) {
353 [ + - ]: 171 : auto iter = impl_->url_to_schema_type_map.find(frame.url);
354 : :
355 [ + + ]: 171 : if VLIKELY (iter != impl_->url_to_schema_type_map.end()) {
356 : 166 : frame.schema_type = iter->second;
357 : : }
358 : : }
359 : 171 : }
360 : :
361 : 843 : std::unordered_map<std::string, std::string>& BagReader::url_ser_map() { return impl_->url_to_ser_map; }
362 : :
363 : 843 : std::unordered_map<std::string, SchemaType>& BagReader::url_schema_type_map() { return impl_->url_to_schema_type_map; }
364 : :
365 : 26 : std::string BagReader::get_ser_type(const std::string& url) const {
366 [ + - ]: 26 : std::shared_lock state_lock(impl_->playback_state_mtx);
367 : :
368 [ + - ]: 26 : auto iter = impl_->url_to_ser_map.find(url);
369 : :
370 [ + + ]: 26 : if VLIKELY (iter != impl_->url_to_ser_map.end()) {
371 [ + - ]: 17 : return iter->second;
372 : : }
373 : :
374 : 9 : return {};
375 : 26 : }
376 : :
377 : 14 : SchemaType BagReader::get_schema_type(const std::string& url) const {
378 [ + - ]: 14 : std::shared_lock state_lock(impl_->playback_state_mtx);
379 : :
380 [ + - ]: 14 : auto iter = impl_->url_to_schema_type_map.find(url);
381 : :
382 [ + + ]: 14 : if VLIKELY (iter != impl_->url_to_schema_type_map.end()) {
383 : 11 : return iter->second;
384 : : }
385 : :
386 : 3 : return SchemaType::kUnknown;
387 : 14 : }
388 : :
389 : 24 : void BagReader::process_url_metas(std::vector<Info::UrlMeta>& url_metas) {
390 : 24 : std::shared_ptr<BagPluginInterface> plugin_interface;
391 : :
392 : : {
393 [ + - ]: 24 : std::shared_lock state_lock(impl_->playback_state_mtx);
394 : 24 : plugin_interface = impl_->plugin_interface;
395 : 24 : }
396 : :
397 : 24 : std::unordered_map<std::string, std::string> playback_url_remap;
398 : 24 : std::unordered_set<std::string> excluded_playback_urls;
399 : :
400 [ + + ]: 24 : if (plugin_interface) {
401 [ + - ]: 42 : url_metas.erase(
402 [ + - ]: 21 : std::remove_if(url_metas.begin(), url_metas.end(),
403 : 103 : [&plugin_interface, &playback_url_remap, &excluded_playback_urls](Info::UrlMeta& meta) {
404 [ + - ]: 45 : const std::string input_url = meta.url;
405 : :
406 [ + - + + ]: 45 : if VUNLIKELY (!plugin_interface->convert_url_meta(meta.url, meta.ser_type, meta.schema_type)) {
407 [ + - ]: 5 : excluded_playback_urls.emplace(input_url);
408 : 5 : return true;
409 : : }
410 : :
411 [ + + ]: 40 : if (meta.url != input_url) {
412 [ + - + - ]: 8 : playback_url_remap[input_url] = meta.url;
413 : : }
414 : :
415 : 40 : return false;
416 : 45 : }),
417 : 42 : url_metas.end());
418 : : }
419 : :
420 : : {
421 [ + - ]: 24 : std::unique_lock state_lock(impl_->playback_state_mtx);
422 : :
423 [ + - ]: 24 : if (impl_->plugin_interface == plugin_interface) {
424 : 24 : impl_->playback_url_remap = std::move(playback_url_remap);
425 : 24 : impl_->excluded_playback_urls = std::move(excluded_playback_urls);
426 : 48 : impl_->playback_url_rules_enabled.store(
427 [ + + + + ]: 24 : !impl_->playback_url_remap.empty() || !impl_->excluded_playback_urls.empty(), std::memory_order_release);
428 : : }
429 : 24 : }
430 : 24 : }
431 : :
432 : 276 : bool BagReader::convert_playback_url(const std::string& input_url, std::string& output_url) const {
433 [ + - ]: 276 : std::shared_lock state_lock(impl_->playback_state_mtx);
434 : :
435 [ + - + + ]: 276 : if VUNLIKELY (impl_->excluded_playback_urls.count(input_url) != 0U) {
436 : 4 : return false;
437 : : }
438 : :
439 [ + - ]: 272 : auto iter = impl_->playback_url_remap.find(input_url);
440 : :
441 [ + + ]: 272 : if (iter != impl_->playback_url_remap.end()) {
442 [ + - ]: 10 : output_url = iter->second;
443 : : } else {
444 [ + - ]: 262 : output_url = input_url;
445 : : }
446 : :
447 : 272 : return true;
448 : 276 : }
449 : :
450 : 133 : bool BagReader::match_playback_url_filter(std::string_view input_url,
451 : : const std::unordered_set<std::string>& filter_urls) const {
452 [ + + ]: 133 : if VUNLIKELY (!input_url.data()) {
453 : 1 : return false;
454 : : }
455 : :
456 [ + + + + : 132 : if (filter_urls.empty() && !has_playback_url_rules()) {
+ + ]
457 : 58 : return true;
458 : : }
459 : :
460 : 74 : std::string output_url;
461 : :
462 [ + - + - : 74 : if VUNLIKELY (!convert_playback_url(std::string(input_url), output_url)) {
+ + ]
463 : 3 : return false;
464 : : }
465 : :
466 [ + + + - : 71 : return filter_urls.empty() || filter_urls.count(output_url) != 0U;
+ + ]
467 : 74 : }
468 : :
469 : 128 : bool BagReader::has_playback_url_rules() const noexcept {
470 : 128 : return impl_->playback_url_rules_enabled.load(std::memory_order_acquire);
471 : : }
472 : :
473 : 298 : void BagReader::rebuild_url_meta_lookup(const std::vector<Info::UrlMeta>& url_metas) {
474 [ + - ]: 298 : std::unique_lock state_lock(impl_->playback_state_mtx);
475 : :
476 [ + - ]: 298 : rebuild_url_meta_maps(url_metas, impl_->url_to_ser_map, impl_->url_to_schema_type_map);
477 : 298 : }
478 : :
479 : 300 : void BagReader::rebuild_url_meta_maps(const std::vector<Info::UrlMeta>& url_metas,
480 : : std::unordered_map<std::string, std::string>& ser_map,
481 : : std::unordered_map<std::string, SchemaType>& schema_type_map) {
482 : 300 : ser_map.clear();
483 : 300 : schema_type_map.clear();
484 [ + - ]: 300 : ser_map.reserve(url_metas.size());
485 [ + - ]: 300 : schema_type_map.reserve(url_metas.size());
486 : :
487 : 300 : std::unordered_set<std::string> ser_conflict_urls;
488 : 300 : std::unordered_set<std::string> schema_conflict_urls;
489 : :
490 [ + - ]: 300 : ser_conflict_urls.reserve(url_metas.size());
491 [ + - ]: 300 : schema_conflict_urls.reserve(url_metas.size());
492 : :
493 [ + + ]: 729 : for (const auto& meta : url_metas) {
494 [ + - ]: 429 : auto& merged_ser_type = ser_map[meta.url];
495 [ + - ]: 429 : auto& merged_schema_type = schema_type_map[meta.url];
496 : :
497 [ + - + + : 429 : if (ser_conflict_urls.count(meta.url) == 0U && !meta.ser_type.empty()) {
+ + + + ]
498 [ + + + + : 418 : if (merged_ser_type.empty() || merged_ser_type == "Bytes") {
+ + ]
499 [ + - ]: 413 : merged_ser_type = meta.ser_type;
500 [ + - + + : 5 : } else if (meta.ser_type != "Bytes" && meta.ser_type != merged_ser_type) {
+ + ]
501 [ + - + - ]: 2 : CLOG_E("BagReader: URL remap collision on %s, keeping ser_type unknown. ser [%s] vs [%s].", meta.url.c_str(),
502 : : merged_ser_type.c_str(), meta.ser_type.c_str());
503 : 1 : merged_ser_type.clear();
504 [ + - ]: 1 : ser_conflict_urls.emplace(meta.url);
505 : : }
506 : : }
507 : :
508 [ + - + + : 429 : if (schema_conflict_urls.count(meta.url) == 0U && meta.schema_type != SchemaType::kUnknown) {
+ + + + ]
509 [ + + ]: 380 : if (merged_schema_type == SchemaType::kUnknown) {
510 : 375 : merged_schema_type = meta.schema_type;
511 [ + + ]: 5 : } else if (merged_schema_type != meta.schema_type) {
512 : 1 : const auto current_label = SchemaData::convert_type(merged_schema_type);
513 : 1 : const auto new_label = SchemaData::convert_type(meta.schema_type);
514 [ + - + - ]: 2 : CLOG_E("BagReader: URL remap collision on %s, keeping schema_type unknown. schema [%.*s] vs [%.*s].",
515 : : meta.url.c_str(), static_cast<int>(current_label.size()), current_label.data(),
516 : : static_cast<int>(new_label.size()), new_label.data());
517 : 1 : merged_schema_type = SchemaType::kUnknown;
518 [ + - ]: 1 : schema_conflict_urls.emplace(meta.url);
519 : : }
520 : : }
521 : : }
522 : 300 : }
523 : :
524 : 323 : ActionType BagReader::convert_action(std::string_view str) {
525 [ + + ]: 323 : if (str == "C/Req") {
526 : 27 : return ActionType::kClientRequest;
527 [ + + ]: 296 : } else if (str == "C/Resp") {
528 : 4 : return ActionType::kClientResponse;
529 [ + + ]: 292 : } else if (str == "S/Req") {
530 : 3 : return ActionType::kServerRequest;
531 [ + + ]: 289 : } else if (str == "S/Resp") {
532 : 3 : return ActionType::kServerResponse;
533 [ + + ]: 286 : } else if (str == "Pub") {
534 : 240 : return ActionType::kPublish;
535 [ + + ]: 46 : } else if (str == "Sub") {
536 : 6 : return ActionType::kSubscribe;
537 [ + + ]: 40 : } else if (str == "Set") {
538 : 31 : return ActionType::kSet;
539 [ + + ]: 9 : } else if (str == "Get") {
540 : 2 : return ActionType::kGet;
541 : : } else {
542 : 7 : return ActionType::kUnknownAction;
543 : : }
544 : : }
545 : :
546 : : } // namespace vlink
|
