Line data Source code
1 : /*
2 : * Copyright (C) 2004-2024 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "message_engine.h"
19 : #include "sip/sipaccountbase.h"
20 : #include "manager.h"
21 : #include "fileutils.h"
22 :
23 : #include "client/ring_signal.h"
24 : #include "jami/account_const.h"
25 :
26 : #include <opendht/thread_pool.h>
27 : #include <fmt/std.h>
28 :
29 : #include <fstream>
30 :
31 : namespace jami {
32 : namespace im {
33 :
34 808 : MessageEngine::MessageEngine(SIPAccountBase& acc, const std::filesystem::path& path)
35 808 : : account_(acc)
36 808 : , savePath_(path)
37 808 : , ioContext_(Manager::instance().ioContext())
38 808 : , saveTimer_(*ioContext_)
39 : {
40 808 : dhtnet::fileutils::check_dir(savePath_.parent_path());
41 808 : }
42 :
43 : MessageToken
44 18589 : MessageEngine::sendMessage(const std::string& to,
45 : const std::string& deviceId,
46 : const std::map<std::string, std::string>& payloads,
47 : uint64_t refreshToken)
48 : {
49 18589 : if (payloads.empty() or to.empty())
50 0 : return 0;
51 18590 : MessageToken token = 0;
52 : {
53 18590 : std::lock_guard lock(messagesMutex_);
54 18589 : auto& peerMessages = deviceId.empty() ? messages_[to] : messagesDevices_[deviceId];
55 18589 : if (refreshToken != 0) {
56 16305 : for (auto& m : peerMessages) {
57 2612 : if (m.token == refreshToken) {
58 2611 : token = refreshToken;
59 2611 : m.to = to;
60 2611 : m.payloads = payloads;
61 2611 : m.status = MessageStatus::IDLE;
62 2611 : break;
63 : }
64 : }
65 : }
66 18590 : if (token == 0) {
67 15980 : token = std::uniform_int_distribution<MessageToken> {1, JAMI_ID_MAX_VAL}(account_.rand);
68 15976 : auto& m = peerMessages.emplace_back(Message {token});
69 15976 : m.to = to;
70 15977 : m.payloads = payloads;
71 : }
72 18589 : scheduleSave();
73 18591 : }
74 37182 : ioContext_->post([this, to, deviceId]() { retrySend(to, deviceId, true); });
75 18591 : return token;
76 : }
77 :
78 : void
79 4169 : MessageEngine::onPeerOnline(const std::string& peer,
80 : const std::string& deviceId,
81 : bool retryOnTimeout)
82 : {
83 4169 : retrySend(peer, deviceId, retryOnTimeout);
84 4166 : }
85 :
86 : void
87 22760 : MessageEngine::retrySend(const std::string& peer, const std::string& deviceId, bool retryOnTimeout)
88 : {
89 22760 : if (account_.getRegistrationState() != RegistrationState::REGISTERED)
90 2877 : return;
91 : struct PendingMsg {
92 : MessageToken token;
93 : std::string to;
94 : std::map<std::string, std::string> payloads;
95 : };
96 22758 : std::vector<PendingMsg> pending {};
97 22757 : auto now = clock::now();
98 : {
99 22757 : std::lock_guard lock(messagesMutex_);
100 22758 : auto& m = deviceId.empty() ? messages_ : messagesDevices_;
101 22758 : auto p = m.find(deviceId.empty() ? peer : deviceId);
102 22755 : if (p == m.end())
103 2877 : return;
104 19880 : auto& messages = p->second;
105 :
106 37633 : for (auto& m: messages) {
107 17752 : if (m.status == MessageStatus::IDLE) {
108 17705 : m.status = MessageStatus::SENDING;
109 17705 : m.retried++;
110 17705 : m.last_op = now;
111 17705 : pending.emplace_back(PendingMsg {m.token, m.to, m.payloads});
112 : }
113 : }
114 22757 : }
115 : // avoid locking while calling callback
116 37586 : for (const auto& p : pending) {
117 53118 : JAMI_DEBUG("[message {:d}] Reattempt sending", p.token);
118 17706 : if (p.payloads.find("application/im-gitmessage-id") == p.payloads.end())
119 747 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
120 747 : account_.getAccountID(),
121 : "",
122 747 : p.to,
123 1494 : std::to_string(p.token),
124 : (int) libjami::Account::MessageStates::SENDING);
125 17706 : account_.sendMessage(p.to, deviceId, p.payloads, p.token, retryOnTimeout, false);
126 : }
127 22758 : }
128 :
129 : MessageStatus
130 0 : MessageEngine::getStatus(MessageToken t) const
131 : {
132 0 : std::lock_guard lock(messagesMutex_);
133 0 : for (const auto& p : messages_) {
134 0 : for (const auto& m : p.second) {
135 0 : if (m.token == t)
136 0 : return m.status;
137 : }
138 : }
139 0 : return MessageStatus::UNKNOWN;
140 0 : }
141 :
142 : void
143 18001 : MessageEngine::onMessageSent(const std::string& peer,
144 : MessageToken token,
145 : bool ok,
146 : const std::string& deviceId)
147 : {
148 54003 : JAMI_DEBUG("[message {:d}] Message sent: {:s}", token, ok ? "success"sv : "failure"sv);
149 18001 : std::lock_guard lock(messagesMutex_);
150 18001 : auto& m = deviceId.empty() ? messages_ : messagesDevices_;
151 :
152 18001 : auto p = m.find(deviceId.empty() ? peer : deviceId);
153 18001 : if (p == m.end()) {
154 0 : JAMI_WARNING("onMessageSent: Peer not found: id:{} device:{}", peer, deviceId);
155 0 : return;
156 : }
157 :
158 18001 : auto f = std::find_if(p->second.begin(), p->second.end(), [&](const Message& m) {
159 18064 : return m.token == token;
160 : });
161 18001 : if (f != p->second.end()) {
162 35448 : auto emit = f->payloads.find("application/im-gitmessage-id")
163 35448 : == f->payloads.end();
164 17724 : if (f->status == MessageStatus::SENDING) {
165 17675 : if (ok) {
166 15653 : f->status = MessageStatus::SENT;
167 46959 : JAMI_LOG("[message {:d}] Status changed to SENT", token);
168 15653 : if (emit)
169 419 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
170 419 : account_.getAccountID(),
171 : "",
172 419 : f->to,
173 838 : std::to_string(token),
174 : static_cast<int>(libjami::Account::MessageStates::SENT));
175 15653 : p->second.erase(f);
176 15653 : scheduleSave();
177 2022 : } else if (f->retried >= MAX_RETRIES) {
178 6 : f->status = MessageStatus::FAILURE;
179 18 : JAMI_WARNING("[message {:d}] Status changed to FAILURE", token);
180 6 : if (emit)
181 0 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(
182 0 : account_.getAccountID(),
183 : "",
184 0 : f->to,
185 0 : std::to_string(token),
186 : static_cast<int>(libjami::Account::MessageStates::FAILURE));
187 6 : p->second.erase(f);
188 6 : scheduleSave();
189 : } else {
190 2016 : f->status = MessageStatus::IDLE;
191 6048 : JAMI_DEBUG("[message {:d}] Status changed to IDLE", token);
192 : }
193 : } else {
194 147 : JAMI_DEBUG("[message {:d}] State is not SENDING", token);
195 : }
196 : } else {
197 831 : JAMI_DEBUG("[message {:d}] Unable to find message", token);
198 : }
199 18001 : }
200 :
201 : void
202 25 : MessageEngine::load()
203 : {
204 : try {
205 25 : decltype(messages_) root;
206 : {
207 25 : std::lock_guard lock(dhtnet::fileutils::getFileLock(savePath_));
208 25 : std::ifstream file;
209 25 : file.exceptions(std::ifstream::failbit | std::ifstream::badbit);
210 25 : file.open(savePath_);
211 1 : if (file.is_open()) {
212 1 : msgpack::unpacker up;
213 1 : up.reserve_buffer(UINT16_MAX);
214 1 : while (file.read(up.buffer(), UINT16_MAX)) {
215 0 : up.buffer_consumed(file.gcount());
216 0 : msgpack::object_handle oh;
217 0 : if (up.next(oh)) {
218 0 : root = oh.get().as<std::map<std::string, std::list<Message>>>();
219 0 : break;
220 : }
221 0 : up.reserve_buffer(UINT16_MAX);
222 0 : }
223 1 : }
224 50 : }
225 0 : std::lock_guard lock(messagesMutex_);
226 0 : messages_ = std::move(root);
227 0 : if (not messages_.empty()) {
228 0 : JAMI_LOG("[Account {}] Loaded {} messages from {}",
229 : account_.getAccountID(),
230 : messages_.size(),
231 : savePath_);
232 : }
233 50 : } catch (const std::exception& e) {
234 75 : JAMI_LOG("[Account {}] Unable to load messages from {}: {}",
235 : account_.getAccountID(),
236 : savePath_,
237 : e.what());
238 25 : }
239 25 : }
240 :
241 : void
242 853 : MessageEngine::save() const
243 : {
244 853 : std::lock_guard lock(messagesMutex_);
245 853 : save_();
246 853 : }
247 :
248 : void
249 34247 : MessageEngine::scheduleSave()
250 : {
251 34247 : saveTimer_.expires_after(std::chrono::seconds(5));
252 34249 : saveTimer_.async_wait([this, w = account_.weak_from_this()](const std::error_code& ec) {
253 34250 : if (!ec)
254 852 : if (auto acc = w.lock())
255 852 : save();
256 34250 : });
257 34249 : }
258 :
259 : void
260 853 : MessageEngine::save_() const
261 : {
262 : try {
263 853 : std::ofstream file;
264 853 : file.exceptions(std::ifstream::failbit | std::ifstream::badbit);
265 853 : file.open(savePath_, std::ios::trunc);
266 853 : if (file.is_open())
267 853 : msgpack::pack(file, messages_);
268 853 : } catch (const std::exception& e) {
269 0 : JAMI_ERROR("[Account {}] Unable to serialize pending messages: {}",
270 : account_.getAccountID(), e.what());
271 0 : }
272 853 : }
273 :
274 : } // namespace im
275 : } // namespace jami
|
