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_writer.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <memory>
29 : : #include <mutex>
30 : : #include <shared_mutex>
31 : : #include <string>
32 : : #include <unordered_map>
33 : : #include <vector>
34 : :
35 : : #include "./base/helpers.h"
36 : : #include "./base/logger.h"
37 : : #include "./base/utils.h"
38 : : #include "./extension/bag_plugin_interface.h"
39 : : #include "./extension/schema_plugin_manager.h"
40 : : #include "./extension/vcap_writer.h"
41 : : #include "./extension/vdb_writer.h"
42 : :
43 : : namespace vlink {
44 : :
45 : : // GlobalWriter
46 : : struct GlobalWriter final {
47 : 56 : GlobalWriter() {
48 [ + - + - ]: 56 : const std::string& bag_path = Utils::get_env("VLINK_BAG_PATH");
49 : :
50 [ + + ]: 56 : if (!bag_path.empty()) {
51 [ + - + - ]: 4 : VLOG_I("BagWriter: Global recorder is enabled.");
52 [ + - + - ]: 4 : CLOG_I("BagWriter: Record path: %s.", bag_path.c_str());
53 : :
54 [ + - ]: 2 : instance = BagWriter::create(bag_path);
55 : :
56 [ + + ]: 2 : if VLIKELY (instance) {
57 [ + - ]: 1 : instance->async_run();
58 : : } else {
59 [ + - + - ]: 2 : CLOG_E("BagWriter: Global recorder is disabled because VLINK_BAG_PATH has an unsupported suffix.");
60 : : }
61 : : }
62 : 56 : }
63 : :
64 : 56 : ~GlobalWriter() {
65 [ + + ]: 56 : if (instance) {
66 : 1 : instance.reset();
67 : : }
68 : 56 : }
69 : :
70 : 3892 : static GlobalWriter& get() {
71 [ + + + - : 3892 : static GlobalWriter global_writer;
+ - - - ]
72 : 3890 : return global_writer;
73 : : }
74 : :
75 : : std::mutex mtx;
76 : : std::unordered_map<std::string, std::weak_ptr<BagWriter>> writer_map;
77 : :
78 : : std::shared_ptr<BagWriter> instance;
79 : :
80 : : VLINK_DISALLOW_COPY_AND_ASSIGN(GlobalWriter)
81 : : };
82 : :
83 : : // BagWriter::Impl
84 : : struct BagWriter::Impl final {
85 : : std::unordered_map<int, std::string> index_to_url_map;
86 : : std::unordered_map<int, std::string> index_to_ser_map;
87 : : std::unordered_map<std::string, int> url_to_index_map;
88 : : std::unordered_map<std::string, int> ser_to_index_map;
89 : : int current_url_index{0};
90 : : int current_ser_index{0};
91 : : mutable std::shared_mutex shared_mtx;
92 : :
93 : : std::shared_ptr<BagPluginInterface> plugin_interface;
94 : : std::unordered_map<std::string, std::string> recorded_url_remap;
95 : : std::unordered_map<std::string, std::vector<std::string>> recorded_urls_by_origin;
96 : : std::unordered_map<std::string, std::string> recorded_url_origin;
97 : : mutable std::shared_mutex record_state_mtx;
98 : :
99 : : std::mutex sample_mtx;
100 : : std::unordered_map<std::string, double> url_loss_map;
101 : : std::unordered_map<std::string, double> total_url_loss_map;
102 : :
103 : : std::mutex active_write_mtx;
104 : : std::atomic<uint64_t> active_thread_id{0};
105 : : std::string active_origin_url;
106 : : int64_t active_record_result{0};
107 : :
108 : : std::atomic_bool stream_fail{false};
109 : : };
110 : :
111 : : // BagWriter
112 : 197 : std::shared_ptr<BagWriter> BagWriter::create(const std::string& path, const Config& config) {
113 [ + - ]: 197 : std::string suffix_check = path;
114 : :
115 : 197 : std::transform(suffix_check.begin(), suffix_check.end(), suffix_check.begin(),
116 : 15108 : [](unsigned char c) { return std::tolower(c); });
117 : :
118 [ + + + + : 197 : if (Helpers::has_endwith(suffix_check, ".vdb") || Helpers::has_endwith(suffix_check, ".vdbx")) {
+ + ]
119 [ + - ]: 119 : return std::make_shared<VDBWriter>(path, config);
120 [ + + + + : 78 : } else if (Helpers::has_endwith(suffix_check, ".vcap") || Helpers::has_endwith(suffix_check, ".vcapx")) {
+ + ]
121 [ + - ]: 76 : return std::make_shared<VCAPWriter>(path, config);
122 : : } else {
123 [ + - + - ]: 4 : CLOG_E("BagWriter: Unknown bag suffix, path=%s", path.c_str());
124 : 2 : return nullptr;
125 : : }
126 : 197 : }
127 : :
128 : 7 : std::shared_ptr<BagWriter> BagWriter::filter_get(const std::string& path) {
129 [ + + + - : 7 : static auto& instance = GlobalWriter::get();
+ - - - ]
130 [ + - ]: 7 : std::string suffix_check = path;
131 : :
132 : 7 : std::transform(suffix_check.begin(), suffix_check.end(), suffix_check.begin(),
133 : 496 : [](unsigned char c) { return std::tolower(c); });
134 : :
135 [ + + - + ]: 7 : const bool is_database = Helpers::has_endwith(suffix_check, ".vdb") || Helpers::has_endwith(suffix_check, ".vdbx");
136 [ + + - + ]: 7 : const bool is_mcap = Helpers::has_endwith(suffix_check, ".vcap") || Helpers::has_endwith(suffix_check, ".vcapx");
137 : :
138 [ + + + + : 7 : if VUNLIKELY (!is_database && !is_mcap) {
+ + ]
139 [ + - + - ]: 2 : CLOG_E("BagWriter: Unknown bag suffix, path=%s", path.c_str());
140 : 1 : return nullptr;
141 : : }
142 : :
143 [ + - ]: 6 : std::lock_guard lock(instance.mtx);
144 : :
145 [ + - ]: 6 : auto iter = instance.writer_map.find(path);
146 : :
147 [ + + ]: 6 : if (iter != instance.writer_map.end()) {
148 [ + - ]: 1 : if (auto target = iter->second.lock()) {
149 : 1 : return target;
150 [ - + ]: 1 : }
151 : :
152 : : instance.writer_map.erase(iter); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
153 : : }
154 : :
155 : : {
156 : 5 : std::shared_ptr<BagWriter> target;
157 : :
158 [ + + ]: 5 : if (is_mcap) {
159 [ + - + - ]: 2 : auto* ptr = new VCAPWriter(path);
160 : :
161 [ + - + - ]: 4 : target = std::shared_ptr<VCAPWriter>(ptr, [path](VCAPWriter* ptr) {
162 : : {
163 [ + - ]: 2 : std::lock_guard lock(instance.mtx);
164 [ + - ]: 2 : auto iter = instance.writer_map.find(path);
165 : :
166 [ + - + - : 2 : if (iter != instance.writer_map.end() && iter->second.expired()) {
+ - ]
167 [ + - ]: 2 : instance.writer_map.erase(iter);
168 : : }
169 : 2 : }
170 : :
171 [ + - ]: 2 : delete ptr;
172 : 4 : });
173 : : } else {
174 [ + - + - ]: 3 : auto* ptr = new VDBWriter(path);
175 : :
176 [ + - + - ]: 6 : target = std::shared_ptr<VDBWriter>(ptr, [path](VDBWriter* ptr) {
177 : : {
178 [ + - ]: 3 : std::lock_guard lock(instance.mtx);
179 [ + - ]: 3 : auto iter = instance.writer_map.find(path);
180 : :
181 [ + - + - : 3 : if (iter != instance.writer_map.end() && iter->second.expired()) {
+ - ]
182 [ + - ]: 3 : instance.writer_map.erase(iter);
183 : : }
184 : 3 : }
185 : :
186 [ + - ]: 3 : delete ptr;
187 : 6 : });
188 : : }
189 : :
190 [ + - ]: 5 : target->async_run();
191 : :
192 [ + - ]: 5 : instance.writer_map.emplace(path, target);
193 : :
194 : 5 : return target;
195 : 5 : }
196 : 7 : }
197 : :
198 : 3891 : BagWriter* BagWriter::global_get() { return GlobalWriter::get().instance.get(); }
199 : :
200 [ + - ]: 229 : BagWriter::BagWriter(const std::string& path, const Config& config) : impl_(std::make_unique<Impl>()) {
201 : : (void)path;
202 : :
203 [ + + ]: 229 : if (!config.sync_mode) {
204 [ + - ]: 51 : impl_->index_to_url_map.reserve(128);
205 [ + - ]: 51 : impl_->index_to_ser_map.reserve(128);
206 [ + - ]: 51 : impl_->url_to_index_map.reserve(128);
207 [ + - ]: 51 : impl_->ser_to_index_map.reserve(128);
208 : : }
209 : :
210 : 229 : Bytes::init_memory_pool();
211 : 229 : }
212 : :
213 : 34 : void BagWriter::get_url_meta(const std::string& url, const std::string& ser, int& url_index, int& ser_index) const {
214 : : {
215 [ + - ]: 34 : std::shared_lock read_lock(impl_->shared_mtx);
216 : :
217 [ + - ]: 34 : auto url_iter = impl_->url_to_index_map.find(url);
218 [ + - ]: 34 : auto ser_iter = impl_->ser_to_index_map.find(ser);
219 : :
220 [ + + + + : 34 : if VLIKELY (url_iter != impl_->url_to_index_map.end() && ser_iter != impl_->ser_to_index_map.end()) {
+ + ]
221 : 5 : url_index = url_iter->second;
222 : 5 : ser_index = ser_iter->second;
223 : 5 : return;
224 : : }
225 [ + + ]: 34 : }
226 : :
227 [ + - ]: 29 : std::unique_lock write_lock(impl_->shared_mtx);
228 : :
229 [ + - ]: 29 : auto& url_id = impl_->url_to_index_map.try_emplace(url, -1).first->second;
230 : :
231 [ + + ]: 29 : if (url_id < 0) {
232 : 26 : url_id = ++impl_->current_url_index;
233 [ + - + - ]: 26 : impl_->index_to_url_map[url_id] = url;
234 : : }
235 : :
236 [ + - ]: 29 : auto& ser_id = impl_->ser_to_index_map.try_emplace(ser, -1).first->second;
237 : :
238 [ + + ]: 29 : if (ser_id < 0) {
239 : 28 : ser_id = ++impl_->current_ser_index;
240 [ + - + - ]: 28 : impl_->index_to_ser_map[ser_id] = ser;
241 : : }
242 : :
243 : 29 : url_index = url_id;
244 : 29 : ser_index = ser_id;
245 : 29 : }
246 : :
247 : 23 : void BagWriter::get_url_meta(int url_index, int ser_index, std::string& url, std::string& ser) const {
248 [ + - ]: 23 : std::shared_lock read_lock(impl_->shared_mtx);
249 : :
250 [ + - ]: 23 : auto url_iter = impl_->index_to_url_map.find(url_index);
251 : :
252 [ + + ]: 23 : if VLIKELY (url_iter != impl_->index_to_url_map.end()) {
253 [ + - ]: 22 : url = url_iter->second;
254 : : }
255 : :
256 [ + - ]: 23 : auto ser_iter = impl_->index_to_ser_map.find(ser_index);
257 : :
258 [ + + ]: 23 : if VLIKELY (ser_iter != impl_->index_to_ser_map.end()) {
259 [ + - ]: 22 : ser = ser_iter->second;
260 : : }
261 : 23 : }
262 : :
263 : 229 : BagWriter::~BagWriter() {
264 : 229 : std::shared_ptr<BagPluginInterface> plugin_interface;
265 : :
266 : : {
267 : 229 : std::shared_lock state_lock(impl_->record_state_mtx);
268 : 229 : plugin_interface = impl_->plugin_interface;
269 : 229 : }
270 : :
271 [ + + ]: 229 : if (plugin_interface) {
272 : 9 : plugin_interface->register_callback({});
273 : : }
274 : 229 : }
275 : :
276 : 2 : void BagWriter::flush_plugin() {
277 : 2 : std::shared_ptr<BagPluginInterface> plugin_interface;
278 : :
279 : : {
280 [ + - ]: 2 : std::shared_lock state_lock(impl_->record_state_mtx);
281 : 2 : plugin_interface = impl_->plugin_interface;
282 : 2 : }
283 : :
284 [ + - ]: 2 : if (plugin_interface) {
285 [ + - ]: 2 : plugin_interface->flush();
286 : : }
287 : 2 : }
288 : :
289 : 201 : void BagWriter::detach_plugin() {
290 : 201 : std::shared_ptr<BagPluginInterface> plugin_interface;
291 : :
292 : : {
293 [ + - ]: 201 : std::unique_lock state_lock(impl_->record_state_mtx);
294 : 201 : plugin_interface = std::move(impl_->plugin_interface);
295 : 201 : }
296 : :
297 [ + + ]: 201 : if (plugin_interface) {
298 [ + - ]: 1 : plugin_interface->flush();
299 : 1 : plugin_interface->register_callback({});
300 : : }
301 : 201 : }
302 : :
303 : 30 : void BagWriter::bind_bag_interface(const std::shared_ptr<BagPluginInterface>& bag_interface) {
304 : 30 : std::shared_ptr<BagPluginInterface> old_plugin_interface;
305 : :
306 : : {
307 [ + - ]: 30 : std::shared_lock state_lock(impl_->record_state_mtx);
308 : 30 : old_plugin_interface = impl_->plugin_interface;
309 : 30 : }
310 : :
311 [ + + + - : 30 : if (old_plugin_interface && old_plugin_interface != bag_interface) {
+ + ]
312 [ + - ]: 10 : old_plugin_interface->flush();
313 : 10 : old_plugin_interface->register_callback({});
314 : : }
315 : :
316 [ + + ]: 30 : if VLIKELY (bag_interface) {
317 : 20 : bag_interface->bind_direction(BagPluginInterface::Direction::kWrite);
318 : :
319 [ + - ]: 20 : bag_interface->register_callback([this](const Frame& frame) {
320 : 98 : const bool active = impl_->active_thread_id.load(std::memory_order_acquire) == Utils::get_native_thread_id();
321 : :
322 [ + + ]: 49 : if VUNLIKELY (frame.url.empty()) {
323 [ + - ]: 1 : if (active) {
324 : 1 : impl_->active_record_result = -1;
325 : : } else {
326 : 0 : set_fail();
327 : : }
328 : :
329 : 1 : return;
330 : : }
331 : :
332 [ + + + + : 48 : if (active && impl_->active_origin_url != frame.url) {
+ + ]
333 : 6 : learn_recorded_url(impl_->active_origin_url, frame.url);
334 : : }
335 : :
336 : 48 : const int64_t record_result = record(frame, frame.timestamp);
337 : :
338 [ + + + + : 48 : if VUNLIKELY (active && record_result < 0) {
+ + ]
339 : 1 : impl_->active_record_result = record_result;
340 [ + + ]: 47 : } else if VUNLIKELY (record_result < 0) {
341 : 2 : set_fail();
342 : : }
343 : : });
344 : : }
345 : :
346 [ + - ]: 30 : std::unique_lock state_lock(impl_->record_state_mtx);
347 : :
348 : 30 : impl_->plugin_interface = bag_interface;
349 : 30 : }
350 : :
351 [ + - ]: 7 : void BagWriter::clear_bag_interface() { bind_bag_interface(nullptr); }
352 : :
353 : 537 : int64_t BagWriter::push(const Frame& frame) {
354 [ + + ]: 537 : if VUNLIKELY (frame.url.empty()) {
355 : 1 : return -1;
356 : : }
357 : :
358 [ + + + - ]: 536 : const int64_t target_timestamp = frame.timestamp < 0 ? get_record_timestamp() : frame.timestamp;
359 : :
360 : 536 : std::shared_ptr<BagPluginInterface> plugin_interface;
361 : :
362 : : {
363 [ + - ]: 536 : std::shared_lock state_lock(impl_->record_state_mtx);
364 : 536 : plugin_interface = impl_->plugin_interface;
365 : 536 : }
366 : :
367 [ + + ]: 536 : if VLIKELY (!plugin_interface) {
368 [ + - ]: 486 : return record(frame, target_timestamp);
369 : : }
370 : :
371 : 50 : Frame stamped;
372 : 50 : const Frame* effective = &frame;
373 : :
374 [ + + ]: 50 : if (frame.timestamp != target_timestamp) {
375 : 1 : stamped.timestamp = target_timestamp;
376 [ + - ]: 1 : stamped.url = frame.url;
377 [ + - ]: 1 : stamped.ser_type = frame.ser_type;
378 : 1 : stamped.schema_type = frame.schema_type;
379 : 1 : stamped.action_type = frame.action_type;
380 : 1 : stamped.data = Bytes::shallow_copy(frame.data.data(), frame.data.size());
381 : 1 : effective = &stamped;
382 : : }
383 : :
384 [ + - ]: 50 : std::lock_guard active_lock(impl_->active_write_mtx);
385 : :
386 [ + - ]: 50 : impl_->active_origin_url = effective->url;
387 : 50 : impl_->active_record_result = target_timestamp;
388 : 50 : impl_->active_thread_id.store(Utils::get_native_thread_id(), std::memory_order_release);
389 [ + - ]: 50 : plugin_interface->on_write(*effective);
390 : :
391 : 50 : impl_->active_thread_id.store(0, std::memory_order_release);
392 [ + + ]: 50 : const int64_t result = impl_->active_record_result < 0 ? impl_->active_record_result : target_timestamp;
393 : 50 : impl_->active_origin_url.clear();
394 : 50 : impl_->active_record_result = 0;
395 : :
396 : 50 : return result;
397 : 536 : }
398 : :
399 : 5 : BagWriter& BagWriter::operator<<(const Frame& frame) {
400 [ + + ]: 5 : if VUNLIKELY (push(frame) < 0) {
401 : 1 : set_fail();
402 : : }
403 : :
404 : 5 : return *this;
405 : : }
406 : :
407 : 1 : BagWriter& BagWriter::operator<<(const SchemaData& schema_data) {
408 [ - + ]: 1 : if VUNLIKELY (!push_schema(schema_data)) {
409 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
410 : : }
411 : :
412 : 1 : return *this;
413 : : }
414 : :
415 : 37 : bool BagWriter::fail() const noexcept { return impl_->stream_fail.load(std::memory_order_acquire); }
416 : :
417 : 3 : BagWriter::operator bool() const noexcept { return !impl_->stream_fail.load(std::memory_order_acquire); }
418 : :
419 : 1 : void BagWriter::clear() noexcept { impl_->stream_fail.store(false, std::memory_order_release); }
420 : :
421 : 0 : void BagWriter::close() {}
422 : :
423 : 28 : bool BagWriter::post_persistent_task(Callback&& callback) {
424 : 28 : return post_untracked_task(std::move(callback), TaskOverflowPolicy::kReject, TaskDropPolicy::kProtected);
425 : : }
426 : :
427 : 5 : void BagWriter::set_fail() noexcept { impl_->stream_fail.store(true, std::memory_order_release); }
428 : :
429 : 6 : void BagWriter::learn_recorded_url(const std::string& origin_url, const std::string& recorded_url) {
430 : : {
431 [ + - ]: 6 : std::shared_lock state_lock(impl_->record_state_mtx);
432 : :
433 [ + - ]: 6 : auto iter = impl_->recorded_url_remap.find(origin_url);
434 : :
435 [ + + - + : 6 : if (iter != impl_->recorded_url_remap.end() && iter->second == recorded_url) {
- + ]
436 : 0 : return;
437 : : }
438 [ + - ]: 6 : }
439 : :
440 [ + - ]: 6 : std::unique_lock state_lock(impl_->record_state_mtx);
441 : :
442 [ + - ]: 6 : impl_->recorded_url_remap.try_emplace(origin_url, recorded_url);
443 [ + - ]: 6 : auto& recorded_urls = impl_->recorded_urls_by_origin[origin_url];
444 : :
445 [ + - + - ]: 6 : if (std::find(recorded_urls.begin(), recorded_urls.end(), recorded_url) == recorded_urls.end()) {
446 [ + - ]: 6 : recorded_urls.emplace_back(recorded_url);
447 : : }
448 : :
449 [ + - + - ]: 6 : impl_->recorded_url_origin[recorded_url] = origin_url;
450 : 6 : }
451 : :
452 : 3 : std::string BagWriter::convert_recorded_url(const std::string& url) const {
453 [ + - ]: 3 : std::shared_lock state_lock(impl_->record_state_mtx);
454 : :
455 [ + - ]: 3 : auto iter = impl_->recorded_url_remap.find(url);
456 : :
457 [ + + + - ]: 6 : return iter == impl_->recorded_url_remap.end() ? url : iter->second;
458 : 3 : }
459 : :
460 : 40 : std::vector<std::string> BagWriter::recorded_urls_for_origin(const std::string& url) const {
461 [ + - ]: 40 : std::shared_lock state_lock(impl_->record_state_mtx);
462 : 40 : std::vector<std::string> urls;
463 [ + - ]: 40 : urls.emplace_back(url);
464 : :
465 [ + - ]: 40 : auto iter = impl_->recorded_urls_by_origin.find(url);
466 : :
467 [ + + ]: 40 : if (iter != impl_->recorded_urls_by_origin.end()) {
468 [ + + ]: 3 : for (const auto& recorded_url : iter->second) {
469 [ + - ]: 2 : if (recorded_url != url) {
470 [ + - ]: 2 : urls.emplace_back(recorded_url);
471 : : }
472 : : }
473 : : }
474 : :
475 : 80 : return urls;
476 : 40 : }
477 : :
478 : 258 : std::string BagWriter::recover_recorded_url(const std::string& url) const {
479 [ + - ]: 258 : std::shared_lock state_lock(impl_->record_state_mtx);
480 : :
481 [ + - ]: 258 : auto iter = impl_->recorded_url_origin.find(url);
482 : :
483 [ + + + - ]: 516 : return iter == impl_->recorded_url_origin.end() ? url : iter->second;
484 : 258 : }
485 : :
486 : 436 : std::mutex& BagWriter::sample_mutex() { return impl_->sample_mtx; }
487 : :
488 : 874 : std::unordered_map<std::string, double>& BagWriter::url_loss_map_ref() { return impl_->url_loss_map; }
489 : :
490 : 325 : std::unordered_map<std::string, double>& BagWriter::total_url_loss_map_ref() { return impl_->total_url_loss_map; }
491 : :
492 : 51 : void BagWriter::set_url_loss(const std::string& url, double loss) {
493 [ + + ]: 51 : if (loss > 1) {
494 : 2 : loss = -1;
495 : : }
496 : :
497 [ + - ]: 51 : std::lock_guard lock(impl_->sample_mtx);
498 : :
499 [ + - ]: 51 : impl_->url_loss_map[url] = loss;
500 [ + - ]: 51 : impl_->total_url_loss_map[url] = loss;
501 : 51 : }
502 : :
503 : 36 : const std::string& BagWriter::get_default_tag_name() {
504 [ + + + - : 36 : static std::string tag_name_env_str = Utils::get_env("VLINK_BAG_TAG", "Empty");
+ - + - -
- ]
505 : 36 : return tag_name_env_str;
506 : : }
507 : :
508 : 201 : const std::string& BagWriter::get_default_app_name() {
509 [ + + + - ]: 201 : static std::string app_name = Utils::get_app_name();
510 : 201 : return app_name;
511 : : }
512 : :
513 [ + - + - : 200 : SchemaPluginInterface* BagWriter::get_schema_interface() { return SchemaPluginManager::get().get_interface().get(); }
+ - ]
514 : :
515 : 201 : int32_t BagWriter::get_default_timezone_diff() { return Utils::get_timezone_diff(); }
516 : :
517 : 827 : std::string_view BagWriter::convert_action(ActionType type) {
518 [ + + + + : 827 : switch (type) {
+ + + +
+ ]
519 : 63 : case ActionType::kClientRequest:
520 : 63 : return "C/Req";
521 : 9 : case ActionType::kClientResponse:
522 : 9 : return "C/Resp";
523 : 5 : case ActionType::kServerRequest:
524 : 5 : return "S/Req";
525 : 5 : case ActionType::kServerResponse:
526 : 5 : return "S/Resp";
527 : 565 : case ActionType::kPublish:
528 : 565 : return "Pub";
529 : 103 : case ActionType::kSubscribe:
530 : 103 : return "Sub";
531 : 67 : case ActionType::kSet:
532 : 67 : return "Set";
533 : 5 : case ActionType::kGet:
534 : 5 : return "Get";
535 : 5 : default:
536 : 5 : return "Unknown";
537 : : }
538 : : }
539 : :
540 : 462 : std::string BagWriter::get_format_date(SystemClock* current, bool file_format) {
541 : 462 : SystemClock time_point;
542 : :
543 [ + + ]: 462 : if (current) {
544 : 451 : time_point = *current;
545 : : } else {
546 [ + - ]: 11 : time_point = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
547 : : }
548 : :
549 : 462 : auto milliseconds = time_point.time_since_epoch().count() % 1000U;
550 : :
551 [ + - ]: 462 : std::time_t now_time_t = std::chrono::system_clock::to_time_t(time_point);
552 : :
553 : 462 : std::tm now_tm{};
554 : :
555 : : #ifdef _WIN32
556 : : localtime_s(&now_tm, &now_time_t);
557 : : #else
558 : 462 : localtime_r(&now_time_t, &now_tm);
559 : : #endif
560 : :
561 : : char buffer[32];
562 : : char full_buffer[64];
563 : :
564 [ + + ]: 462 : if (file_format) {
565 : 23 : std::strftime(buffer, sizeof(buffer), "%Y-%m-%d_%H-%M-%S", &now_tm);
566 : 23 : std::snprintf(full_buffer, sizeof(full_buffer), "%s-%03lld", buffer,
567 : : static_cast<long long>(milliseconds)); // NOLINT(runtime/int, google-runtime-int)
568 : : } else {
569 : 439 : std::strftime(buffer, sizeof(buffer), "%Y/%m/%d %H:%M:%S", &now_tm);
570 : 439 : std::snprintf(full_buffer, sizeof(full_buffer), "%s:%03lld", buffer,
571 : : static_cast<long long>(milliseconds)); // NOLINT(runtime/int, google-runtime-int)
572 : : }
573 : :
574 [ + - ]: 462 : return full_buffer;
575 : : }
576 : :
577 : : } // namespace vlink
|
