1 /*
2 * Copyright (c) 2023 Shenzhen Kaihong Digital Industry Development Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "rtp_queue.h"
17 #include <arpa/inet.h>
18 #include <iostream>
19 #include <limits>
20 #include <securec.h>
21 #include "common/common_macro.h"
22 #include "common/media_log.h"
23
24 namespace OHOS {
25 namespace Sharing {
RtpPacketSortor(int32_t sampleRate,size_t kMax,size_t kMin)26 RtpPacketSortor::RtpPacketSortor(int32_t sampleRate, size_t kMax, size_t kMin)
27 : sampleRate_(sampleRate), kMin_(kMin), kMax_(kMax)
28 {}
29
InputRtp(TrackType type,uint8_t * ptr,size_t len)30 void RtpPacketSortor::InputRtp(TrackType type, uint8_t *ptr, size_t len)
31 {
32 RETURN_IF_NULL(ptr);
33 if (len < RtpPacket::RTP_HEADER_SIZE) {
34 // hilog rtp packet is too small
35 return;
36 }
37 // todo jduge rtp size
38 if (!sampleRate_) {
39 return;
40 }
41 RtpHeader *header = (RtpHeader *)ptr;
42 if (header->version_ != RtpPacket::RTP_VERSION) {
43 // hilog error version error
44 return;
45 }
46 if (!header->GetPayloadSize(len)) {
47 return;
48 }
49
50 auto rtp = std::make_shared<RtpPacket>();
51 rtp->SetCapacity(len);
52 rtp->SetSize(len);
53 rtp->sampleRate_ = sampleRate_;
54 rtp->type_ = type;
55
56 auto data = rtp->Data();
57 auto ret = memcpy_s(data, len, ptr, len);
58 if (ret != EOK) {
59 return;
60 }
61 MEDIA_LOGD("rtp payload size: %{public}zu.", rtp->GetPayloadSize());
62
63 if (ssrc_ != rtp->GetSSRC() || rtp->GetSeq() == 0) {
64 ssrc_ = rtp->GetSSRC();
65 Flush();
66 Clear();
67 nextSeqOut_ = rtp->GetSeq();
68 }
69
70 SortPacket(rtp->GetSeq(), rtp);
71 return;
72 }
73
SetOnSort(const OnSort & cb)74 void RtpPacketSortor::SetOnSort(const OnSort &cb)
75 {
76 onSort_ = std::move(cb);
77 }
78
Clear()79 void RtpPacketSortor::Clear()
80 {
81 seqCycleCount_ = 0;
82 pktSortCacheMap_.clear();
83 nextSeqOut_ = 0;
84 maxSortSize_ = kMin_;
85 }
86
GetJitterSize() const87 size_t RtpPacketSortor::GetJitterSize() const
88 {
89 return pktSortCacheMap_.size();
90 }
91
GetCycleCount() const92 size_t RtpPacketSortor::GetCycleCount() const
93 {
94 return seqCycleCount_;
95 }
96
SortPacket(uint16_t seq,RtpPacket::Ptr packet)97 void RtpPacketSortor::SortPacket(uint16_t seq, RtpPacket::Ptr packet)
98 {
99 RETURN_IF_NULL(packet);
100 if (seq < nextSeqOut_) {
101 if (nextSeqOut_ < seq + kMax_) {
102 return;
103 }
104 } else if (nextSeqOut_ && seq - nextSeqOut_ > ((std::numeric_limits<uint16_t>::max)() >> 1)) {
105 MEDIA_LOGD("nextSeqOut_ && seq - nextSeqOut_ > ((std::numeric_limits<uint16_t>::max)() >> 1)");
106 return;
107 }
108
109 pktSortCacheMap_.emplace(seq, std::move(packet));
110
111 TryPopPacket();
112 }
113
Flush()114 void RtpPacketSortor::Flush()
115 {
116 while (!pktSortCacheMap_.empty()) {
117 PopIterator(pktSortCacheMap_.begin());
118 }
119 }
120
GetSSRC() const121 uint32_t RtpPacketSortor::GetSSRC() const
122 {
123 return ssrc_;
124 }
125
PopPacket()126 void RtpPacketSortor::PopPacket()
127 {
128 if (pktSortCacheMap_.empty()) {
129 return;
130 }
131
132 auto it = pktSortCacheMap_.begin();
133 if (it->first >= nextSeqOut_) {
134 PopIterator(it);
135 return;
136 }
137
138 if (nextSeqOut_ - it->first > (0xFFFF >> 1)) {
139 if (pktSortCacheMap_.size() < 2 * kMin_) { // 2:fixed size
140 return;
141 }
142 ++seqCycleCount_;
143
144 auto hit = pktSortCacheMap_.upper_bound((nextSeqOut_ - pktSortCacheMap_.size()));
145 while (hit != pktSortCacheMap_.end()) {
146 onSort_(hit->first, hit->second);
147 hit = pktSortCacheMap_.erase(hit);
148 }
149
150 PopIterator(pktSortCacheMap_.begin());
151 return;
152 }
153
154 pktSortCacheMap_.erase(it);
155 }
156
PopIterator(std::map<uint16_t,RtpPacket::Ptr>::iterator it)157 void RtpPacketSortor::PopIterator(std::map<uint16_t, RtpPacket::Ptr>::iterator it)
158 {
159 auto seq = it->first;
160 auto data = std::move(it->second);
161 pktSortCacheMap_.erase(it);
162 nextSeqOut_ = seq + 1;
163 onSort_(seq, data);
164 }
165
TryPopPacket()166 void RtpPacketSortor::TryPopPacket()
167 {
168 int32_t count = 0;
169 while ((!pktSortCacheMap_.empty() && pktSortCacheMap_.begin()->first == nextSeqOut_)) {
170 PopPacket();
171 ++count;
172 }
173
174 if (count) {
175 SetSortSize();
176 } else if (pktSortCacheMap_.size() > maxSortSize_) {
177 PopPacket();
178 SetSortSize();
179 }
180 }
181
SetSortSize()182 void RtpPacketSortor::SetSortSize()
183 {
184 maxSortSize_ = kMin_ + pktSortCacheMap_.size();
185 if (maxSortSize_ > kMax_) {
186 maxSortSize_ = kMax_;
187 }
188 }
189 } // namespace Sharing
190 } // namespace OHOS