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