Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 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 : #include "jamidht/message_channel_handler.h"
18 :
19 : #include <dhtnet/channel_utils.h>
20 : #include <string_view>
21 :
22 : using namespace std::literals;
23 :
24 : static constexpr auto MESSAGE_SCHEME = "msg:"sv;
25 :
26 : namespace jami {
27 :
28 : using Key = std::pair<std::string, DeviceId>;
29 :
30 : struct MessageChannelHandler::Impl : public std::enable_shared_from_this<Impl>
31 : {
32 : dhtnet::ConnectionManager& connectionManager_;
33 : OnMessage onMessage_;
34 : OnPeerStateChanged onPeerStateChanged_;
35 : OnDeviceConnected onDeviceConnected_;
36 : std::recursive_mutex connectionsMtx_;
37 : std::map<std::string, std::map<DeviceId, std::vector<std::shared_ptr<dhtnet::ChannelSocket>>>> connections_;
38 :
39 796 : Impl(dhtnet::ConnectionManager& cm, OnMessage onMessage, OnPeerStateChanged onPeer, OnDeviceConnected onDevice)
40 1592 : : connectionManager_(cm)
41 796 : , onMessage_(std::move(onMessage))
42 796 : , onPeerStateChanged_(std::move(onPeer))
43 1592 : , onDeviceConnected_(std::move(onDevice))
44 796 : {}
45 :
46 : void onChannelShutdown(const std::shared_ptr<dhtnet::ChannelSocket>& socket,
47 : const std::string& peerId,
48 : const DeviceId& device);
49 : };
50 :
51 796 : MessageChannelHandler::MessageChannelHandler(dhtnet::ConnectionManager& cm,
52 : OnMessage onMessage,
53 : OnPeerStateChanged onPeer,
54 796 : OnDeviceConnected onDeviceConnected)
55 : : ChannelHandlerInterface()
56 796 : , pimpl_(std::make_shared<Impl>(cm, std::move(onMessage), std::move(onPeer), std::move(onDeviceConnected)))
57 796 : {}
58 :
59 1592 : MessageChannelHandler::~MessageChannelHandler()
60 : {
61 796 : std::unique_lock lk(pimpl_->connectionsMtx_);
62 1428 : for (const auto& [peerId, _] : pimpl_->connections_) {
63 632 : pimpl_->onPeerStateChanged_(peerId, false);
64 : }
65 796 : auto connections = std::move(pimpl_->connections_);
66 796 : pimpl_->connections_.clear();
67 796 : lk.unlock();
68 1592 : }
69 :
70 : void
71 3563 : MessageChannelHandler::connect(const DeviceId& deviceId,
72 : const std::string&,
73 : ConnectCb&& cb,
74 : const std::string& connectionType,
75 : bool forceNewConnection)
76 : {
77 3563 : auto channelName = concat(MESSAGE_SCHEME, deviceId.to_view());
78 3563 : if (pimpl_->connectionManager_.isConnecting(deviceId, channelName)) {
79 1174 : JAMI_LOG("Already connecting to {}", deviceId);
80 1174 : return;
81 : }
82 2389 : pimpl_->connectionManager_
83 2389 : .connectDevice(deviceId, channelName, std::move(cb), false, forceNewConnection, connectionType);
84 3563 : }
85 :
86 : void
87 1415 : MessageChannelHandler::Impl::onChannelShutdown(const std::shared_ptr<dhtnet::ChannelSocket>& socket,
88 : const std::string& peerId,
89 : const DeviceId& device)
90 : {
91 1415 : std::lock_guard lk(connectionsMtx_);
92 1415 : auto peerIt = connections_.find(peerId);
93 1415 : if (peerIt == connections_.end()) {
94 85 : JAMI_WARNING("onChannelShutdown: No connections found for peer {}", peerId);
95 85 : return;
96 : }
97 1331 : auto connectionsIt = peerIt->second.find(device);
98 1332 : if (connectionsIt == peerIt->second.end()) {
99 0 : JAMI_WARNING("onChannelShutdown: No connections found for device {} of peer {}", device.toString(), peerId);
100 0 : return;
101 : }
102 1332 : auto& connections = connectionsIt->second;
103 1332 : auto conn = std::find(connections.begin(), connections.end(), socket);
104 1332 : if (conn != connections.end())
105 1332 : connections.erase(conn);
106 1332 : if (connections.empty()) {
107 670 : peerIt->second.erase(connectionsIt);
108 : }
109 1330 : if (peerIt->second.empty()) {
110 664 : connections_.erase(peerIt);
111 665 : onPeerStateChanged_(peerId, false);
112 : }
113 1417 : }
114 :
115 : std::shared_ptr<dhtnet::ChannelSocket>
116 15809 : MessageChannelHandler::getChannel(const std::string& peer, const DeviceId& deviceId) const
117 : {
118 15809 : std::lock_guard lk(pimpl_->connectionsMtx_);
119 15810 : auto it = pimpl_->connections_.find(peer);
120 15809 : if (it == pimpl_->connections_.end())
121 3557 : return nullptr;
122 12252 : auto deviceIt = it->second.find(deviceId);
123 12253 : if (deviceIt == it->second.end())
124 15 : return nullptr;
125 12238 : if (deviceIt->second.empty())
126 0 : return nullptr;
127 12238 : return deviceIt->second.back();
128 15809 : }
129 :
130 : std::vector<std::shared_ptr<dhtnet::ChannelSocket>>
131 3256 : MessageChannelHandler::getChannels(const std::string& peer) const
132 : {
133 3256 : std::vector<std::shared_ptr<dhtnet::ChannelSocket>> sockets;
134 3256 : std::lock_guard lk(pimpl_->connectionsMtx_);
135 3256 : auto it = pimpl_->connections_.find(peer);
136 3256 : if (it == pimpl_->connections_.end())
137 1775 : return sockets;
138 1481 : sockets.reserve(it->second.size());
139 2967 : for (auto& [deviceId, channels] : it->second) {
140 3895 : for (auto& channel : channels) {
141 2409 : sockets.push_back(channel);
142 : }
143 : }
144 1481 : return sockets;
145 3256 : }
146 :
147 : bool
148 1309 : MessageChannelHandler::onRequest(const std::shared_ptr<dht::crypto::Certificate>& cert, const std::string& /* name */)
149 : {
150 1309 : if (!cert || !cert->issuer)
151 0 : return false;
152 1310 : return true;
153 : }
154 :
155 : void
156 2599 : MessageChannelHandler::onReady(const std::shared_ptr<dht::crypto::Certificate>& cert,
157 : const std::string&,
158 : std::shared_ptr<dhtnet::ChannelSocket> socket)
159 : {
160 2599 : if (!cert || !cert->issuer)
161 0 : return;
162 2599 : auto peerId = cert->issuer->getId().toString();
163 2595 : auto device = cert->getLongId();
164 2596 : std::lock_guard lk(pimpl_->connectionsMtx_);
165 2599 : auto& connections = pimpl_->connections_[peerId];
166 2599 : bool newPeerConnection = connections.empty();
167 2596 : auto& deviceConnections = connections[device];
168 2595 : bool newDeviceConnection = deviceConnections.empty();
169 2597 : deviceConnections.push_back(socket);
170 2594 : if (newPeerConnection)
171 1293 : pimpl_->onPeerStateChanged_(peerId, true);
172 2599 : if (newDeviceConnection)
173 1310 : pimpl_->onDeviceConnected_(peerId, device);
174 :
175 2599 : socket->setOnRecv(dhtnet::buildMsgpackReader<Message>([onMessage = pimpl_->onMessage_, cert](Message&& msg) {
176 12198 : onMessage(cert, msg.t, msg.c);
177 12197 : return std::error_code();
178 : }));
179 :
180 5193 : socket->onShutdown(
181 5193 : [w = pimpl_->weak_from_this(), peerId, device, s = std::weak_ptr(socket)](const std::error_code& /*ec*/) {
182 2598 : if (auto shared = w.lock())
183 2596 : shared->onChannelShutdown(s.lock(), peerId, device);
184 2600 : });
185 2596 : }
186 :
187 : void
188 2 : MessageChannelHandler::closeChannel(const std::string& peer,
189 : const DeviceId& device,
190 : const std::shared_ptr<dhtnet::ChannelSocket>& conn)
191 : {
192 2 : if (!conn)
193 0 : return;
194 2 : std::unique_lock lk(pimpl_->connectionsMtx_);
195 2 : auto it = pimpl_->connections_.find(peer);
196 2 : if (it != pimpl_->connections_.end()) {
197 0 : auto deviceIt = it->second.find(device);
198 0 : if (deviceIt != it->second.end()) {
199 0 : auto& channels = deviceIt->second;
200 0 : channels.erase(std::remove(channels.begin(), channels.end(), conn), channels.end());
201 0 : if (channels.empty()) {
202 0 : it->second.erase(deviceIt);
203 0 : if (it->second.empty()) {
204 0 : pimpl_->connections_.erase(it);
205 0 : pimpl_->onPeerStateChanged_(peer, false);
206 : }
207 : }
208 : }
209 : }
210 2 : lk.unlock();
211 2 : conn->stop();
212 2 : }
213 :
214 : bool
215 12203 : MessageChannelHandler::sendMessage(const std::shared_ptr<dhtnet::ChannelSocket>& socket, const Message& message)
216 : {
217 12203 : if (!socket)
218 0 : return false;
219 12203 : msgpack::sbuffer buffer(UINT16_MAX); // Use max
220 12204 : msgpack::pack(buffer, message);
221 12203 : std::error_code ec;
222 12203 : auto sent = socket->write(reinterpret_cast<const uint8_t*>(buffer.data()), buffer.size(), ec);
223 12204 : if (ec) {
224 2 : JAMI_WARNING("Error sending message: {:s}", ec.message());
225 : }
226 12204 : return !ec && sent == buffer.size();
227 12204 : }
228 :
229 : } // namespace jami
|
