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->ReplaceData(reinterpret_cast<char*>(ptr), len);
52     rtp->sampleRate_ = sampleRate_;
53     rtp->type_ = type;
54 
55     if (rtp->Size() != len) {
56         return;
57     }
58     MEDIA_LOGD("rtp payload size: %{public}zu.", rtp->GetPayloadSize());
59 
60     if (ssrc_ != rtp->GetSSRC() || rtp->GetSeq() == 0) {
61         ssrc_ = rtp->GetSSRC();
62         Flush();
63         Clear();
64         nextSeqOut_ = rtp->GetSeq();
65     }
66 
67     SortPacket(rtp->GetSeq(), rtp);
68     return;
69 }
70 
SetOnSort(const OnSort &cb)71 void RtpPacketSortor::SetOnSort(const OnSort &cb)
72 {
73     onSort_ = std::move(cb);
74 }
75 
Clear()76 void RtpPacketSortor::Clear()
77 {
78     seqCycleCount_ = 0;
79     pktSortCacheMap_.clear();
80     nextSeqOut_ = 0;
81     maxSortSize_ = kMin_;
82 }
83 
GetJitterSize() const84 size_t RtpPacketSortor::GetJitterSize() const
85 {
86     return pktSortCacheMap_.size();
87 }
88 
GetCycleCount() const89 size_t RtpPacketSortor::GetCycleCount() const
90 {
91     return seqCycleCount_;
92 }
93 
SortPacket(uint16_t seq, RtpPacket::Ptr packet)94 void RtpPacketSortor::SortPacket(uint16_t seq, RtpPacket::Ptr packet)
95 {
96     RETURN_IF_NULL(packet);
97     if (seq < nextSeqOut_) {
98         if (nextSeqOut_ < seq + kMax_) {
99             return;
100         }
101     } else if (nextSeqOut_ && seq - nextSeqOut_ > ((std::numeric_limits<uint16_t>::max)() >> 1)) {
102         MEDIA_LOGD("nextSeqOut_ && seq - nextSeqOut_ > ((std::numeric_limits<uint16_t>::max)() >> 1)");
103         return;
104     }
105 
106     pktSortCacheMap_.emplace(seq, std::move(packet));
107 
108     TryPopPacket();
109 }
110 
Flush()111 void RtpPacketSortor::Flush()
112 {
113     while (!pktSortCacheMap_.empty()) {
114         PopIterator(pktSortCacheMap_.begin());
115     }
116 }
117 
GetSSRC() const118 uint32_t RtpPacketSortor::GetSSRC() const
119 {
120     return ssrc_;
121 }
122 
PopPacket()123 void RtpPacketSortor::PopPacket()
124 {
125     if (pktSortCacheMap_.empty()) {
126         return;
127     }
128 
129     auto it = pktSortCacheMap_.begin();
130     if (it->first >= nextSeqOut_) {
131         PopIterator(it);
132         return;
133     }
134 
135     if (nextSeqOut_ - it->first > (0xFFFF >> 1)) {
136         if (pktSortCacheMap_.size() < 2 * kMin_) { // 2:fixed size
137             return;
138         }
139         ++seqCycleCount_;
140 
141         auto hit = pktSortCacheMap_.upper_bound((nextSeqOut_ - pktSortCacheMap_.size()));
142         while (hit != pktSortCacheMap_.end()) {
143             onSort_(hit->first, hit->second);
144             hit = pktSortCacheMap_.erase(hit);
145         }
146 
147         PopIterator(pktSortCacheMap_.begin());
148         return;
149     }
150 
151     pktSortCacheMap_.erase(it);
152 }
153 
PopIterator(std::map<uint16_t, RtpPacket::Ptr>::iterator it)154 void RtpPacketSortor::PopIterator(std::map<uint16_t, RtpPacket::Ptr>::iterator it)
155 {
156     auto seq = it->first;
157     auto data = std::move(it->second);
158     pktSortCacheMap_.erase(it);
159     nextSeqOut_ = seq + 1;
160     onSort_(seq, data);
161 }
162 
TryPopPacket()163 void RtpPacketSortor::TryPopPacket()
164 {
165     int32_t count = 0;
166     while ((!pktSortCacheMap_.empty() && pktSortCacheMap_.begin()->first == nextSeqOut_)) {
167         PopPacket();
168         ++count;
169     }
170 
171     if (count) {
172         SetSortSize();
173     } else if (pktSortCacheMap_.size() > maxSortSize_) {
174         PopPacket();
175         SetSortSize();
176     }
177 }
178 
SetSortSize()179 void RtpPacketSortor::SetSortSize()
180 {
181     maxSortSize_ = kMin_ + pktSortCacheMap_.size();
182     if (maxSortSize_ > kMax_) {
183         maxSortSize_ = kMax_;
184     }
185 }
186 } // namespace Sharing
187 } // namespace OHOS