Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : * Copyright (c) 2007 The FFmpeg Project
4 : *
5 : * This program is free software: you can redistribute it and/or modify
6 : * it under the terms of the GNU General Public License as published by
7 : * the Free Software Foundation, either version 3 of the License, or
8 : * (at your option) any later version.
9 : *
10 : * This program is distributed in the hope that it will be useful,
11 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 : * GNU General Public License for more details.
14 : *
15 : * You should have received a copy of the GNU General Public License
16 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
17 : */
18 :
19 : #include <dhtnet/ip_utils.h> // MUST BE INCLUDED FIRST
20 :
21 : #include "libav_deps.h" // THEN THIS ONE AFTER
22 :
23 : #include "socket_pair.h"
24 : #include "logger.h"
25 : #include "connectivity/security/memory.h"
26 :
27 : #include <dhtnet/ice_socket.h>
28 :
29 : #include <string>
30 : #include <algorithm>
31 :
32 : extern "C" {
33 : #include "srtp.h"
34 : }
35 :
36 : #include <cstring>
37 : #include <stdexcept>
38 : #include <unistd.h>
39 : #include <sys/types.h>
40 :
41 : #ifdef _WIN32
42 : #define SOCK_NONBLOCK FIONBIO
43 : #define poll WSAPoll
44 : #define close(x) closesocket(x)
45 : #endif
46 :
47 : #ifdef __ANDROID__
48 : #include <asm-generic/fcntl.h>
49 : #define SOCK_NONBLOCK O_NONBLOCK
50 : #endif
51 :
52 : #ifdef __APPLE__
53 : #include <fcntl.h>
54 : #endif
55 :
56 : // Swap 2 byte, 16 bit values:
57 : #define Swap2Bytes(val) ((((val) >> 8) & 0x00FF) | (((val) << 8) & 0xFF00))
58 :
59 : // Swap 4 byte, 32 bit values:
60 : #define Swap4Bytes(val) \
61 : ((((val) >> 24) & 0x000000FF) | (((val) >> 8) & 0x0000FF00) | (((val) << 8) & 0x00FF0000) \
62 : | (((val) << 24) & 0xFF000000))
63 :
64 : // Swap 8 byte, 64 bit values:
65 : #define Swap8Bytes(val) \
66 : ((((val) >> 56) & 0x00000000000000FF) | (((val) >> 40) & 0x000000000000FF00) \
67 : | (((val) >> 24) & 0x0000000000FF0000) | (((val) >> 8) & 0x00000000FF000000) \
68 : | (((val) << 8) & 0x000000FF00000000) | (((val) << 24) & 0x0000FF0000000000) \
69 : | (((val) << 40) & 0x00FF000000000000) | (((val) << 56) & 0xFF00000000000000))
70 :
71 : namespace jami {
72 :
73 : static constexpr int NET_POLL_TIMEOUT = 100; /* poll() timeout in ms */
74 : static constexpr int RTP_MAX_PACKET_LENGTH = 2048;
75 : static constexpr unsigned MAX_REJECTED_SRTP_PACKETS = 32;
76 : static constexpr auto UDP_HEADER_SIZE = 8;
77 : static constexpr auto SRTP_OVERHEAD = 10;
78 : static constexpr uint32_t RTCP_RR_FRACTION_MASK = 0xFF000000;
79 : static constexpr unsigned MINIMUM_RTP_HEADER_SIZE = 16;
80 :
81 : enum class DataType : uint8_t { RTP = 1 << 0, RTCP = 1 << 1 };
82 :
83 : class SRTPProtoContext
84 : {
85 : public:
86 400 : SRTPProtoContext(const char* out_suite, const char* out_key, const char* in_suite, const char* in_key)
87 400 : {
88 400 : jami_secure_memzero(&srtp_out, sizeof(srtp_out));
89 400 : jami_secure_memzero(&srtp_in, sizeof(srtp_in));
90 400 : if (out_suite && out_key) {
91 : // XXX: see srtp_open from libavformat/srtpproto.c
92 400 : if (ff_srtp_set_crypto(&srtp_out, out_suite, out_key) < 0) {
93 0 : srtp_close();
94 0 : throw std::runtime_error("Unable to set crypto on output");
95 : }
96 : }
97 :
98 400 : if (in_suite && in_key) {
99 400 : if (ff_srtp_set_crypto(&srtp_in, in_suite, in_key) < 0) {
100 0 : srtp_close();
101 0 : throw std::runtime_error("Unable to set crypto on input");
102 : }
103 : }
104 400 : }
105 :
106 400 : ~SRTPProtoContext() { srtp_close(); }
107 :
108 : SRTPContext srtp_out {};
109 : SRTPContext srtp_in {};
110 : uint8_t encryptbuf[RTP_MAX_PACKET_LENGTH];
111 :
112 : private:
113 400 : void srtp_close() noexcept
114 : {
115 400 : ff_srtp_free(&srtp_out);
116 400 : ff_srtp_free(&srtp_in);
117 400 : }
118 : };
119 :
120 : static int
121 0 : ff_network_wait_fd(int fd)
122 : {
123 0 : struct pollfd p = {fd, POLLOUT, 0};
124 0 : auto ret = poll(&p, 1, NET_POLL_TIMEOUT);
125 0 : return ret < 0 ? errno : p.revents & (POLLOUT | POLLERR | POLLHUP) ? 0 : -EAGAIN;
126 : }
127 :
128 : static int
129 12 : udp_socket_create(int family, int port)
130 : {
131 12 : int udp_fd = -1;
132 :
133 : #ifdef __APPLE__
134 : udp_fd = socket(family, SOCK_DGRAM, 0);
135 : if (udp_fd >= 0 && fcntl(udp_fd, F_SETFL, O_NONBLOCK) < 0) {
136 : close(udp_fd);
137 : udp_fd = -1;
138 : }
139 : #elif defined _WIN32
140 : udp_fd = socket(family, SOCK_DGRAM, 0);
141 : u_long block = 1;
142 : if (udp_fd >= 0 && ioctlsocket(udp_fd, FIONBIO, &block) < 0) {
143 : close(udp_fd);
144 : udp_fd = -1;
145 : }
146 : #else
147 12 : udp_fd = socket(family, SOCK_DGRAM | SOCK_NONBLOCK, 0);
148 : #endif
149 :
150 12 : if (udp_fd < 0) {
151 0 : JAMI_ERROR("socket() failed");
152 0 : strErr();
153 0 : return -1;
154 : }
155 :
156 12 : auto bind_addr = dhtnet::ip_utils::getAnyHostAddr(family);
157 12 : if (not bind_addr.isIpv4() and not bind_addr.isIpv6()) {
158 0 : JAMI_ERROR("No IPv4/IPv6 host found for family {}", family);
159 0 : close(udp_fd);
160 0 : return -1;
161 : }
162 :
163 12 : bind_addr.setPort(port);
164 12 : JAMI_LOG("use local address: {}", bind_addr.toString(true, true));
165 12 : if (::bind(udp_fd, bind_addr, bind_addr.getLength()) < 0) {
166 0 : JAMI_ERROR("bind() failed");
167 0 : strErr();
168 0 : close(udp_fd);
169 0 : udp_fd = -1;
170 : }
171 :
172 12 : return udp_fd;
173 : }
174 :
175 6 : SocketPair::SocketPair(const char* uri, int localPort)
176 : {
177 6 : openSockets(uri, localPort);
178 6 : }
179 :
180 394 : SocketPair::SocketPair(std::unique_ptr<dhtnet::IceSocket> rtp_sock, std::unique_ptr<dhtnet::IceSocket> rtcp_sock)
181 394 : : rtp_sock_(std::move(rtp_sock))
182 788 : , rtcp_sock_(std::move(rtcp_sock))
183 : {
184 394 : JAMI_LOG("[{}] Creating instance using ICE sockets for comp {} and {}",
185 : fmt::ptr(this),
186 : rtp_sock_->getCompId(),
187 : rtcp_sock_->getCompId());
188 :
189 394 : rtp_sock_->setOnRecv([this](uint8_t* buf, size_t len) {
190 8930 : std::lock_guard l(dataBuffMutex_);
191 8930 : rtpDataBuff_.emplace_back(buf, buf + len);
192 8930 : cv_.notify_one();
193 8930 : return len;
194 8930 : });
195 394 : rtcp_sock_->setOnRecv([this](uint8_t* buf, size_t len) {
196 395 : std::lock_guard l(dataBuffMutex_);
197 395 : rtcpDataBuff_.emplace_back(buf, buf + len);
198 395 : cv_.notify_one();
199 395 : return len;
200 395 : });
201 394 : }
202 :
203 400 : SocketPair::~SocketPair()
204 : {
205 400 : interrupt();
206 400 : closeSockets();
207 400 : JAMI_LOG("[{}] Instance destroyed", fmt::ptr(this));
208 400 : }
209 :
210 : bool
211 1019 : SocketPair::waitForRTCP(std::chrono::seconds interval)
212 : {
213 1019 : std::unique_lock lock(rtcpInfo_mutex_);
214 1019 : return cvRtcpPacketReadyToRead_.wait_for(lock, interval, [this] {
215 2935 : return interrupted_ or not listRtcpRRHeader_.empty() or not listRtcpREMBHeader_.empty();
216 2038 : });
217 1019 : }
218 :
219 : void
220 19 : SocketPair::saveRtcpRRPacket(uint8_t* buf, size_t len)
221 : {
222 19 : if (len < sizeof(rtcpRRHeader))
223 0 : return;
224 :
225 19 : auto* header = reinterpret_cast<rtcpRRHeader*>(buf);
226 19 : if (header->pt != 201) // 201 = RR PT
227 0 : return;
228 :
229 19 : std::lock_guard lock(rtcpInfo_mutex_);
230 :
231 19 : if (listRtcpRRHeader_.size() >= MAX_LIST_SIZE) {
232 0 : listRtcpRRHeader_.pop_front();
233 : }
234 :
235 19 : listRtcpRRHeader_.emplace_back(*header);
236 :
237 19 : cvRtcpPacketReadyToRead_.notify_one();
238 19 : }
239 :
240 : void
241 276 : SocketPair::saveRtcpREMBPacket(uint8_t* buf, size_t len)
242 : {
243 276 : if (len < sizeof(rtcpREMBHeader))
244 0 : return;
245 :
246 276 : auto* header = reinterpret_cast<rtcpREMBHeader*>(buf);
247 276 : if (header->pt != 206) // 206 = REMB PT
248 0 : return;
249 :
250 276 : if (header->uid != 0x424D4552) // uid must be "REMB"
251 0 : return;
252 :
253 276 : std::lock_guard lock(rtcpInfo_mutex_);
254 :
255 276 : if (listRtcpREMBHeader_.size() >= MAX_LIST_SIZE) {
256 0 : listRtcpREMBHeader_.pop_front();
257 : }
258 :
259 276 : listRtcpREMBHeader_.push_back(*header);
260 :
261 276 : cvRtcpPacketReadyToRead_.notify_one();
262 276 : }
263 :
264 : std::list<rtcpRRHeader>
265 1019 : SocketPair::getRtcpRR()
266 : {
267 1019 : std::lock_guard lock(rtcpInfo_mutex_);
268 2038 : return std::move(listRtcpRRHeader_);
269 1019 : }
270 :
271 : std::list<rtcpREMBHeader>
272 558 : SocketPair::getRtcpREMB()
273 : {
274 558 : std::lock_guard lock(rtcpInfo_mutex_);
275 1116 : return std::move(listRtcpREMBHeader_);
276 558 : }
277 :
278 : void
279 400 : SocketPair::createSRTP(const char* out_suite, const char* out_key, const char* in_suite, const char* in_key)
280 : {
281 400 : srtpContext_.reset(new SRTPProtoContext(out_suite, out_key, in_suite, in_key));
282 400 : }
283 :
284 : void
285 800 : SocketPair::interrupt()
286 : {
287 800 : JAMI_WARNING("[{}] Interrupting RTP sockets", fmt::ptr(this));
288 800 : interrupted_ = true;
289 800 : if (rtp_sock_)
290 788 : rtp_sock_->setOnRecv(nullptr);
291 800 : if (rtcp_sock_)
292 788 : rtcp_sock_->setOnRecv(nullptr);
293 800 : cv_.notify_all();
294 800 : cvRtcpPacketReadyToRead_.notify_all();
295 800 : }
296 :
297 : void
298 1157 : SocketPair::setReadBlockingMode(bool block)
299 : {
300 1157 : JAMI_LOG("[{}] Read operations in blocking mode [{}]", fmt::ptr(this), block ? "YES" : "NO");
301 1157 : readBlockingMode_ = block;
302 1157 : cv_.notify_all();
303 1157 : cvRtcpPacketReadyToRead_.notify_all();
304 1157 : }
305 :
306 : void
307 1156 : SocketPair::stopSendOp(bool state)
308 : {
309 1156 : noWrite_ = state;
310 1156 : }
311 :
312 : void
313 400 : SocketPair::closeSockets()
314 : {
315 400 : if (rtcpHandle_ > 0 and close(rtcpHandle_))
316 0 : strErr();
317 400 : if (rtpHandle_ > 0 and close(rtpHandle_))
318 0 : strErr();
319 400 : }
320 :
321 : void
322 6 : SocketPair::openSockets(const char* uri, int local_rtp_port)
323 : {
324 6 : JAMI_LOG("[{}] Creating rtp socket for uri {} on port {}", fmt::ptr(this), uri, local_rtp_port);
325 :
326 : char hostname[256];
327 : char path[1024];
328 : int dst_rtp_port;
329 :
330 6 : av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &dst_rtp_port, path, sizeof(path), uri);
331 :
332 6 : const int local_rtcp_port = local_rtp_port + 1;
333 6 : const int dst_rtcp_port = dst_rtp_port + 1;
334 :
335 6 : rtpDestAddr_ = dhtnet::IpAddr {hostname};
336 6 : rtpDestAddr_.setPort(dst_rtp_port);
337 6 : rtcpDestAddr_ = dhtnet::IpAddr {hostname};
338 6 : rtcpDestAddr_.setPort(dst_rtcp_port);
339 :
340 : // Open local sockets (RTP/RTCP)
341 6 : if ((rtpHandle_ = udp_socket_create(rtpDestAddr_.getFamily(), local_rtp_port)) == -1
342 6 : or (rtcpHandle_ = udp_socket_create(rtcpDestAddr_.getFamily(), local_rtcp_port)) == -1) {
343 0 : closeSockets();
344 0 : JAMI_ERROR("[{}] Sockets creation failed", fmt::ptr(this));
345 0 : throw std::runtime_error("Sockets creation failed");
346 : }
347 :
348 6 : JAMI_WARNING("SocketPair: local({},{}) / {}({},{})",
349 : local_rtp_port,
350 : local_rtcp_port,
351 : hostname,
352 : dst_rtp_port,
353 : dst_rtcp_port);
354 6 : }
355 :
356 : MediaIOHandle*
357 789 : SocketPair::createIOContext(const uint16_t mtu)
358 : {
359 : unsigned ip_header_size;
360 789 : if (rtp_sock_)
361 777 : ip_header_size = rtp_sock_->getTransportOverhead();
362 12 : else if (rtpDestAddr_.getFamily() == AF_INET6)
363 0 : ip_header_size = 40;
364 : else
365 12 : ip_header_size = 20;
366 : return new MediaIOHandle(
367 789 : mtu - (srtpContext_ ? SRTP_OVERHEAD : 0) - UDP_HEADER_SIZE - ip_header_size,
368 : true,
369 9876 : [](void* sp, uint8_t* buf, int len) { return static_cast<SocketPair*>(sp)->readCallback(buf, len); },
370 9869 : [](void* sp, io_writebuffer* buf, int len) { return static_cast<SocketPair*>(sp)->writeCallback(buf, len); },
371 : 0,
372 2367 : reinterpret_cast<void*>(this));
373 : }
374 :
375 : int
376 9876 : SocketPair::waitForData()
377 : {
378 : // System sockets
379 9876 : if (rtpHandle_ >= 0) {
380 : int ret;
381 : do {
382 88 : if (interrupted_) {
383 4 : errno = EINTR;
384 6 : return -1;
385 : }
386 :
387 84 : if (not readBlockingMode_) {
388 2 : return 0;
389 : }
390 :
391 : // work with system socket
392 82 : struct pollfd p[2] = {{rtpHandle_, POLLIN, 0}, {rtcpHandle_, POLLIN, 0}};
393 82 : ret = poll(p, 2, NET_POLL_TIMEOUT);
394 82 : if (ret > 0) {
395 0 : ret = 0;
396 0 : if (p[0].revents & POLLIN)
397 0 : ret |= static_cast<int>(DataType::RTP);
398 0 : if (p[1].revents & POLLIN)
399 0 : ret |= static_cast<int>(DataType::RTCP);
400 : }
401 82 : } while (!ret or (ret < 0 and errno == EAGAIN));
402 :
403 0 : return ret;
404 : }
405 :
406 : // work with IceSocket
407 : {
408 9870 : std::unique_lock lk(dataBuffMutex_);
409 9870 : cv_.wait(lk, [this] {
410 19567 : return interrupted_ or not rtpDataBuff_.empty() or not rtcpDataBuff_.empty() or not readBlockingMode_;
411 : });
412 9870 : }
413 :
414 9870 : if (interrupted_) {
415 13 : errno = EINTR;
416 13 : return -1;
417 : }
418 :
419 9857 : return static_cast<int>(DataType::RTP) | static_cast<int>(DataType::RTCP);
420 : }
421 :
422 : int
423 9462 : SocketPair::readRtpData(void* buf, int buf_size)
424 : {
425 : // handle system socket
426 9462 : if (rtpHandle_ >= 0) {
427 : struct sockaddr_storage from;
428 0 : socklen_t from_len = sizeof(from);
429 0 : return static_cast<int>(recvfrom(rtpHandle_,
430 : static_cast<char*>(buf),
431 : buf_size,
432 : 0,
433 : reinterpret_cast<struct sockaddr*>(&from),
434 0 : &from_len));
435 : }
436 :
437 : // handle ICE
438 9462 : std::unique_lock lk(dataBuffMutex_);
439 9462 : if (not rtpDataBuff_.empty()) {
440 8929 : auto pkt = std::move(rtpDataBuff_.front());
441 8929 : rtpDataBuff_.pop_front();
442 8929 : lk.unlock(); // to not block our ICE callbacks
443 8929 : int pkt_size = static_cast<int>(pkt.size());
444 8929 : int len = std::min(pkt_size, buf_size);
445 8929 : std::copy_n(pkt.begin(), len, static_cast<char*>(buf));
446 8929 : return len;
447 8929 : }
448 :
449 533 : return 0;
450 9462 : }
451 :
452 : int
453 9857 : SocketPair::readRtcpData(void* buf, int buf_size)
454 : {
455 : // handle system socket
456 9857 : if (rtcpHandle_ >= 0) {
457 : struct sockaddr_storage from;
458 0 : socklen_t from_len = sizeof(from);
459 0 : return static_cast<int>(recvfrom(rtcpHandle_,
460 : static_cast<char*>(buf),
461 : buf_size,
462 : 0,
463 : reinterpret_cast<struct sockaddr*>(&from),
464 0 : &from_len));
465 : }
466 :
467 : // handle ICE
468 9857 : std::unique_lock lk(dataBuffMutex_);
469 9857 : if (not rtcpDataBuff_.empty()) {
470 395 : auto pkt = std::move(rtcpDataBuff_.front());
471 395 : rtcpDataBuff_.pop_front();
472 395 : lk.unlock();
473 395 : int pkt_size = static_cast<int>(pkt.size());
474 395 : int len = std::min(pkt_size, buf_size);
475 395 : std::copy_n(pkt.begin(), len, static_cast<char*>(buf));
476 395 : return len;
477 395 : }
478 :
479 9462 : return 0;
480 9857 : }
481 :
482 : int
483 9876 : SocketPair::readCallback(uint8_t* buf, int buf_size)
484 : {
485 9876 : for (unsigned rejected = 0; rejected < MAX_REJECTED_SRTP_PACKETS; ++rejected) {
486 9876 : auto datatype = waitForData();
487 9876 : if (datatype < 0)
488 9876 : return datatype;
489 :
490 9859 : int len = 0;
491 9859 : bool fromRTCP = false;
492 :
493 9859 : if (datatype & static_cast<int>(DataType::RTCP)) {
494 9857 : len = readRtcpData(buf, buf_size);
495 9857 : if (len > 0) {
496 395 : auto* header = reinterpret_cast<rtcpRRHeader*>(buf);
497 : // 201 = RR PT
498 395 : if (header->pt == 201) {
499 19 : lastDLSR_ = Swap4Bytes(header->dlsr);
500 : // JAMI_WARN("Read RR, lastDLSR : %d", lastDLSR_);
501 19 : lastRR_time = std::chrono::steady_clock::now();
502 19 : saveRtcpRRPacket(buf, len);
503 : }
504 : // 206 = REMB PT
505 376 : else if (header->pt == 206)
506 276 : saveRtcpREMBPacket(buf, len);
507 : // 200 = SR PT
508 100 : else if (header->pt == 200) {
509 : // not used yet
510 : } else {
511 0 : unsigned pt = header->pt;
512 0 : JAMI_LOG("Unable to read RTCP: unknown packet type {}", pt);
513 : }
514 395 : fromRTCP = true;
515 : }
516 : }
517 :
518 : // No RTCP… attempt RTP
519 9859 : if (!len and (datatype & static_cast<int>(DataType::RTP))) {
520 9462 : len = readRtpData(buf, buf_size);
521 9462 : fromRTCP = false;
522 : }
523 :
524 9859 : if (len <= 0)
525 535 : return len;
526 :
527 9324 : if (not fromRTCP && (buf_size < static_cast<int>(MINIMUM_RTP_HEADER_SIZE)))
528 0 : return len;
529 :
530 : // SRTP decrypt
531 9324 : if (not fromRTCP and srtpContext_ and srtpContext_->srtp_in.aes) {
532 8929 : auto err = ff_srtp_decrypt(&srtpContext_->srtp_in, buf, &len);
533 8929 : if (err < 0) {
534 0 : if (rejected == 0)
535 0 : JAMI_WARNING("decrypt error {}", err);
536 0 : continue;
537 : }
538 :
539 8929 : int32_t gradient = 0;
540 8929 : int32_t deltaT = 0;
541 8929 : float abs = 0.0f;
542 8929 : bool res_parse = parse_RTP_ext(buf, &abs);
543 8929 : bool marker = (buf[1] & 0x80) >> 7;
544 8929 : bool res_delay = res_parse and getOneWayDelayGradient(abs, marker, &gradient, &deltaT);
545 :
546 : // rtpDelayCallback_ is not set for audio
547 8929 : if (rtpDelayCallback_ and res_delay)
548 8815 : rtpDelayCallback_(gradient, deltaT);
549 :
550 8929 : if (packetLossCallback_ and (buf[2] << 8 | buf[3]) != lastSeqNumIn_ + 1)
551 0 : packetLossCallback_();
552 8929 : lastSeqNumIn_ = buf[2] << 8 | buf[3];
553 : }
554 :
555 9324 : if (len != 0)
556 9324 : return len;
557 : else
558 0 : return AVERROR_EOF;
559 : }
560 :
561 : // Let the demuxer regain control under sustained invalid traffic.
562 0 : return AVERROR(EAGAIN);
563 : }
564 :
565 : int
566 9356 : SocketPair::writeData(const uint8_t* buf, int buf_size)
567 : {
568 9356 : bool isRTCP = RTP_PT_IS_RTCP(buf[1]);
569 :
570 : // System sockets?
571 9356 : if (rtpHandle_ >= 0) {
572 : int fd;
573 : dhtnet::IpAddr* dest_addr;
574 :
575 0 : if (isRTCP) {
576 0 : fd = rtcpHandle_;
577 0 : dest_addr = &rtcpDestAddr_;
578 : } else {
579 0 : fd = rtpHandle_;
580 0 : dest_addr = &rtpDestAddr_;
581 : }
582 :
583 0 : auto ret = ff_network_wait_fd(fd);
584 0 : if (ret < 0)
585 0 : return ret;
586 :
587 0 : if (noWrite_)
588 0 : return buf_size;
589 : return static_cast<int>(
590 0 : ::sendto(fd, reinterpret_cast<const char*>(buf), buf_size, 0, *dest_addr, dest_addr->getLength()));
591 : }
592 :
593 9356 : if (noWrite_)
594 0 : return buf_size;
595 :
596 : // IceSocket
597 9356 : if (isRTCP)
598 397 : return static_cast<int>(rtcp_sock_->send(buf, buf_size));
599 : else
600 8959 : return static_cast<int>(rtp_sock_->send(buf, buf_size));
601 : }
602 :
603 : int
604 9080 : SocketPair::writeCallback(const uint8_t* buf, int buf_size)
605 : {
606 9080 : if (noWrite_)
607 0 : return 0;
608 :
609 : int ret;
610 9080 : bool isRTCP = RTP_PT_IS_RTCP(buf[1]);
611 : unsigned int ts_LSB, ts_MSB;
612 : double currentSRTS, currentLatency;
613 :
614 : // Encrypt?
615 9080 : if (not isRTCP and srtpContext_ and srtpContext_->srtp_out.aes) {
616 8959 : buf_size = ff_srtp_encrypt(&srtpContext_->srtp_out,
617 : buf,
618 : buf_size,
619 8959 : srtpContext_->encryptbuf,
620 : sizeof(srtpContext_->encryptbuf));
621 8959 : if (buf_size < 0) {
622 0 : JAMI_WARNING("encrypt error {}", buf_size);
623 0 : return buf_size;
624 : }
625 :
626 8959 : buf = srtpContext_->encryptbuf;
627 : }
628 :
629 : // check if we're sending an RR, if so, detect packet loss
630 : // buf_size gives length of buffer, not just header
631 9080 : if (isRTCP && static_cast<unsigned>(buf_size) >= sizeof(rtcpRRHeader)) {
632 19 : auto* header = reinterpret_cast<const rtcpRRHeader*>(buf);
633 19 : rtcpPacketLoss_ = (header->pt == 201 && ntohl(header->fraction_lost) & RTCP_RR_FRACTION_MASK);
634 : }
635 :
636 : do {
637 9080 : if (interrupted_)
638 0 : return -EINTR;
639 9080 : ret = writeData(buf, buf_size);
640 9080 : } while (ret < 0 and errno == EAGAIN);
641 :
642 9080 : if (buf[1] == 200) // Sender Report
643 : {
644 102 : auto* header = reinterpret_cast<const rtcpSRHeader*>(buf);
645 102 : ts_LSB = Swap4Bytes(header->timestampLSB);
646 102 : ts_MSB = Swap4Bytes(header->timestampMSB);
647 :
648 102 : currentSRTS = ts_MSB + (ts_LSB / pow(2, 32));
649 :
650 102 : if (lastSRTS_ != 0 && lastDLSR_ != 0) {
651 28 : if (histoLatency_.size() >= MAX_LIST_SIZE)
652 0 : histoLatency_.pop_front();
653 :
654 28 : currentLatency = (currentSRTS - lastSRTS_) / 2;
655 : // JAMI_WARN("Current Latency : %f from sender %X", currentLatency, header->ssrc);
656 28 : histoLatency_.push_back(currentLatency);
657 : }
658 :
659 102 : lastSRTS_ = currentSRTS;
660 :
661 : // JAMI_WARN("SENDING NEW RTCP SR !! ");
662 :
663 8978 : } else if (buf[1] == 201) // Receiver Report
664 : {
665 : // auto header = reinterpret_cast<rtcpRRHeader*>(buf);
666 : // JAMI_WARN("SENDING NEW RTCP RR !! ");
667 : }
668 :
669 9080 : return ret < 0 ? -errno : ret;
670 : }
671 :
672 : double
673 19 : SocketPair::getLastLatency()
674 : {
675 19 : if (not histoLatency_.empty())
676 6 : return histoLatency_.back();
677 : else
678 13 : return -1;
679 : }
680 :
681 : void
682 179 : SocketPair::setRtpDelayCallback(std::function<void(int, int)> cb)
683 : {
684 179 : rtpDelayCallback_ = std::move(cb);
685 179 : }
686 :
687 : bool
688 8929 : SocketPair::getOneWayDelayGradient(float sendTS, bool marker, int32_t* gradient, int32_t* deltaT)
689 : {
690 : // Keep only last packet of each frame
691 8929 : if (not marker) {
692 57 : return false;
693 : }
694 :
695 : // 1st frame
696 8872 : if (lastSendTS_ == 0.0f) {
697 57 : lastSendTS_ = sendTS;
698 57 : lastReceiveTS_ = std::chrono::steady_clock::now();
699 57 : return false;
700 : }
701 :
702 8815 : int32_t deltaS = static_cast<int32_t>((sendTS - lastSendTS_) * 1000); // milliseconds
703 8815 : if (deltaS < 0)
704 0 : deltaS += 64000;
705 8815 : lastSendTS_ = sendTS;
706 :
707 8815 : std::chrono::steady_clock::time_point arrival_TS = std::chrono::steady_clock::now();
708 : auto deltaR = static_cast<int32_t>(
709 8815 : std::chrono::duration_cast<std::chrono::milliseconds>(arrival_TS - lastReceiveTS_).count());
710 8815 : lastReceiveTS_ = arrival_TS;
711 :
712 8815 : *gradient = deltaR - deltaS;
713 8815 : *deltaT = deltaR;
714 :
715 8815 : return true;
716 : }
717 :
718 : bool
719 8929 : SocketPair::parse_RTP_ext(uint8_t* buf, float* abs)
720 : {
721 8929 : if (not(buf[0] & 0x10))
722 0 : return false;
723 :
724 8929 : uint16_t magic_word = (buf[12] << 8) + buf[13];
725 8929 : if (magic_word != 0xBEDE)
726 0 : return false;
727 :
728 8929 : uint8_t sec = buf[17] >> 2;
729 8929 : uint32_t fract = ((buf[17] & 0x3) << 16 | (buf[18] << 8) | buf[19]) << 14;
730 8929 : float milli = static_cast<float>(fract / pow(2, 32));
731 :
732 8929 : *abs = static_cast<float>(sec) + (milli);
733 8929 : return true;
734 : }
735 :
736 : uint16_t
737 178 : SocketPair::lastSeqValOut()
738 : {
739 178 : if (srtpContext_)
740 178 : return srtpContext_->srtp_out.seq_largest;
741 0 : JAMI_ERROR("SRTP context not found.");
742 0 : return 0;
743 : }
744 :
745 : } // namespace jami
|
