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 "./shm_client_impl.h"
25 : :
26 : : #include <utility>
27 : :
28 : : #include "./base/elapsed_timer.h"
29 : : #include "./base/helpers.h"
30 : :
31 : : namespace vlink {
32 : :
33 : : // ShmClientImpl
34 [ + - + - ]: 24 : ShmClientImpl::ShmClientImpl(const ShmConf& conf) : conf_(conf) {}
35 : :
36 : 24 : void ShmClientImpl::init() {
37 [ + + + - : 24 : static auto& factory = ShmFactory::get();
+ - - - ]
38 : :
39 : 24 : conf_.hash_code = Helpers::get_hash_code(conf_.event);
40 : :
41 [ + - ]: 48 : object_ = factory.get_object<Object>(
42 [ + - ]: 48 : {kImplType, conf_.address, conf_.domain, conf_.depth, conf_.history, conf_.wait, conf_.event, nullptr});
43 : :
44 : 24 : object_->add_impl(this);
45 : :
46 [ + - + - ]: 45 : object_->register_server_connect_callback(this, [this](bool) { ClientImpl::update_connected(); });
47 : :
48 : 24 : ClientImpl::update_connected();
49 : 24 : }
50 : :
51 : 24 : void ShmClientImpl::deinit() {
52 : 24 : detach();
53 : :
54 [ + - ]: 24 : if (object_) {
55 : 24 : object_->cancel_calls(this);
56 : 24 : object_->remove_impl(this);
57 : : }
58 : 24 : }
59 : :
60 : 24 : void ShmClientImpl::interrupt() {
61 : 24 : ClientImpl::interrupt();
62 : :
63 [ + - ]: 24 : if (object_) {
64 : 24 : object_->cancel_calls(this);
65 : : }
66 : :
67 : 24 : ack_manager_.clear();
68 : 24 : }
69 : :
70 : 25 : bool ShmClientImpl::is_support_loan() const { return true; }
71 : :
72 : 98 : Bytes ShmClientImpl::loan(int64_t size) { return object_->loan(static_cast<uint64_t>(conf_.hash_code), size); }
73 : :
74 : 33 : bool ShmClientImpl::return_loan(const Bytes& bytes) { return object_->release(bytes); }
75 : :
76 : 0 : const Conf* ShmClientImpl::get_conf() const { return &conf_; }
77 : :
78 : 2 : const AbstractNode* ShmClientImpl::get_abstract_node() const { return object_.get(); }
79 : :
80 : 185 : bool ShmClientImpl::is_connected() const { return object_->is_connected(); }
81 : :
82 : 2 : void ShmClientImpl::detect_connected(ConnectCallback&& callback) {
83 : 2 : object_->enable_detect_timer();
84 : :
85 : 2 : ClientImpl::detect_connected(std::move(callback));
86 : 2 : }
87 : :
88 : 65 : bool ShmClientImpl::wait_for_connected(std::chrono::milliseconds timeout) {
89 : 65 : object_->enable_detect_timer();
90 : :
91 : 65 : return ClientImpl::wait_for_connected(timeout);
92 : : }
93 : :
94 : 90 : bool ShmClientImpl::call(const Bytes& req_data, MsgCallback&& callback, std::chrono::milliseconds timeout) {
95 [ + + ]: 90 : if VUNLIKELY (!callback) {
96 [ + - ]: 3 : return object_->call(this, static_cast<uint64_t>(conf_.hash_code), req_data);
97 : : }
98 : :
99 [ + + ]: 87 : if (timeout.count() != 0) {
100 : 49 : ack_manager_.reset_interrupted();
101 : :
102 : 49 : ElapsedTimer timer;
103 : 49 : timer.start();
104 : :
105 [ + - + + ]: 49 : if (!wait_for_connected(timeout)) {
106 [ + - ]: 33 : object_->release(req_data);
107 : :
108 : 33 : return false;
109 : : }
110 : :
111 : 16 : auto elapsed = timer.get();
112 : :
113 [ + - - + : 16 : if (timeout.count() > 0 && elapsed >= timeout.count()) {
- + ]
114 [ # # ]: 0 : object_->release(req_data);
115 : :
116 : 0 : return false;
117 : : }
118 : :
119 : 16 : auto ack_request = ack_manager_.create_request();
120 : :
121 : 16 : auto ack_function = [ack_request, callback = std::move(callback)](const Bytes& resp_data) mutable {
122 [ + - ]: 30 : AckManager::notify(ack_request, [&callback, &resp_data]() { callback(resp_data); });
123 : 47 : };
124 : :
125 : 16 : uint64_t response_seq = 0;
126 : 16 : bool has_response_seq = false;
127 : 16 : bool submitted = false;
128 : :
129 [ + - ]: 16 : bool ret = ack_manager_.process(ack_request, timeout.count() - elapsed,
130 : 16 : [this, &req_data, &response_seq, &has_response_seq, &submitted,
131 : 16 : ack_function = std::move(ack_function)]() mutable {
132 : 16 : submitted = true;
133 : 16 : has_response_seq =
134 [ + - ]: 32 : object_->call(this, static_cast<uint64_t>(conf_.hash_code), req_data,
135 : 16 : std::move(ack_function), &response_seq, false);
136 : :
137 : 16 : return has_response_seq;
138 : : });
139 : :
140 [ - + ]: 16 : if (!submitted) {
141 [ # # ]: 0 : object_->release(req_data);
142 : : }
143 : :
144 [ + + + - : 16 : if VUNLIKELY (!ret && has_response_seq) {
+ + ]
145 [ + - ]: 1 : object_->remove_response_callback(response_seq);
146 : : }
147 : :
148 : 16 : return ret;
149 : 49 : }
150 : :
151 : 38 : return object_->call(this, static_cast<uint64_t>(conf_.hash_code), req_data, std::move(callback));
152 : : }
153 : :
154 : : } // namespace vlink
|
