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/discovery_viewer.h"
25 : :
26 : : #include <algorithm>
27 : : #include <charconv>
28 : : #include <cstring>
29 : : #include <ctime>
30 : : #include <exception>
31 : : #include <map>
32 : : #include <memory>
33 : : #include <shared_mutex>
34 : : #include <string>
35 : : #include <thread>
36 : : #include <unordered_map>
37 : : #include <unordered_set>
38 : : #include <utility>
39 : : #include <vector>
40 : :
41 : : #include "./base/bytes.h"
42 : : #include "./base/elapsed_timer.h"
43 : : #include "./base/helpers.h"
44 : : #include "./base/logger.h"
45 : : #include "./base/utils.h"
46 : : #include "./impl/url.h"
47 : :
48 : : #if __has_include(<unistd.h>)
49 : : #include <unistd.h>
50 : : #endif
51 : :
52 : : #ifdef _WIN32
53 : : #include <Winsock2.h>
54 : : #include <ws2tcpip.h>
55 : : #else
56 : : #ifdef __QNX__
57 : : #include <sys/time.h>
58 : : #endif
59 : : #include <arpa/inet.h>
60 : : #include <netinet/in.h>
61 : : #include <sys/socket.h>
62 : : #endif
63 : :
64 : : #define VLINK_DISCOVERY_MULTICAST 1
65 : : #define VLINK_DISCOVERY_OFFLINE 0
66 : :
67 : : namespace vlink {
68 : :
69 : : [[maybe_unused]] static constexpr int kCollectInterval = 500;
70 : : [[maybe_unused]] static constexpr size_t kMaxTaskSize = 50000U;
71 : : [[maybe_unused]] static constexpr uint32_t kMaxElapsedTime = 1000;
72 : : [[maybe_unused]] static constexpr int kBroadcastBindPort = 51694;
73 : : [[maybe_unused]] static constexpr size_t kBufferSize = 1024 * 1024U;
74 : :
75 : : #if VLINK_DISCOVERY_MULTICAST
76 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "239.255.0.100";
77 : : #else
78 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "255.255.255.255";
79 : : #endif
80 : :
81 : 29 : [[maybe_unused]] static std::string node_count_to_string(size_t node_count) {
82 [ + + ]: 29 : if (node_count > 9) {
83 [ + - ]: 1 : return std::string("~");
84 : : }
85 : :
86 : 28 : return std::to_string(node_count);
87 : : }
88 : :
89 : : // DiscoveryViewer::Impl
90 : : struct DiscoveryViewer::Impl final {
91 : : struct Comparator final {
92 : 1177 : bool operator()(const DiscoveryViewer::Info& lhs, const DiscoveryViewer::Info& rhs) const {
93 [ + + ]: 1177 : if (lhs.sort_index < rhs.sort_index) {
94 : 43 : return true;
95 [ + + ]: 1134 : } else if (lhs.sort_index > rhs.sort_index) {
96 : 17 : return false;
97 : : }
98 : :
99 [ + + ]: 1117 : if (lhs.url < rhs.url) {
100 : 595 : return true;
101 [ + + ]: 522 : } else if (lhs.url > rhs.url) {
102 : 251 : return false;
103 : : }
104 : :
105 [ + + ]: 271 : if (lhs.type < rhs.type) {
106 : 3 : return true;
107 [ + + ]: 268 : } else if (lhs.type > rhs.type) {
108 : 2 : return false;
109 : : }
110 : :
111 [ - + ]: 266 : if (lhs.schema_type < rhs.schema_type) {
112 : 0 : return true;
113 [ - + ]: 266 : } else if (lhs.schema_type > rhs.schema_type) {
114 : 0 : return false;
115 : : }
116 : :
117 [ - + ]: 266 : if (lhs.ser_type < rhs.ser_type) {
118 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
119 [ - + ]: 266 : } else if (lhs.ser_type > rhs.ser_type) {
120 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
121 : : }
122 : :
123 : 266 : return lhs.process_list < rhs.process_list;
124 : : }
125 : : };
126 : :
127 : : DiscoveryViewer::FilterType filter_type{DiscoveryViewer::kFilterNone};
128 : : std::string native_hostname;
129 : :
130 : : std::map<DiscoveryViewer::Info, ElapsedTimer, Comparator> info_map;
131 : : std::map<std::pair<int, uint32_t>, std::string> process_map;
132 : : std::unordered_map<std::string, std::string> ser_map;
133 : : std::unordered_map<std::string, SchemaType> schema_type_map;
134 : : std::vector<DiscoveryViewer::Info> info_list;
135 : :
136 : : std::recursive_mutex mtx;
137 : : std::shared_mutex ser_mtx;
138 : : bool info_dirty{false};
139 : :
140 : : DiscoveryViewer::Callback callback;
141 : : std::thread thread;
142 : : std::vector<uint8_t> buffer;
143 : : Timer timer;
144 : :
145 : : int sock{-1};
146 : : sockaddr_in address;
147 : : bool enable_native_discovery{false};
148 : : };
149 : :
150 : : // DiscoveryViewer::Process
151 : 554 : bool DiscoveryViewer::Process::operator<(const DiscoveryViewer::Process& target) const noexcept {
152 [ + + ]: 554 : if (type < target.type) {
153 : 1 : return true;
154 [ + + ]: 553 : } else if (type > target.type) {
155 : 10 : return false;
156 [ + + ]: 543 : } else if (host < target.host) {
157 : 1 : return true;
158 [ + + ]: 542 : } else if (host > target.host) {
159 : 1 : return false;
160 [ + + ]: 541 : } else if (ip < target.ip) {
161 : 1 : return true;
162 [ - + ]: 540 : } else if (ip > target.ip) {
163 : 0 : return false;
164 [ + + ]: 540 : } else if (name < target.name) {
165 : 1 : return true;
166 [ - + ]: 539 : } else if (name > target.name) {
167 : 0 : return false;
168 : : } else {
169 : 539 : return pid < target.pid;
170 : : }
171 : : }
172 : :
173 : : // DiscoveryViewer::Info
174 : 9 : bool DiscoveryViewer::Info::operator<(const DiscoveryViewer::Info& target) const noexcept {
175 [ + + ]: 9 : if (type < target.type) {
176 : 1 : return true;
177 [ - + ]: 8 : } else if (type > target.type) {
178 : 0 : return false;
179 [ + + ]: 8 : } else if (sort_index < target.sort_index) {
180 : 1 : return true;
181 [ + + ]: 7 : } else if (sort_index > target.sort_index) {
182 : 1 : return false;
183 [ + + ]: 6 : } else if (url < target.url) {
184 : 1 : return true;
185 [ + + ]: 5 : } else if (url > target.url) {
186 : 1 : return false;
187 [ + + ]: 4 : } else if (schema_type < target.schema_type) {
188 : 1 : return true;
189 [ - + ]: 3 : } else if (schema_type > target.schema_type) {
190 : 0 : return false;
191 [ + + ]: 3 : } else if (ser_type < target.ser_type) {
192 : 1 : return true;
193 [ - + ]: 2 : } else if (ser_type > target.ser_type) {
194 : 0 : return false;
195 : : } else {
196 : 2 : return process_list < target.process_list;
197 : : }
198 : : }
199 : :
200 : : // DiscoveryViewer
201 : 144 : ImplType DiscoveryViewer::convert_type(std::string_view str) {
202 [ + + ]: 144 : if (str == "Ser") {
203 : 1 : return kServer;
204 [ + + ]: 143 : } else if (str == "Cli") {
205 : 24 : return kClient;
206 [ + + ]: 119 : } else if (str == "Pub") {
207 : 5 : return kPublisher;
208 [ + + ]: 114 : } else if (str == "Sub") {
209 : 89 : return kSubscriber;
210 [ + + ]: 25 : } else if (str == "Set") {
211 : 2 : return kSetter;
212 [ + + ]: 23 : } else if (str == "Get") {
213 : 19 : return kGetter;
214 : : } else {
215 : 4 : return kUnknownImplType;
216 : : }
217 : : }
218 : :
219 : 20 : std::string DiscoveryViewer::convert_type_to_view(uint32_t type) {
220 [ + + + + : 20 : switch (type) {
+ + + + +
+ + + ]
221 : 1 : case kPublisher | kSubscriber:
222 [ + - ]: 1 : return "Pub|Sub";
223 : 1 : case kSetter | kGetter:
224 [ + - ]: 1 : return "Set|Get";
225 : 1 : case kServer | kClient:
226 [ + - ]: 1 : return "Ser|Cli";
227 : 1 : case kPublisher:
228 [ + - ]: 1 : return "Pub|---";
229 : 1 : case kSubscriber:
230 [ + - ]: 1 : return "---|Sub";
231 : 1 : case kSetter:
232 [ + - ]: 1 : return "Set|---";
233 : 1 : case kGetter:
234 [ + - ]: 1 : return "---|Get";
235 : 1 : case kServer:
236 [ + - ]: 1 : return "Ser|---";
237 : 1 : case kClient:
238 [ + - ]: 1 : return "---|Cli";
239 : 1 : case kPublisher | kGetter:
240 [ + - ]: 1 : return "Pub|Get";
241 : 1 : case kSetter | kSubscriber:
242 [ + - ]: 1 : return "Set|Sub";
243 : 9 : default:
244 [ + + ]: 9 : if (type == (kPublisher | kSetter)) {
245 [ + - ]: 1 : return "Pub|---";
246 [ + + ]: 8 : } else if (type == (kPublisher | kSetter | kSubscriber)) {
247 [ + - ]: 1 : return "Pub|Sub";
248 [ + + ]: 7 : } else if (type == (kPublisher | kSetter | kGetter)) {
249 [ + - ]: 1 : return "Pub|Get";
250 [ + + ]: 6 : } else if (type == (kPublisher | kSetter | kSubscriber | kGetter)) {
251 [ + - ]: 1 : return "Pub|Sub";
252 [ + + ]: 5 : } else if (type == (kSubscriber | kGetter)) {
253 [ + - ]: 1 : return "---|Sub";
254 [ + + ]: 4 : } else if (type == (kSubscriber | kGetter | kPublisher)) {
255 [ + - ]: 1 : return "Pub|Sub";
256 [ + + ]: 3 : } else if (type == (kSubscriber | kGetter | kSetter)) {
257 [ + - ]: 1 : return "Set|Sub";
258 : : } else {
259 [ + - ]: 2 : return "???|???";
260 : : }
261 : : }
262 : : }
263 : :
264 : 21 : std::string DiscoveryViewer::convert_type_to_view(uint32_t type, const std::vector<Process>& process_list) {
265 : 21 : int left_cnt = 0;
266 : 21 : int right_cnt = 0;
267 : :
268 [ + + + + : 21 : switch (type) {
+ + + + +
+ + + ]
269 : 1 : case kPublisher | kSubscriber:
270 [ + + ]: 3 : for (const auto& process : process_list) {
271 [ + + ]: 2 : if (process.type == kPublisher) {
272 : 1 : ++left_cnt;
273 [ + - ]: 1 : } else if (process.type == kSubscriber) {
274 : 1 : ++right_cnt;
275 : : }
276 : : }
277 : :
278 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
279 : 1 : case kSetter | kGetter:
280 [ + + ]: 4 : for (const auto& process : process_list) {
281 [ + + ]: 3 : if (process.type == kSetter) {
282 : 1 : ++left_cnt;
283 [ + - ]: 2 : } else if (process.type == kGetter) {
284 : 2 : ++right_cnt;
285 : : }
286 : : }
287 : :
288 [ + - + - : 2 : return "Set*" + node_count_to_string(left_cnt) + "|Get*" + node_count_to_string(right_cnt);
+ - + - ]
289 : 1 : case kServer | kClient:
290 [ + + ]: 4 : for (const auto& process : process_list) {
291 [ + + ]: 3 : if (process.type == kServer) {
292 : 2 : ++left_cnt;
293 [ + - ]: 1 : } else if (process.type == kClient) {
294 : 1 : ++right_cnt;
295 : : }
296 : : }
297 : :
298 [ + - + - : 2 : return "Ser*" + node_count_to_string(left_cnt) + "|Cli*" + node_count_to_string(right_cnt);
+ - + - ]
299 : 2 : case kPublisher:
300 [ + + ]: 23 : for (const auto& process : process_list) {
301 [ + + ]: 21 : if (process.type == kPublisher) {
302 : 16 : ++left_cnt;
303 : : }
304 : : }
305 : :
306 [ + - + - ]: 4 : return "Pub*" + node_count_to_string(left_cnt) + "|-----";
307 : 1 : case kSubscriber:
308 [ + + ]: 7 : for (const auto& process : process_list) {
309 [ + + ]: 6 : if (process.type == kSubscriber) {
310 : 1 : ++right_cnt;
311 : : }
312 : : }
313 : :
314 [ + - ]: 2 : return "-----|Sub*" + node_count_to_string(right_cnt);
315 : 1 : case kSetter:
316 [ + + ]: 7 : for (const auto& process : process_list) {
317 [ + + ]: 6 : if (process.type == kSetter) {
318 : 1 : ++left_cnt;
319 : : }
320 : : }
321 : :
322 [ + - + - ]: 2 : return "Set*" + node_count_to_string(left_cnt) + "|-----";
323 : 1 : case kGetter:
324 [ + + ]: 7 : for (const auto& process : process_list) {
325 [ + + ]: 6 : if (process.type == kGetter) {
326 : 1 : ++right_cnt;
327 : : }
328 : : }
329 : :
330 [ + - ]: 2 : return "-----|Get*" + node_count_to_string(right_cnt);
331 : 1 : case kServer:
332 [ + + ]: 7 : for (const auto& process : process_list) {
333 [ + + ]: 6 : if (process.type == kServer) {
334 : 1 : ++left_cnt;
335 : : }
336 : : }
337 : :
338 [ + - + - ]: 2 : return "Ser*" + node_count_to_string(left_cnt) + "|-----";
339 : 1 : case kClient:
340 [ + + ]: 7 : for (const auto& process : process_list) {
341 [ + + ]: 6 : if (process.type == kClient) {
342 : 1 : ++right_cnt;
343 : : }
344 : : }
345 : :
346 [ + - ]: 2 : return "-----|Cli*" + node_count_to_string(right_cnt);
347 : :
348 : 1 : case kPublisher | kGetter:
349 [ + + ]: 7 : for (const auto& process : process_list) {
350 [ + + ]: 6 : if (process.type == kPublisher) {
351 : 1 : ++left_cnt;
352 [ + + ]: 5 : } else if (process.type == kGetter) {
353 : 1 : ++right_cnt;
354 : : }
355 : : }
356 : :
357 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Get*" + node_count_to_string(right_cnt);
+ - + - ]
358 : :
359 : 1 : case kSetter | kSubscriber:
360 [ + + ]: 7 : for (const auto& process : process_list) {
361 [ + + ]: 6 : if (process.type == kSetter) {
362 : 1 : ++left_cnt;
363 [ + + ]: 5 : } else if (process.type == kSubscriber) {
364 : 1 : ++right_cnt;
365 : : }
366 : : }
367 : :
368 [ + - + - : 2 : return "Set*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
369 : 9 : default:
370 [ + + ]: 9 : if (type == (kPublisher | kSetter)) {
371 [ + + ]: 7 : for (const auto& process : process_list) {
372 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
373 : 2 : ++left_cnt;
374 : : }
375 : : }
376 : :
377 [ + - + - ]: 2 : return "Pub*" + node_count_to_string(left_cnt) + "|-----";
378 [ + + ]: 8 : } else if (type == (kPublisher | kSetter | kSubscriber)) {
379 [ + + ]: 7 : for (const auto& process : process_list) {
380 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
381 : 2 : ++left_cnt;
382 [ + + ]: 4 : } else if (process.type == kSubscriber) {
383 : 1 : ++right_cnt;
384 : : }
385 : : }
386 : :
387 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
388 [ + + ]: 7 : } else if (type == (kPublisher | kSetter | kGetter)) {
389 [ + + ]: 7 : for (const auto& process : process_list) {
390 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
391 : 2 : ++left_cnt;
392 [ + + ]: 4 : } else if (process.type == kGetter) {
393 : 1 : ++right_cnt;
394 : : }
395 : : }
396 : :
397 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Get*" + node_count_to_string(right_cnt);
+ - + - ]
398 [ + + ]: 6 : } else if (type == (kPublisher | kSetter | kSubscriber | kGetter)) {
399 [ + + ]: 7 : for (const auto& process : process_list) {
400 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
401 : 2 : ++left_cnt;
402 [ + + + + ]: 4 : } else if (process.type == kSubscriber || process.type == kGetter) {
403 : 2 : ++right_cnt;
404 : : }
405 : : }
406 : :
407 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
408 [ + + ]: 5 : } else if (type == (kSubscriber | kGetter)) {
409 [ + + ]: 7 : for (const auto& process : process_list) {
410 [ + + + + ]: 6 : if (process.type == kSubscriber || process.type == kGetter) {
411 : 2 : ++right_cnt;
412 : : }
413 : : }
414 : :
415 [ + - ]: 2 : return "-----|Sub*" + node_count_to_string(right_cnt);
416 [ + + ]: 4 : } else if (type == (kSubscriber | kGetter | kPublisher)) {
417 [ + + ]: 7 : for (const auto& process : process_list) {
418 [ + + ]: 6 : if (process.type == kPublisher) {
419 : 1 : ++left_cnt;
420 [ + + + + ]: 5 : } else if (process.type == kSubscriber || process.type == kGetter) {
421 : 2 : ++right_cnt;
422 : : }
423 : : }
424 : :
425 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
426 [ + + ]: 3 : } else if (type == (kSubscriber | kGetter | kSetter)) {
427 [ + + ]: 7 : for (const auto& process : process_list) {
428 [ + + ]: 6 : if (process.type == kSetter) {
429 : 1 : ++left_cnt;
430 [ + + + + ]: 5 : } else if (process.type == kSubscriber || process.type == kGetter) {
431 : 2 : ++right_cnt;
432 : : }
433 : : }
434 : :
435 [ + - + - : 2 : return "Set*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
436 : : } else {
437 [ + - ]: 2 : return "?????|?????";
438 : : }
439 : : }
440 : : }
441 : :
442 [ + - ]: 2 : std::string DiscoveryViewer::get_listen_address() { return kBroadcastAddress; }
443 : :
444 [ + - ]: 8 : DiscoveryViewer::DiscoveryViewer(FilterType type) : impl_(std::make_unique<Impl>()) {
445 [ + - + - ]: 8 : set_name("DiscoveryViewer");
446 : :
447 [ + - + - ]: 16 : const std::string native_discovery = Utils::get_env("VLINK_DISCOVER_NATIVE");
448 : :
449 [ - + ]: 8 : if (native_discovery == "1") {
450 : 0 : impl_->enable_native_discovery = true;
451 : : }
452 : :
453 [ + - ]: 8 : impl_->ser_map.reserve(128);
454 [ + - ]: 8 : impl_->schema_type_map.reserve(128);
455 : :
456 : 8 : impl_->filter_type = type;
457 : 8 : impl_->native_hostname = Utils::get_host_name();
458 : :
459 : : #ifdef _WIN32
460 : : ::WSADATA wsa_data;
461 : :
462 : : if VUNLIKELY (::WSAStartup(MAKEWORD(2, 2), &wsa_data) != 0) {
463 : : VLOG_F("DiscoveryViewer: Failed to initialize winsock.");
464 : : return;
465 : : }
466 : : #endif
467 : :
468 : : // timer
469 [ + - ]: 8 : impl_->timer.set_interval(kCollectInterval);
470 [ + - ]: 8 : impl_->timer.set_loop_count(Timer::kInfinite);
471 [ + - ]: 8 : impl_->timer.attach(this);
472 : :
473 [ + - + - ]: 171 : impl_->timer.start([this]() { process_timeout(); });
474 : :
475 : : // socket init
476 : 8 : impl_->sock = ::socket(AF_INET, SOCK_DGRAM, 0);
477 : :
478 [ - + ]: 8 : if VUNLIKELY (impl_->sock < 0) {
479 : : VLOG_F("DiscoveryViewer: Failed to create socket."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
480 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
481 : : }
482 : :
483 : : #ifdef _WIN32
484 : : DWORD timeout = 100U;
485 : :
486 : : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout),
487 : : sizeof(timeout)) < 0) {
488 : : VLOG_F("DiscoveryViewer: Failed to set receive timeout.");
489 : : return;
490 : : }
491 : : #else
492 : : timeval timeout;
493 : 8 : std::memset(&timeout, 0, sizeof(timeout));
494 : 8 : timeout.tv_sec = 0U;
495 : 8 : timeout.tv_usec = 1000 * 100U;
496 : :
497 [ - + ]: 8 : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout),
498 : : sizeof(timeout)) < 0) {
499 : : VLOG_F("DiscoveryViewer: Failed to set receive timeout."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
500 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
501 : : }
502 : : #endif
503 : :
504 : 8 : int enable_reuse_addr = 1;
505 : :
506 [ - + ]: 8 : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<const char*>(&enable_reuse_addr),
507 : : sizeof(enable_reuse_addr)) < 0) {
508 : : VLOG_F("DiscoveryViewer: Failed to set reuse address option."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
509 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
510 : : }
511 : :
512 : : #ifdef SO_REUSEPORT
513 : 8 : int enable_reuse_port = 1;
514 : :
515 [ - + ]: 8 : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_REUSEPORT, reinterpret_cast<const char*>(&enable_reuse_port),
516 : : sizeof(enable_reuse_port)) < 0) {
517 : : VLOG_F("DiscoveryViewer: Failed to set reuse port option."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
518 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
519 : : }
520 : : #endif
521 : :
522 : 8 : std::memset(&impl_->address, 0, sizeof(impl_->address));
523 : 8 : impl_->address.sin_family = AF_INET;
524 : 8 : impl_->address.sin_port = htons(kBroadcastBindPort);
525 : 8 : impl_->address.sin_addr.s_addr = htonl(INADDR_ANY);
526 : :
527 [ - + ]: 8 : if VUNLIKELY (::bind(impl_->sock, reinterpret_cast<sockaddr*>(&impl_->address), sizeof(impl_->address)) < 0) {
528 : : VLOG_F("DiscoveryViewer: Failed to bind socket."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
529 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
530 : : }
531 : :
532 : : #if VLINK_DISCOVERY_MULTICAST
533 : : ip_mreq mreq;
534 : 8 : std::memset(&mreq, 0, sizeof(mreq));
535 : :
536 [ - + ]: 8 : if (impl_->enable_native_discovery) {
537 : 0 : mreq.imr_multiaddr.s_addr = inet_addr(kBroadcastAddress);
538 : 0 : mreq.imr_interface.s_addr = inet_addr("127.0.0.1");
539 : :
540 [ # # ]: 0 : if VUNLIKELY (::setsockopt(impl_->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, reinterpret_cast<const char*>(&mreq),
541 : : sizeof(mreq)) < 0) {
542 : : VLOG_F("DiscoveryViewer: Failed to send multicast to 127.0.0.1."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
543 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
544 : : }
545 : : } else {
546 : 8 : mreq.imr_multiaddr.s_addr = inet_addr(kBroadcastAddress);
547 : 8 : mreq.imr_interface.s_addr = htonl(INADDR_ANY);
548 : :
549 [ - + ]: 8 : if VUNLIKELY (::setsockopt(impl_->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, reinterpret_cast<const char*>(&mreq),
550 : : sizeof(mreq)) < 0) {
551 : : #ifdef __QNX__
552 : : CLOG_F(
553 : : "DiscoveryViewer: Failed to set multicast, please add address [%s] to target device. "
554 : : "\nExamples(QNX): route add -net %s -interface eth0.",
555 : : kBroadcastAddress, kBroadcastAddress);
556 : : #elif defined(__APPLE__)
557 : : CLOG_F(
558 : : "DiscoveryViewer: Failed to set multicast, please add address [%s] to target device. "
559 : : "\nExamples(MACOS): route add -net %s -interface eth0.",
560 : : kBroadcastAddress, kBroadcastAddress);
561 : : #else
562 : : CLOG_F( // LCOV_EXCL_LINE GCOVR_EXCL_LINE
563 : : "DiscoveryViewer: Failed to set multicast, please add address [%s] to target device. "
564 : : "\nExamples(Linux): route add %s eth0.",
565 : : kBroadcastAddress, kBroadcastAddress);
566 : : #endif
567 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
568 : : }
569 : : }
570 : :
571 : : #else
572 : : int enable_broadcast = 1;
573 : :
574 : : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_BROADCAST, reinterpret_cast<const char*>(&enable_broadcast),
575 : : sizeof(enable_broadcast)) < 0) {
576 : : VLOG_F("DiscoveryViewer: Failed to enable broadcast.");
577 : : return;
578 : : }
579 : : #endif
580 : :
581 [ + - ]: 8 : impl_->buffer.resize(kBufferSize);
582 : 8 : std::memset(impl_->buffer.data(), 0, impl_->buffer.size());
583 : :
584 : 2 : static auto global_ip_set = []() {
585 : 2 : auto list = Utils::get_all_ipv4_address();
586 [ + - ]: 4 : return std::unordered_set<std::string>(list.begin(), list.end());
587 [ + + + - : 10 : }();
+ - - - ]
588 : :
589 : : // socket listen
590 [ + - ]: 16 : impl_->thread = std::thread([this]() {
591 : : sockaddr_in target_address;
592 : 8 : socklen_t target_address_len = sizeof(target_address);
593 : 8 : std::memset(&target_address, 0, target_address_len);
594 : :
595 : 8 : char target_ip[INET_ADDRSTRLEN] = {0};
596 : :
597 : : for (;;) {
598 [ + - ]: 824 : int size = ::recvfrom(impl_->sock, reinterpret_cast<char*>(impl_->buffer.data()), impl_->buffer.size(), 0,
599 : 824 : reinterpret_cast<sockaddr*>(&target_address), &target_address_len);
600 : :
601 [ + - + + ]: 824 : if VUNLIKELY (is_ready_to_quit()) {
602 : 8 : break;
603 : : }
604 : :
605 [ + + - + : 816 : if VUNLIKELY (size <= 0 || static_cast<size_t>(size) > impl_->buffer.size()) {
+ + ]
606 : 693 : continue;
607 : : }
608 : :
609 [ - + ]: 123 : if VUNLIKELY (::inet_ntop(AF_INET, &target_address.sin_addr, target_ip, INET_ADDRSTRLEN) == nullptr) {
610 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
611 : : }
612 : :
613 [ + - ]: 123 : if (impl_->filter_type == kFilterNative) {
614 [ + - + - : 123 : if (global_ip_set.count(target_ip) == 0) {
- + ]
615 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
616 : : }
617 : : }
618 : :
619 : 123 : std::string message = Bytes::shallow_copy(impl_->buffer.data(), size).to_string();
620 : :
621 [ + - - + ]: 123 : if VUNLIKELY (!is_running()) {
622 : : Utils::yield_cpu(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
623 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
624 : : }
625 : :
626 [ + - + - : 123 : post_task([this, message, target_ip_str = std::string(target_ip)]() {
+ - + - ]
627 : : #if VLINK_DISCOVERY_OFFLINE
628 : : if VUNLIKELY (Helpers::has_startwith(message, "offline")) {
629 : : auto offline_list_view = Helpers::split_view(message, '\n');
630 : :
631 : : if VUNLIKELY (offline_list_view.size() < 2) {
632 : : return;
633 : : }
634 : :
635 : : if VUNLIKELY (offline_list_view.at(0) != "offline") {
636 : : return;
637 : : }
638 : :
639 : : auto process_view = offline_list_view.at(1);
640 : :
641 : : auto process_list_view = Helpers::split_view(process_view, ':');
642 : :
643 : : if VUNLIKELY (process_list_view.size() < 3) {
644 : : return;
645 : : }
646 : :
647 : : uint32_t process_pid = 0;
648 : :
649 : : const std::string hostname = Helpers::unescape_field(process_list_view.at(0));
650 : :
651 : : auto process_pid_view = process_list_view.at(1);
652 : :
653 : : auto [ptr, error] =
654 : : std::from_chars(process_pid_view.data(), process_pid_view.data() + process_pid_view.size(), process_pid);
655 : :
656 : : if VUNLIKELY (error != std::errc{}) {
657 : : process_pid = 0;
658 : : }
659 : :
660 : : const std::string process_name = Helpers::unescape_field(process_list_view.at(2));
661 : :
662 : : process_offline(hostname, process_pid, process_name);
663 : :
664 : : return;
665 : : }
666 : : #endif
667 : :
668 : 123 : auto message_list_view = Helpers::split_view(message, '\n');
669 : :
670 [ - + ]: 123 : if VUNLIKELY (message_list_view.empty()) {
671 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
672 : : }
673 : :
674 : : {
675 [ + - ]: 123 : std::lock_guard lock(impl_->mtx);
676 : :
677 [ + + ]: 257 : for (auto str : message_list_view) {
678 : 134 : auto list = Helpers::split_view(str, ' ');
679 : :
680 [ - + ]: 134 : if VUNLIKELY (list.size() < 5) {
681 : 0 : continue;
682 : : }
683 : :
684 [ + - + - ]: 134 : ImplType type = convert_type(list.at(0));
685 [ + - ]: 134 : auto url_view = list.at(1);
686 [ + - ]: 134 : auto ser_type_view = list.at(2);
687 [ + - ]: 134 : auto schema_type_view = list.at(3);
688 [ + - ]: 134 : auto process_view = list.at(4);
689 : :
690 : 134 : uint32_t schema_type_num = 0;
691 : 134 : SchemaType schema_type = SchemaType::kUnknown;
692 : 134 : uint32_t process_pid = 0;
693 : 134 : double profiler = -1;
694 : :
695 : : {
696 [ + - ]: 134 : auto [ptr, error] = std::from_chars(schema_type_view.data(),
697 : 134 : schema_type_view.data() + schema_type_view.size(), schema_type_num);
698 : :
699 [ - + ]: 134 : if VUNLIKELY (error != std::errc{}) {
700 : 0 : schema_type_num = 0;
701 : : }
702 : :
703 [ + - ]: 134 : if (error == std::errc{}) {
704 : 134 : auto parsed_schema_type = static_cast<SchemaType>(schema_type_num);
705 : :
706 [ + - ]: 134 : if (SchemaData::is_valid_type(parsed_schema_type)) {
707 : 134 : schema_type = parsed_schema_type;
708 : : }
709 : : }
710 : : }
711 : :
712 : 134 : auto process_list_view = Helpers::split_view(process_view, ':');
713 : :
714 [ - + ]: 134 : if (process_list_view.size() < 3) {
715 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
716 : : }
717 : :
718 : 134 : std::string url = Helpers::unescape_field(url_view);
719 : 134 : std::string ser_type = Helpers::unescape_field(ser_type_view);
720 [ + - ]: 134 : std::string hostname = Helpers::unescape_field(process_list_view.at(0));
721 : :
722 [ - + ]: 134 : if (impl_->filter_type == kFilterAvailable) {
723 [ # # ]: 0 : if (hostname != impl_->native_hostname) {
724 [ # # # # ]: 0 : if (Url::is_local_type(url)) {
725 : 0 : continue;
726 : : }
727 : : }
728 [ + - ]: 134 : } else if (impl_->filter_type == kFilterNative) {
729 [ - + ]: 134 : if (hostname != impl_->native_hostname) {
730 : 0 : continue;
731 : : }
732 : : }
733 : :
734 [ + - ]: 134 : auto process_pid_view = process_list_view.at(1);
735 : :
736 : : {
737 [ + - ]: 134 : auto [ptr, error] = std::from_chars(process_pid_view.data(),
738 : 134 : process_pid_view.data() + process_pid_view.size(), process_pid);
739 : :
740 [ - + ]: 134 : if VUNLIKELY (error != std::errc{}) {
741 : 0 : process_pid = 0;
742 : : }
743 : : }
744 : :
745 [ + - ]: 134 : std::string process_name = Helpers::unescape_field(process_list_view.at(2));
746 : :
747 [ - + ]: 134 : if (process_list_view.size() >= 4) {
748 [ # # ]: 0 : auto profiler_view = process_list_view.at(3);
749 : :
750 : : try {
751 [ # # # # ]: 0 : profiler = std::stod(std::string(profiler_view));
752 [ - - ]: 0 : } catch (std::exception&) {
753 : 0 : profiler = -1;
754 : 0 : }
755 : : }
756 : :
757 [ + - - + : 134 : if VUNLIKELY (type == kUnknownImplType || url.empty()) {
- + ]
758 : 0 : continue;
759 : : }
760 : :
761 [ + - ]: 134 : int sort_index = Url::get_sort_index(url);
762 : :
763 [ - + ]: 134 : if VUNLIKELY (sort_index < 0) {
764 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
765 : : }
766 : :
767 [ + + ]: 134 : if (ser_type == "{}") {
768 : 23 : ser_type.clear();
769 : : }
770 : :
771 : 134 : Info info{sort_index, type,
772 : : url, ser_type,
773 [ + - + - : 268 : schema_type, {Process{type, hostname, process_pid, process_name, target_ip_str, profiler}}};
+ - + - +
- + - + +
- - ]
774 : :
775 [ + - ]: 134 : auto [iter, inserted] = impl_->info_map.try_emplace(std::move(info), ElapsedTimer{});
776 : 134 : iter->second.restart();
777 : :
778 [ + + ]: 134 : if (!inserted) {
779 : 21 : auto& existing_info = const_cast<Info&>(iter->first);
780 : 21 : existing_info.process_list[0].profiler = profiler;
781 : : }
782 : :
783 : 134 : impl_->info_dirty = true;
784 : :
785 [ + - ]: 134 : if (!url.empty()) {
786 : 134 : std::string merged_ser_type;
787 : 134 : SchemaType merged_schema_type = SchemaType::kUnknown;
788 : 134 : bool has_ser_conflict = false;
789 : 134 : bool has_schema_conflict = false;
790 : :
791 [ + + ]: 882 : for (const auto& [active_info, active_timer] : impl_->info_map) {
792 : : (void)active_timer;
793 : :
794 [ + + ]: 748 : if (active_info.url != url) {
795 : 613 : continue;
796 : : }
797 : :
798 [ + + ]: 135 : if (!active_info.ser_type.empty()) {
799 [ + + - + : 112 : if (merged_ser_type.empty() || merged_ser_type == "Bytes") {
+ + ]
800 [ + - ]: 111 : merged_ser_type = active_info.ser_type;
801 [ + - - + : 1 : } else if (active_info.ser_type != "Bytes" && active_info.ser_type != merged_ser_type) {
- + ]
802 : 0 : has_ser_conflict = true;
803 : : }
804 : : }
805 : :
806 [ + - - + ]: 135 : if (has_schema_conflict || active_info.schema_type == SchemaType::kUnknown) {
807 : 0 : continue;
808 : : }
809 : :
810 [ + + ]: 135 : if (merged_schema_type == SchemaType::kUnknown) {
811 : 134 : merged_schema_type = active_info.schema_type;
812 [ - + ]: 1 : } else if (merged_schema_type != active_info.schema_type) {
813 : 0 : merged_schema_type = SchemaType::kUnknown;
814 : 0 : has_schema_conflict = true;
815 : : }
816 : : }
817 : :
818 [ - + ]: 134 : if VUNLIKELY (has_ser_conflict) {
819 [ # # # # ]: 0 : CLOG_W(
820 : : "DiscoveryViewer: Different ser: url = %s, current_ser = %s, new_ser = %s, process_name = %s, "
821 : : "process_pid = %u.",
822 : : url.c_str(), merged_ser_type.c_str(), ser_type.c_str(), process_name.c_str(), process_pid);
823 : 0 : merged_ser_type.clear();
824 : : }
825 : :
826 [ + - ]: 134 : std::lock_guard ser_lock(impl_->ser_mtx);
827 : :
828 [ + + ]: 134 : if (!merged_ser_type.empty()) {
829 [ + - + - ]: 111 : impl_->ser_map[url] = merged_ser_type;
830 : : } else {
831 [ + - ]: 23 : impl_->ser_map.erase(url);
832 : : }
833 : :
834 [ + - ]: 134 : impl_->schema_type_map[url] = merged_schema_type;
835 : 134 : }
836 [ + - + - : 134 : }
+ - + - +
- + - ]
837 : 123 : }
838 : :
839 [ + - ]: 123 : report_list();
840 [ + - ]: 123 : });
841 [ + - ]: 939 : }
842 : 16 : });
843 [ + - ]: 8 : }
844 : :
845 : 8 : DiscoveryViewer::~DiscoveryViewer() {
846 : 8 : quit(true);
847 : :
848 [ + - ]: 8 : if VLIKELY (impl_->sock >= 0) {
849 : : #ifdef _WIN32
850 : : ::shutdown(impl_->sock, SD_BOTH);
851 : : #else
852 : 8 : ::shutdown(impl_->sock, SHUT_RDWR);
853 : : #endif
854 : : }
855 : :
856 : 8 : wait_for_quit();
857 : :
858 [ + - ]: 8 : if VLIKELY (impl_->thread.joinable()) {
859 : 8 : impl_->thread.join();
860 : : }
861 : :
862 [ + - ]: 8 : if VLIKELY (impl_->sock >= 0) {
863 : : #ifdef _WIN32
864 : : ::closesocket(impl_->sock);
865 : : #else
866 : 8 : ::close(impl_->sock);
867 : : #endif
868 : 8 : impl_->sock = -1;
869 : : }
870 : :
871 : : #ifdef _WIN32
872 : : ::WSACleanup();
873 : : #endif
874 : :
875 : 8 : impl_->buffer.clear();
876 : 8 : }
877 : :
878 : 1 : void DiscoveryViewer::register_callback(Callback&& callback) {
879 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
880 : 1 : impl_->callback = std::move(callback);
881 : 1 : }
882 : :
883 : 1 : std::vector<DiscoveryViewer::Info> DiscoveryViewer::get_info_list() {
884 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
885 [ + - ]: 2 : return impl_->info_list;
886 : 1 : }
887 : :
888 : 1 : std::string DiscoveryViewer::get_ser_type(const std::string& url) const {
889 [ + - ]: 1 : std::shared_lock lock(impl_->ser_mtx);
890 [ + - ]: 1 : auto iter = impl_->ser_map.find(url);
891 : :
892 [ - + ]: 1 : if VLIKELY (iter != impl_->ser_map.end()) {
893 [ # # ]: 0 : return iter->second;
894 : : }
895 : :
896 : 1 : return {};
897 : 1 : }
898 : :
899 : 1 : SchemaType DiscoveryViewer::get_schema_type(const std::string& url) const {
900 [ + - ]: 1 : std::shared_lock lock(impl_->ser_mtx);
901 [ + - ]: 1 : auto iter = impl_->schema_type_map.find(url);
902 : :
903 [ - + ]: 1 : if VLIKELY (iter != impl_->schema_type_map.end()) {
904 : 0 : return iter->second;
905 : : }
906 : :
907 : 1 : return SchemaType::kUnknown;
908 : 1 : }
909 : :
910 : 448 : size_t DiscoveryViewer::get_max_task_count() const { return kMaxTaskSize; }
911 : :
912 : 1344 : uint32_t DiscoveryViewer::get_max_elapsed_time() const { return kMaxElapsedTime; }
913 : :
914 : 1 : void DiscoveryViewer::on_begin() { MessageLoop::on_begin(); }
915 : :
916 : 1 : void DiscoveryViewer::on_end() { MessageLoop::on_end(); }
917 : :
918 : 163 : void DiscoveryViewer::process_timeout() {
919 : 163 : std::vector<DiscoveryViewer::Info> erase_list;
920 : :
921 : : {
922 [ + - ]: 163 : std::lock_guard lock(impl_->mtx);
923 : :
924 [ + + ]: 738 : for (const auto& [info, elapsed] : impl_->info_map) {
925 [ + + - + : 575 : if (elapsed.get() > kCollectInterval * 4 || !elapsed.is_active()) {
+ + ]
926 [ + - ]: 112 : erase_list.emplace_back(info);
927 : : }
928 : : }
929 : :
930 [ + + ]: 163 : if (!erase_list.empty()) {
931 [ + + ]: 187 : for (const auto& info : erase_list) {
932 [ + - ]: 112 : impl_->info_map.erase(info);
933 : : }
934 : :
935 : 75 : impl_->info_dirty = true;
936 : : }
937 : 163 : }
938 : :
939 [ + - ]: 163 : report_list();
940 : 163 : }
941 : :
942 : : // LCOV_EXCL_START GCOVR_EXCL_START
943 : : void DiscoveryViewer::process_offline(std::string_view hostname, uint32_t pid, std::string_view process_name) {
944 : : std::vector<std::pair<DiscoveryViewer::Info, DiscoveryViewer::Info>> updated_list;
945 : :
946 : : {
947 : : std::lock_guard lock(impl_->mtx);
948 : :
949 : : for (const auto& [info, timer] : impl_->info_map) {
950 : : auto it = std::find_if(info.process_list.begin(), info.process_list.end(),
951 : : [pid, process_name, hostname](const auto& process) {
952 : : return process.pid == pid && process.name == process_name && process.host == hostname;
953 : : });
954 : :
955 : : if (it != info.process_list.end()) {
956 : : DiscoveryViewer::Info new_info = info;
957 : : auto offset = std::distance(info.process_list.begin(), it);
958 : : auto it_copy = new_info.process_list.begin() + offset;
959 : :
960 : : new_info.process_list.erase(it_copy);
961 : : updated_list.emplace_back(info, std::move(new_info));
962 : : }
963 : : }
964 : :
965 : : if (updated_list.empty()) {
966 : : return;
967 : : }
968 : :
969 : : for (const auto& [old_info, new_info] : updated_list) {
970 : : impl_->info_map.erase(old_info);
971 : :
972 : : if (!new_info.process_list.empty()) {
973 : : auto [it, inserted] = impl_->info_map.try_emplace(new_info, ElapsedTimer{});
974 : : it->second.restart();
975 : : }
976 : : }
977 : :
978 : : impl_->info_dirty = true;
979 : : }
980 : :
981 : : report_list();
982 : : }
983 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
984 : :
985 : 198 : void DiscoveryViewer::sort_url() {
986 : 198 : impl_->info_list.clear();
987 : 198 : std::unordered_map<std::string, std::string> next_ser_map;
988 : 198 : std::unordered_map<std::string, SchemaType> next_schema_type_map;
989 : 198 : std::unordered_set<std::string> ser_conflict_urls;
990 : 198 : std::unordered_set<std::string> schema_conflict_urls;
991 [ + - ]: 198 : next_ser_map.reserve(impl_->info_map.size());
992 [ + - ]: 198 : next_schema_type_map.reserve(impl_->info_map.size());
993 [ + - ]: 198 : ser_conflict_urls.reserve(impl_->info_map.size());
994 [ + - ]: 198 : schema_conflict_urls.reserve(impl_->info_map.size());
995 : :
996 [ + + ]: 1190 : for (const auto& [info, timer] : impl_->info_map) {
997 [ + + ]: 992 : if VLIKELY (!impl_->info_list.empty()) {
998 : 797 : auto& merged = impl_->info_list.back();
999 : :
1000 [ + + ]: 797 : if (merged.url == info.url) {
1001 : 5 : merged.type |= info.type;
1002 [ + - ]: 5 : merged.process_list.insert(merged.process_list.end(), info.process_list.begin(), info.process_list.end());
1003 : :
1004 [ + - + - : 5 : if (!info.ser_type.empty() && ser_conflict_urls.count(info.url) == 0) {
+ - + - ]
1005 [ + - - + : 5 : if (merged.ser_type.empty() || (merged.ser_type == "Bytes" && info.ser_type != "Bytes")) {
- - - + ]
1006 [ # # ]: 0 : merged.ser_type = info.ser_type;
1007 [ + - - + : 5 : } else if (merged.ser_type != "Bytes" && info.ser_type != merged.ser_type) {
- + ]
1008 : 0 : merged.ser_type.clear();
1009 [ # # ]: 0 : ser_conflict_urls.emplace(info.url);
1010 : : }
1011 : : }
1012 : :
1013 [ + - + - : 5 : if (info.schema_type != SchemaType::kUnknown && schema_conflict_urls.count(info.url) == 0) {
+ - + - ]
1014 [ - + ]: 5 : if (merged.schema_type == SchemaType::kUnknown) {
1015 : 0 : merged.schema_type = info.schema_type;
1016 [ - + ]: 5 : } else if (merged.schema_type != info.schema_type) {
1017 : 0 : merged.schema_type = SchemaType::kUnknown;
1018 [ # # ]: 0 : schema_conflict_urls.emplace(info.url);
1019 : : }
1020 : : }
1021 : :
1022 : 5 : continue;
1023 : : }
1024 : : }
1025 : :
1026 [ + - ]: 987 : impl_->info_list.emplace_back(info);
1027 : : }
1028 : :
1029 [ + + ]: 1185 : for (auto& info : impl_->info_list) {
1030 [ + + ]: 987 : if (info.process_list.size() > 1) {
1031 [ + - ]: 5 : std::sort(info.process_list.begin(), info.process_list.end());
1032 [ + - + - ]: 5 : info.process_list.erase(std::unique(info.process_list.begin(), info.process_list.end(),
1033 : 5 : [](const auto& lhs, const auto& rhs) {
1034 [ # # # # : 0 : return lhs.type == rhs.type && lhs.host == rhs.host && lhs.pid == rhs.pid &&
# # ]
1035 [ - + - - ]: 5 : lhs.name == rhs.name && lhs.ip == rhs.ip;
1036 : : }),
1037 : 10 : info.process_list.end());
1038 : : }
1039 : :
1040 [ + - ]: 987 : auto& ser_type = next_ser_map[info.url];
1041 [ + - ]: 987 : auto& schema_type = next_schema_type_map[info.url];
1042 : :
1043 [ + - - + ]: 987 : if (ser_conflict_urls.count(info.url) != 0) {
1044 : 0 : ser_type.clear();
1045 [ - + - - : 987 : } else if ((ser_type.empty() || ser_type == "Bytes") && !info.ser_type.empty()) {
+ + + + ]
1046 [ + - ]: 823 : ser_type = info.ser_type;
1047 : : }
1048 : :
1049 [ + - - + ]: 987 : if (schema_conflict_urls.count(info.url) != 0) {
1050 : 0 : schema_type = SchemaType::kUnknown;
1051 [ + - ]: 987 : } else if (info.schema_type != SchemaType::kUnknown) {
1052 [ + - ]: 987 : if (schema_type == SchemaType::kUnknown) {
1053 : 987 : schema_type = info.schema_type;
1054 : : } else if (schema_type != info.schema_type) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1055 : : schema_type = SchemaType::kUnknown; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1056 : : schema_conflict_urls.emplace(info.url); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1057 : : }
1058 : : }
1059 : : }
1060 : :
1061 : : {
1062 [ + - ]: 198 : std::unique_lock lock(impl_->ser_mtx);
1063 : 198 : impl_->ser_map.swap(next_ser_map);
1064 : 198 : impl_->schema_type_map.swap(next_schema_type_map);
1065 : 198 : }
1066 : 198 : }
1067 : :
1068 : 286 : void DiscoveryViewer::report_list() {
1069 : 286 : Callback callback;
1070 : 286 : std::vector<Info> info_list;
1071 : :
1072 : : {
1073 [ + - ]: 286 : std::lock_guard lock(impl_->mtx);
1074 : :
1075 [ + + ]: 286 : if (!impl_->info_dirty) {
1076 : 88 : return;
1077 : : }
1078 : :
1079 [ + - ]: 198 : sort_url();
1080 : :
1081 : 198 : impl_->info_dirty = false;
1082 [ + - ]: 198 : callback = impl_->callback;
1083 [ + - ]: 198 : info_list = impl_->info_list;
1084 [ + + ]: 286 : }
1085 : :
1086 [ - + ]: 198 : if (callback) {
1087 [ # # ]: 0 : callback(info_list);
1088 : : }
1089 [ + + + + ]: 374 : }
1090 : :
1091 : : } // namespace vlink
|
