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 :
18 : #include "libav_deps.h" // MUST BE INCLUDED FIRST
19 :
20 : #include "audio_rtp_session.h"
21 :
22 : #include "logger.h"
23 :
24 : #include "audio_receive_thread.h"
25 : #include "audio_sender.h"
26 : #include "socket_pair.h"
27 : #include "media_recorder.h"
28 : #include "media_encoder.h"
29 : #include "media_device.h"
30 : #include "media_const.h"
31 :
32 : #include "audio/audio_input.h"
33 : #include "audio/ringbufferpool.h"
34 : #include "client/videomanager.h"
35 : #include "manager.h"
36 :
37 : #include <asio/io_context.hpp>
38 : #include <asio/post.hpp>
39 :
40 : namespace jami {
41 :
42 397 : AudioRtpSession::AudioRtpSession(const std::string& callId,
43 : const std::string& streamId,
44 397 : const std::shared_ptr<MediaRecorder>& rec)
45 : : RtpSession(callId, streamId, MediaType::MEDIA_AUDIO)
46 1129 : , rtcpCheckerThread_([] { return true; }, [this] { processRtcpChecker(); }, [] {})
47 :
48 : {
49 397 : recorder_ = rec;
50 1588 : JAMI_DEBUG("Created Audio RTP session: {} - stream id {}", fmt::ptr(this), streamId_);
51 :
52 : // don't move this into the initializer list or Cthulus will emerge
53 397 : ringbuffer_ = Manager::instance().getRingBufferPool().createRingBuffer(streamId_);
54 397 : }
55 :
56 397 : AudioRtpSession::~AudioRtpSession()
57 : {
58 397 : deinitRecorder();
59 397 : stop();
60 1588 : JAMI_DEBUG("Destroyed Audio RTP session: {} - stream id {}", fmt::ptr(this), streamId_);
61 397 : }
62 :
63 : void
64 197 : AudioRtpSession::startSender()
65 : {
66 197 : std::lock_guard lock(mutex_);
67 788 : JAMI_DEBUG("Start audio RTP sender: input [{}] - muted [{}]", input_, muteState_ ? "YES" : "NO");
68 :
69 197 : if (not send_.enabled or send_.hold) {
70 0 : JAMI_WARNING("Audio sending disabled");
71 0 : if (sender_) {
72 0 : if (socketPair_)
73 0 : socketPair_->interrupt();
74 0 : if (audioInput_)
75 0 : audioInput_->detach(sender_.get());
76 0 : sender_.reset();
77 : }
78 0 : return;
79 : }
80 :
81 197 : if (sender_)
82 0 : JAMI_WARNING("Restarting audio sender");
83 197 : if (audioInput_)
84 0 : audioInput_->detach(sender_.get());
85 :
86 197 : bool fileAudio = !input_.empty() && input_.find("file://") != std::string::npos;
87 197 : auto audioInputId = streamId_;
88 197 : if (fileAudio) {
89 0 : auto suffix = input_;
90 : static const std::string& sep = libjami::Media::VideoProtocolPrefix::SEPARATOR;
91 0 : const auto pos = input_.find(sep);
92 0 : if (pos != std::string::npos) {
93 0 : suffix = input_.substr(pos + sep.size());
94 : }
95 0 : audioInputId = suffix;
96 0 : }
97 :
98 : // sender sets up input correctly, we just keep a reference in case startSender is called
99 197 : audioInput_ = jami::getAudioInput(audioInputId);
100 197 : audioInput_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
101 0 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
102 0 : if (auto shared = w.lock())
103 0 : shared->attachLocalRecorder(ms);
104 0 : });
105 0 : });
106 197 : audioInput_->setMuted(muteState_);
107 197 : audioInput_->setSuccessfulSetupCb(onSuccessfulSetup_);
108 197 : if (!fileAudio) {
109 197 : auto newParams = audioInput_->switchInput(input_);
110 : try {
111 197 : if (newParams.valid() && newParams.wait_for(NEWPARAMS_TIMEOUT) == std::future_status::ready) {
112 197 : localAudioParams_ = newParams.get();
113 : } else {
114 0 : JAMI_ERROR("No valid new audio parameters");
115 0 : return;
116 : }
117 0 : } catch (const std::exception& e) {
118 0 : JAMI_ERROR("Exception while retrieving audio parameters: {}", e.what());
119 0 : return;
120 0 : }
121 197 : }
122 197 : if (streamId_ != audioInput_->getId())
123 0 : Manager::instance().getRingBufferPool().bindHalfDuplexOut(streamId_, audioInput_->getId());
124 :
125 197 : send_.fecEnabled = true;
126 :
127 : // be sure to not send any packets before saving last RTP seq value
128 197 : socketPair_->stopSendOp();
129 197 : if (sender_)
130 0 : initSeqVal_ = sender_->getLastSeqValue() + 1;
131 : try {
132 197 : sender_.reset();
133 197 : socketPair_->stopSendOp(false);
134 197 : sender_.reset(new AudioSender(getRemoteRtpUri(), send_, *socketPair_, initSeqVal_, mtu_));
135 0 : } catch (const MediaEncoderException& e) {
136 0 : JAMI_ERROR("{}", e.what());
137 0 : send_.enabled = false;
138 0 : }
139 :
140 197 : if (voiceCallback_)
141 197 : sender_->setVoiceCallback(voiceCallback_);
142 :
143 : // NOTE do after sender/encoder are ready
144 197 : auto codec = std::static_pointer_cast<SystemAudioCodecInfo>(send_.codec);
145 197 : audioInput_->setFormat(codec->audioformat);
146 197 : audioInput_->attach(sender_.get());
147 :
148 197 : if (not rtcpCheckerThread_.isRunning())
149 197 : rtcpCheckerThread_.start();
150 197 : }
151 :
152 : void
153 0 : AudioRtpSession::restartSender()
154 : {
155 0 : std::lock_guard lock(mutex_);
156 : // ensure that start has been called before restart
157 0 : if (not socketPair_) {
158 0 : return;
159 : }
160 :
161 0 : startSender();
162 0 : }
163 :
164 : void
165 197 : AudioRtpSession::startReceiver()
166 : {
167 197 : if (socketPair_)
168 197 : socketPair_->setReadBlockingMode(true);
169 :
170 197 : if ((not receive_.enabled) or receive_.hold) {
171 0 : JAMI_WARNING("Audio receiving disabled");
172 0 : receiveThread_.reset();
173 0 : return;
174 : }
175 :
176 197 : if (receiveThread_)
177 0 : JAMI_WARNING("Restarting audio receiver");
178 :
179 197 : auto accountAudioCodec = std::static_pointer_cast<SystemAudioCodecInfo>(receive_.codec);
180 197 : receiveThread_.reset(
181 197 : new AudioReceiveThread(streamId_, accountAudioCodec->audioformat, receive_.receiving_sdp, mtu_));
182 :
183 197 : receiveThread_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
184 0 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
185 0 : if (auto shared = w.lock())
186 0 : shared->attachRemoteRecorder(ms);
187 0 : });
188 0 : });
189 197 : receiveThread_->addIOContext(*socketPair_);
190 197 : receiveThread_->setSuccessfulSetupCb(onSuccessfulSetup_);
191 197 : receiveThread_->startReceiver();
192 :
193 : // Make the default ring buffer read the audio from the stream
194 394 : Manager::instance().getRingBufferPool().bindHalfDuplexOut(RingBufferPool::DEFAULT_ID, streamId_);
195 197 : }
196 :
197 : void
198 197 : AudioRtpSession::start(std::unique_ptr<dhtnet::IceSocket> rtp_sock, std::unique_ptr<dhtnet::IceSocket> rtcp_sock)
199 : {
200 197 : std::lock_guard lock(mutex_);
201 :
202 197 : if (not send_.enabled and not receive_.enabled) {
203 0 : stop();
204 0 : return;
205 : }
206 :
207 : try {
208 197 : if (rtp_sock and rtcp_sock) {
209 192 : if (send_.addr) {
210 192 : rtp_sock->setDefaultRemoteAddress(send_.addr);
211 : }
212 :
213 192 : auto& rtcpAddr = send_.rtcp_addr ? send_.rtcp_addr : send_.addr;
214 192 : if (rtcpAddr) {
215 192 : rtcp_sock->setDefaultRemoteAddress(rtcpAddr);
216 : }
217 :
218 192 : socketPair_.reset(new SocketPair(std::move(rtp_sock), std::move(rtcp_sock)));
219 : } else {
220 5 : socketPair_.reset(new SocketPair(getRemoteRtpUri().c_str(), receive_.addr.getPort()));
221 : }
222 :
223 197 : if (send_.crypto and receive_.crypto) {
224 788 : socketPair_->createSRTP(receive_.crypto.getCryptoSuite().c_str(),
225 394 : receive_.crypto.getSrtpKeyInfo().c_str(),
226 394 : send_.crypto.getCryptoSuite().c_str(),
227 394 : send_.crypto.getSrtpKeyInfo().c_str());
228 : }
229 0 : } catch (const std::runtime_error& e) {
230 0 : JAMI_ERROR("Socket creation failed: {}", e.what());
231 0 : return;
232 0 : }
233 :
234 197 : startSender();
235 197 : startReceiver();
236 197 : }
237 :
238 : void
239 849 : AudioRtpSession::stop()
240 : {
241 849 : std::lock_guard lock(mutex_);
242 :
243 3394 : JAMI_DEBUG("[{}] Stopping receiver", fmt::ptr(this));
244 :
245 849 : if (not receiveThread_)
246 652 : return;
247 :
248 197 : if (socketPair_)
249 197 : socketPair_->setReadBlockingMode(false);
250 :
251 197 : receiveThread_->stopReceiver();
252 :
253 : // Unbind the default ring buffer from this audio stream
254 394 : Manager::instance().getRingBufferPool().unBindHalfDuplexOut(RingBufferPool::DEFAULT_ID, streamId_);
255 :
256 197 : if (audioInput_)
257 197 : audioInput_->detach(sender_.get());
258 :
259 197 : if (socketPair_)
260 197 : socketPair_->interrupt();
261 :
262 197 : rtcpCheckerThread_.join();
263 :
264 197 : receiveThread_.reset();
265 197 : sender_.reset();
266 197 : socketPair_.reset();
267 197 : audioInput_.reset();
268 849 : }
269 :
270 : void
271 207 : AudioRtpSession::setMuted(bool muted, Direction dir)
272 : {
273 207 : asio::post(*Manager::instance().ioContext(), [w = weak_from_this(), muted, dir]() {
274 207 : if (auto shared = w.lock()) {
275 206 : std::lock_guard lock(shared->mutex_);
276 206 : if (dir == Direction::SEND) {
277 201 : shared->muteState_ = muted;
278 201 : if (shared->audioInput_) {
279 198 : shared->audioInput_->setMuted(muted);
280 : }
281 : } else {
282 5 : if (shared->receiveThread_) {
283 2 : auto ms = shared->receiveThread_->getInfo();
284 2 : ms.name = shared->streamId_ + ":remote";
285 2 : if (muted) {
286 2 : if (auto* ob = shared->recorder_->getStream(ms.name)) {
287 0 : shared->receiveThread_->detach(ob);
288 0 : shared->recorder_->removeStream(ms);
289 : }
290 : } else {
291 0 : if (auto* ob = shared->recorder_->addStream(ms)) {
292 0 : shared->receiveThread_->attach(ob);
293 : }
294 : }
295 2 : }
296 : }
297 413 : }
298 207 : });
299 207 : }
300 :
301 : void
302 197 : AudioRtpSession::setVoiceCallback(std::function<void(bool)> cb)
303 : {
304 197 : std::lock_guard lock(mutex_);
305 197 : voiceCallback_ = std::move(cb);
306 197 : if (sender_) {
307 27 : sender_->setVoiceCallback(voiceCallback_);
308 : }
309 197 : }
310 :
311 : bool
312 338 : AudioRtpSession::check_RCTP_Info_RR(RTCPInfo& rtcpi)
313 : {
314 338 : auto rtcpInfoVect = socketPair_->getRtcpRR();
315 338 : unsigned totalLost = 0;
316 338 : unsigned totalJitter = 0;
317 338 : unsigned nbDropNotNull = 0;
318 338 : auto vectSize = rtcpInfoVect.size();
319 :
320 338 : if (vectSize != 0) {
321 0 : for (const auto& it : rtcpInfoVect) {
322 0 : if (it.fraction_lost != 0) // Exclude null drop
323 0 : nbDropNotNull++;
324 0 : totalLost += it.fraction_lost;
325 0 : totalJitter += ntohl(it.jitter);
326 : }
327 0 : rtcpi.packetLoss = nbDropNotNull ? static_cast<float>((100 * totalLost) / (256.0 * nbDropNotNull)) : 0;
328 : // Jitter is expressed in timestamp unit -> convert to milliseconds
329 : // https://stackoverflow.com/questions/51956520/convert-jitter-from-rtp-timestamp-unit-to-millisseconds
330 0 : rtcpi.jitter = static_cast<unsigned int>(
331 0 : (static_cast<float>(totalJitter) / static_cast<float>(vectSize) / 90000.0f) * 1000.0f);
332 0 : rtcpi.nb_sample = vectSize;
333 0 : rtcpi.latency = static_cast<float>(socketPair_->getLastLatency());
334 0 : return true;
335 : }
336 338 : return false;
337 338 : }
338 :
339 : void
340 338 : AudioRtpSession::adaptQualityAndBitrate()
341 : {
342 338 : RTCPInfo rtcpi {};
343 338 : if (check_RCTP_Info_RR(rtcpi)) {
344 0 : dropProcessing(&rtcpi);
345 : }
346 338 : }
347 :
348 : void
349 0 : AudioRtpSession::dropProcessing(RTCPInfo* rtcpi)
350 : {
351 0 : auto pondLoss = getPonderateLoss(rtcpi->packetLoss);
352 0 : setNewPacketLoss(static_cast<unsigned int>(pondLoss));
353 0 : }
354 :
355 : void
356 0 : AudioRtpSession::setNewPacketLoss(unsigned int newPL)
357 : {
358 0 : newPL = std::clamp((int) newPL, 0, 100);
359 0 : if (newPL != packetLoss_) {
360 0 : if (sender_) {
361 0 : auto ret = sender_->setPacketLoss(newPL);
362 0 : packetLoss_ = newPL;
363 0 : if (ret == -1)
364 0 : JAMI_ERROR("Fail to access the encoder");
365 : } else {
366 0 : JAMI_ERROR("Fail to access the sender");
367 : }
368 : }
369 0 : }
370 :
371 : float
372 0 : AudioRtpSession::getPonderateLoss(float lastLoss)
373 : {
374 : static float pond = 10.0f;
375 :
376 0 : pond = floor(0.5 * lastLoss + 0.5 * pond);
377 0 : if (lastLoss > pond) {
378 0 : return lastLoss;
379 : } else {
380 0 : return pond;
381 : }
382 : }
383 :
384 : void
385 338 : AudioRtpSession::processRtcpChecker()
386 : {
387 338 : adaptQualityAndBitrate();
388 338 : socketPair_->waitForRTCP(std::chrono::seconds(rtcp_checking_interval));
389 338 : }
390 :
391 : void
392 0 : AudioRtpSession::attachRemoteRecorder(const MediaStream& ms)
393 : {
394 0 : std::lock_guard lock(mutex_);
395 0 : if (!recorder_ || !receiveThread_)
396 0 : return;
397 0 : MediaStream remoteMS = ms;
398 0 : remoteMS.name = streamId_ + ":remote";
399 0 : if (auto* ob = recorder_->addStream(remoteMS)) {
400 0 : receiveThread_->attach(ob);
401 : }
402 0 : }
403 :
404 : void
405 0 : AudioRtpSession::attachLocalRecorder(const MediaStream& ms)
406 : {
407 0 : std::lock_guard lock(mutex_);
408 0 : if (!recorder_ || !audioInput_)
409 0 : return;
410 0 : MediaStream localMS = ms;
411 0 : localMS.name = streamId_ + ":local";
412 0 : if (auto* ob = recorder_->addStream(localMS)) {
413 0 : audioInput_->attach(ob);
414 : }
415 0 : }
416 :
417 : void
418 9 : AudioRtpSession::initRecorder()
419 : {
420 9 : if (!recorder_)
421 0 : return;
422 9 : if (receiveThread_)
423 3 : receiveThread_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
424 0 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
425 0 : if (auto shared = w.lock())
426 0 : shared->attachRemoteRecorder(ms);
427 0 : });
428 0 : });
429 9 : if (audioInput_)
430 9 : audioInput_->setRecorderCallback([w = weak_from_this()](const MediaStream& ms) {
431 0 : asio::post(*Manager::instance().ioContext(), [w = std::move(w), ms]() {
432 0 : if (auto shared = w.lock())
433 0 : shared->attachLocalRecorder(ms);
434 0 : });
435 0 : });
436 : }
437 :
438 : void
439 401 : AudioRtpSession::deinitRecorder()
440 : {
441 401 : if (!recorder_)
442 0 : return;
443 401 : if (receiveThread_) {
444 1 : auto ms = receiveThread_->getInfo();
445 1 : ms.name = streamId_ + ":remote";
446 1 : if (auto* ob = recorder_->getStream(ms.name)) {
447 0 : receiveThread_->detach(ob);
448 0 : recorder_->removeStream(ms);
449 : }
450 1 : }
451 401 : if (audioInput_) {
452 3 : auto ms = audioInput_->getInfo();
453 3 : ms.name = streamId_ + ":local";
454 3 : if (auto* ob = recorder_->getStream(ms.name)) {
455 0 : audioInput_->detach(ob);
456 0 : recorder_->removeStream(ms);
457 : }
458 3 : }
459 : }
460 :
461 : } // namespace jami
|
