RTPUdpReceiver.cpp
9.18 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
#include "RTPUdpReceiver.h"
#include <iostream>
#include <time.h>
#include <thread>
#include <chrono>
#include "../logger.hpp"
using namespace std;
#define BUFFERSIZE_1024 4096
#define BUFFERSIZE_GAP 4096//5120 //1024*5
namespace
{
const int kVideoFrameSize = BUFFERSIZE_1024*BUFFERSIZE_1024*5*2;
const int kRtpRecvBufferSize = BUFFERSIZE_1024*BUFFERSIZE_1024*2;
const uint16_t kInvalidPort = 0;
}; // namespace
class UdpRTPSession : public RTPSession
{
public:
UdpRTPSession() {}
virtual ~UdpRTPSession() {}
private:
virtual void OnRTPPacket(RTPPacket* pack, const RTPTime& receiverTime, const RTPAddress* senderAddress)
{
AddDestination(*senderAddress);
}
virtual void OnRTCPCompoundPacket(RTCPCompoundPacket *pack, const RTPTime &receivetime,const RTPAddress *senderaddress)
{
//AddDestination(*senderaddress);
//const char* name = "hi~";
//SendRTCPAPPPacket(0, (const uint8_t*)name, "keeplive", 8);
//printf("send rtcp app");
}
};
static long long get_cur_time() {
chrono::time_point<chrono::system_clock, chrono::milliseconds> tpMicro
= chrono::time_point_cast<chrono::milliseconds>(chrono::system_clock::now());
return tpMicro.time_since_epoch().count();
}
static int rtp_revc_thread_(void* param)
{
if (!param)
{
return -1;
}
RTPUdpReceiver* self = (RTPUdpReceiver*)param;
return self->OnRtpRecv();
}
RTPUdpReceiver::RTPUdpReceiver()
: m_bRtpExit(false)
, m_bOpened(false)
, m_idleCount(-1)
,m_noDataCount(-1)
{
m_sessparamsPtr = new RTPSessionParams();
m_transparamsPtr = new RTPUDPv4TransmissionParams();
m_rtpSessionPtr = new UdpRTPSession();
}
RTPUdpReceiver::~RTPUdpReceiver()
{
if (IsOpened())
Close();
if(nullptr != m_sessparamsPtr){
delete m_sessparamsPtr;
m_sessparamsPtr = nullptr;
}
if(nullptr != m_transparamsPtr){
delete m_transparamsPtr;
m_transparamsPtr = nullptr;
}
if(nullptr != m_rtpSessionPtr){
delete m_rtpSessionPtr;
m_rtpSessionPtr = nullptr;
}
}
bool RTPUdpReceiver::Open(uint16_t localPort)
{
m_sessparamsPtr->SetUsePollThread(true);
m_sessparamsPtr->SetMinimumRTCPTransmissionInterval(10);
m_sessparamsPtr->SetOwnTimestampUnit(1.0/90000.0);
m_sessparamsPtr->SetAcceptOwnPackets(true);
m_transparamsPtr->SetPortbase(localPort);
m_transparamsPtr->SetRTPReceiveBuffer(kRtpRecvBufferSize);
LOG_INFO("[{}] port: {}", m_deviceID, localPort);
int err = m_rtpSessionPtr->Create(*m_sessparamsPtr, m_transparamsPtr);
if (err != 0)
{
LOG_ERROR("[{}] Create error: {}", m_deviceID, err);
return false;
}
m_rtpThreadPtr = new std::thread(rtp_revc_thread_, this);
if (nullptr == m_rtpThreadPtr)
{
LOG_ERROR("[{}] Create m_rtpThreadPtr error", m_deviceID);
return false;
}
if (InitPS() != 0)
{
return false;
}
m_bOpened = true;
LOG_INFO("[{}] Open ok", m_deviceID);
return true;
}
bool RTPUdpReceiver::IsOpened()
{
return m_bOpened;
}
void RTPUdpReceiver::Close()
{
m_bRtpExit = true;
// rtp接收线程退出
if (nullptr != m_rtpThreadPtr && m_rtpThreadPtr->joinable())
{
m_rtpThreadPtr->join();
delete m_rtpThreadPtr;
m_rtpThreadPtr = nullptr;
}
m_rtpSessionPtr->Destroy();
ClosePsThread();
m_bOpened = false;
LOG_INFO("[{}] closed.", m_deviceID);
}
// 收RTP包线程
int RTPUdpReceiver::OnRtpRecv()
{
if(nullptr == m_rtpSessionPtr){
return -1;
}
LOG_INFO("[{}] OnRtpRecv started.", m_deviceID);
while (!m_bRtpExit)
{
//try
//{
m_rtpSessionPtr->Poll();
m_rtpSessionPtr->BeginDataAccess();
if (m_rtpSessionPtr->GotoFirstSourceWithData())
{
LOG_INFO("OnRtpRecv GotoFirstSourceWithData --{}", m_deviceID);
last_recv_ts = get_cur_time();
m_idleCount = 0;
m_noDataCount = 0;
do
{
RTPPacket* packet;
while ((packet = m_rtpSessionPtr->GetNextPacket()) != NULL)
{
LOG_INFO("OnRtpRecv GetNextPacket --{}", m_deviceID);
int ret = ParsePacket(packet);
m_rtpSessionPtr->DeletePacket(packet);
if(ret != 0){
m_bRtpExit = true;
}
}
} while (m_rtpSessionPtr->GotoNextSourceWithData());
}
//else {
// if (m_idleCount != -1)
// {
// ++m_idleCount;//流中断计数
// }
// if (m_noDataCount != 0)
// {
// --m_noDataCount;//没流计数
// }
// //if (m_idleCount > 3000) {
// // m_hVodEndFunc(m_usrParam);
// // m_idleCount = 0;
// //历史流结束的时候,也会出现超时，这个是正常的
// if(m_usrParam && ((VideoSession *)GetUsrParam())->video_type() == VideoType::ERECORD)
// {
// if (m_idleCount > 10000)
// {
// //这里要判断下历史流是否结束，如果未结束，就设置为流中断
// //由于record_stream_status这个函数返回值不准确，所以增加一个进度条大于80%
// if(record_stream_status(((VideoSession *)GetUsrParam())->streamHandle()))
// {
// LOG_INFO("************Record stream is finished**{}**m_progress = {}********", m_deviceID, ((VideoSession *)GetUsrParam())->progress());
// m_idleCount = -1;
// m_hVodEndFunc(m_usrParam);
// record_stream_stop(((VideoSession *)GetUsrParam())->streamHandle());
// ((VideoSession *)GetUsrParam())->streamHandle().clear();
// }
// else
// {
// //如果此时进度大于80% 算完成吧
// if(((VideoSession *)GetUsrParam())->progress() > 0.80)
// {
// LOG_INFO("************Record stream is overtime**{}**m_progress = {}********", m_deviceID, ((VideoSession *)GetUsrParam())->progress());
// m_idleCount = 0;
// m_hVodEndFunc(m_usrParam);
// record_stream_stop(((VideoSession *)GetUsrParam())->streamHandle());
// ((VideoSession *)GetUsrParam())->streamHandle().clear();
// }
// else
// {
// m_idleCount = -1;
// //LOG_ERROR("************post ERROR_REALSTREAM_INTERRUPT to structure****{}********", m_deviceID);
// //发送流中断
// //throw GeneralException2(ERROR_REALSTREAM_INTERRUPT, "Record time video streaming interruption!");
// }
// }
//
//
// }
//
// if (m_noDataCount < -200000)//任务开始时没收到流
// {
// //LOG_ERROR("************m_hVodEndFunc(m_usrParam)！！！m_hVodEndFunc(m_usrParam)********{}******", m_deviceID);
// m_noDataCount = -1;
// //throw GeneralException2(ERROR_REALSTREAM_INTERRUPT, "Record time video streaming interruption2!");
// //m_hVodEndFunc(m_usrParam);
// }
// }
// else//实时任务断流
// //if (m_usrParam && ((VideoSession *)GetUsrParam())->video_type() == VideoType::EREAL)
// {
//
// //每超过3000次，发送一次send_vedio_eof 时长大约1.5s
// //若是30000，时长大约 18s
// if(m_idleCount > 30000)
// {
// uint64_t cts = get_cur_time();
// float duration_not_recv = (cts - last_recv_ts) / 1000.0;
//
// //LOG_ERROR("************I haven't got stream from hik gateway exceed {}s,send eof********{}******", duration_not_recv, m_deviceID);
// m_idleCount = -1;
// //throw GeneralException2(ERROR_REALSTREAM_INTERRUPT, "Real time video streaming interruption!");
// }
//
// if (m_noDataCount < -200000)//任务开始时没收到流
// {
// //LOG_ERROR("************m_noDataCount < -200000********{}******", m_deviceID);
// m_noDataCount = -1;
// //throw GeneralException2(ERROR_REALSTREAM_INTERRUPT, "Real time video streaming interruption2!");
// }
//
// }
//}
//}
// catch (GeneralException2& e)
//{
// //LOG_ERROR("---> video streaming interruption!<---{}, error: {}", m_deviceID, e.err_msg());
// byte_buffer bb(64);
// bb << VasCmd::VAS_CMD_REALSTREAM_INTERRUPT << e.err_msg();
// if (m_usrParam)
// {
// if (((VideoSession *)GetUsrParam())->msgChan()->is_valid()) {
// try {
// ((VideoSession *)GetUsrParam())->msgChan()->send_msg(bb.data_ptr(), bb.data_size());
// }
// catch (GeneralException2& e) {
// //LOG_ERROR("[{}] send vas cmd VAS_CMD_REALSTREAM_INTERRUPT error: {}, {}", m_deviceID, e.err_code(), e.err_str());
// }
// }
// //通知网关关闭句柄
// if(!((VideoSession *)GetUsrParam())->streamHandle().empty())
// {
// LOG_INFO("---->Notify hisense gateway release handle = {} !<----{}", ((VideoSession *)GetUsrParam())->streamHandle().c_str(), m_deviceID);
// if (((VideoSession *)GetUsrParam())->video_type() == VideoType::EREAL)
// real_stream_stop(((VideoSession *)GetUsrParam())->streamHandle());
//
// if (((VideoSession *)GetUsrParam())->video_type() == VideoType::ERECORD)
// record_stream_stop(((VideoSession *)GetUsrParam())->streamHandle());
//
// //清理保活的句柄
// ((VideoSession *)GetUsrParam())->streamHandle().clear();
// }
// }
//
// bb.bset(0);
//
//}
m_rtpSessionPtr->EndDataAccess();
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
LOG_INFO("[{}] OnRtpRecv exited.", m_deviceID);
return 0;
}