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