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 721 : Impl(dhtnet::ConnectionManager& cm, OnMessage onMessage, OnPeerStateChanged onPeer, OnDeviceConnected onDevice)
40 1442 : : connectionManager_(cm)
41 721 : , onMessage_(std::move(onMessage))
42 721 : , onPeerStateChanged_(std::move(onPeer))
43 1442 : , onDeviceConnected_(std::move(onDevice))
44 721 : {}
45 :
46 : void onChannelShutdown(const std::shared_ptr<dhtnet::ChannelSocket>& socket,
47 : const std::string& peerId,
48 : const DeviceId& device);
49 : };
50 :
51 721 : MessageChannelHandler::MessageChannelHandler(dhtnet::ConnectionManager& cm,
52 : OnMessage onMessage,
53 : OnPeerStateChanged onPeer,
54 721 : OnDeviceConnected onDeviceConnected)
55 : : ChannelHandlerInterface()
56 721 : , pimpl_(std::make_shared<Impl>(cm, std::move(onMessage), std::move(onPeer), std::move(onDeviceConnected)))
57 721 : {}
58 :
59 1442 : MessageChannelHandler::~MessageChannelHandler()
60 : {
61 721 : std::unique_lock lk(pimpl_->connectionsMtx_);
62 1334 : for (const auto& [peerId, _] : pimpl_->connections_) {
63 613 : pimpl_->onPeerStateChanged_(peerId, false);
64 : }
65 721 : auto connections = std::move(pimpl_->connections_);
66 721 : pimpl_->connections_.clear();
67 721 : lk.unlock();
68 1442 : }
69 :
70 : void
71 3617 : MessageChannelHandler::connect(const DeviceId& deviceId,
72 : const std::string&,
73 : ConnectCb&& cb,
74 : const std::string& connectionType,
75 : bool forceNewConnection)
76 : {
77 3617 : auto channelName = concat(MESSAGE_SCHEME, deviceId.to_view());
78 3617 : if (pimpl_->connectionManager_.isConnecting(deviceId, channelName)) {
79 1278 : JAMI_LOG("Already connecting to {}", deviceId);
80 1278 : return;
81 : }
82 2340 : pimpl_->connectionManager_
83 2340 : .connectDevice(deviceId, channelName, std::move(cb), false, forceNewConnection, connectionType);
84 3618 : }
85 :
86 : void
87 1352 : MessageChannelHandler::Impl::onChannelShutdown(const std::shared_ptr<dhtnet::ChannelSocket>& socket,
88 : const std::string& peerId,
89 : const DeviceId& device)
90 : {
91 1352 : std::lock_guard lk(connectionsMtx_);
92 1352 : auto peerIt = connections_.find(peerId);
93 1352 : if (peerIt == connections_.end()) {
94 82 : JAMI_WARNING("onChannelShutdown: No connections found for peer {}", peerId);
95 82 : return;
96 : }
97 1270 : auto connectionsIt = peerIt->second.find(device);
98 1270 : 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 1270 : auto& connections = connectionsIt->second;
103 1270 : auto conn = std::find(connections.begin(), connections.end(), socket);
104 1270 : if (conn != connections.end())
105 1270 : connections.erase(conn);
106 1270 : if (connections.empty()) {
107 636 : peerIt->second.erase(connectionsIt);
108 : }
109 1270 : if (peerIt->second.empty()) {
110 632 : connections_.erase(peerIt);
111 632 : onPeerStateChanged_(peerId, false);
112 : }
113 1352 : }
114 :
115 : std::shared_ptr<dhtnet::ChannelSocket>
116 16066 : MessageChannelHandler::getChannel(const std::string& peer, const DeviceId& deviceId) const
117 : {
118 16066 : std::lock_guard lk(pimpl_->connectionsMtx_);
119 16065 : auto it = pimpl_->connections_.find(peer);
120 16066 : if (it == pimpl_->connections_.end())
121 3608 : return nullptr;
122 12458 : auto deviceIt = it->second.find(deviceId);
123 12458 : if (deviceIt == it->second.end())
124 19 : return nullptr;
125 12439 : if (deviceIt->second.empty())
126 0 : return nullptr;
127 12439 : return deviceIt->second.back();
128 16066 : }
129 :
130 : std::vector<std::shared_ptr<dhtnet::ChannelSocket>>
131 3096 : MessageChannelHandler::getChannels(const std::string& peer) const
132 : {
133 3096 : std::vector<std::shared_ptr<dhtnet::ChannelSocket>> sockets;
134 3096 : std::lock_guard lk(pimpl_->connectionsMtx_);
135 3096 : auto it = pimpl_->connections_.find(peer);
136 3096 : if (it == pimpl_->connections_.end())
137 1902 : return sockets;
138 1194 : sockets.reserve(it->second.size());
139 2389 : for (auto& [deviceId, channels] : it->second) {
140 3345 : for (auto& channel : channels) {
141 2150 : sockets.push_back(channel);
142 : }
143 : }
144 1194 : return sockets;
145 3096 : }
146 :
147 : bool
148 1258 : MessageChannelHandler::onRequest(const std::shared_ptr<dht::crypto::Certificate>& cert, const std::string& /* name */)
149 : {
150 1258 : if (!cert || !cert->issuer)
151 0 : return false;
152 1256 : return true;
153 : }
154 :
155 : void
156 2506 : MessageChannelHandler::onReady(const std::shared_ptr<dht::crypto::Certificate>& cert,
157 : const std::string&,
158 : std::shared_ptr<dhtnet::ChannelSocket> socket)
159 : {
160 2506 : if (!cert || !cert->issuer)
161 0 : return;
162 2505 : auto peerId = cert->issuer->getId().toString();
163 2508 : auto device = cert->getLongId();
164 2505 : std::lock_guard lk(pimpl_->connectionsMtx_);
165 2508 : auto& connections = pimpl_->connections_[peerId];
166 2508 : bool newPeerConnection = connections.empty();
167 2504 : auto& deviceConnections = connections[device];
168 2509 : bool newDeviceConnection = deviceConnections.empty();
169 2507 : deviceConnections.push_back(socket);
170 2503 : if (newPeerConnection)
171 1242 : pimpl_->onPeerStateChanged_(peerId, true);
172 2509 : if (newDeviceConnection)
173 1259 : pimpl_->onDeviceConnected_(peerId, device);
174 :
175 2509 : socket->setOnRecv(dhtnet::buildMsgpackReader<Message>([onMessage = pimpl_->onMessage_, cert](Message&& msg) {
176 12392 : onMessage(cert, msg.t, msg.c);
177 12392 : return std::error_code();
178 : }));
179 :
180 5015 : socket->onShutdown(
181 5015 : [w = pimpl_->weak_from_this(), peerId, device, s = std::weak_ptr(socket)](const std::error_code& /*ec*/) {
182 2509 : if (auto shared = w.lock())
183 2509 : shared->onChannelShutdown(s.lock(), peerId, device);
184 2509 : });
185 2504 : }
186 :
187 : void
188 4 : MessageChannelHandler::closeChannel(const std::string& peer,
189 : const DeviceId& device,
190 : const std::shared_ptr<dhtnet::ChannelSocket>& conn)
191 : {
192 4 : if (!conn)
193 0 : return;
194 4 : std::unique_lock lk(pimpl_->connectionsMtx_);
195 4 : auto it = pimpl_->connections_.find(peer);
196 4 : if (it != pimpl_->connections_.end()) {
197 2 : auto deviceIt = it->second.find(device);
198 2 : if (deviceIt != it->second.end()) {
199 2 : auto& channels = deviceIt->second;
200 2 : channels.erase(std::remove(channels.begin(), channels.end(), conn), channels.end());
201 2 : if (channels.empty()) {
202 1 : it->second.erase(deviceIt);
203 1 : if (it->second.empty()) {
204 1 : pimpl_->connections_.erase(it);
205 1 : pimpl_->onPeerStateChanged_(peer, false);
206 : }
207 : }
208 : }
209 : }
210 4 : lk.unlock();
211 4 : conn->stop();
212 4 : }
213 :
214 : bool
215 12404 : MessageChannelHandler::sendMessage(const std::shared_ptr<dhtnet::ChannelSocket>& socket, const Message& message)
216 : {
217 12404 : if (!socket)
218 0 : return false;
219 12404 : msgpack::sbuffer buffer(UINT16_MAX); // Use max
220 12404 : msgpack::pack(buffer, message);
221 12403 : std::error_code ec;
222 12403 : auto sent = socket->write(reinterpret_cast<const uint8_t*>(buffer.data()), buffer.size(), ec);
223 12403 : if (ec) {
224 4 : JAMI_WARNING("Error sending message: {:s}", ec.message());
225 : }
226 12403 : return !ec && sent == buffer.size();
227 12403 : }
228 :
229 : } // namespace jami
|
