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