src/decoder/gb28181/rtp/RTPUdpReceiver.cpp


#include "RTPUdpReceiver.h"
#include <iostream>
#include <time.h>
#include <thread>
#include <chrono>
#include "../common_header.h"
#include "../websocket/WebsocketClient.h"
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 int rtp_revc_thread_(void* param)
{
	if (!param)
	{
		return -1;
	}
	RTPUdpReceiver* self = (RTPUdpReceiver*)param;
	return self->OnRtpRecv();
}
static int connecting_thread_(void* param)
{
	if (!param) {
		return -1;
	}
	RTPUdpReceiver* self = (RTPUdpReceiver*)param;
	return self->IsConnecting();
}
RTPUdpReceiver::RTPUdpReceiver() 
:m_bOpened(false)
, m_idleCount(-1)
,m_noDataCount(-1)
{
	m_bRtpExit = false;
	m_sessparamsPtr = new RTPSessionParams();
	m_transparamsPtr = new RTPUDPv4TransmissionParams();
	m_rtpSessionPtr = new UdpRTPSession();
}
RTPUdpReceiver::~RTPUdpReceiver()
{
	m_bRtpExit = true;
	if(nullptr != m_sessparamsPtr){
		delete m_sessparamsPtr;
		m_sessparamsPtr = nullptr;
	}
	if(nullptr != m_transparamsPtr){
		delete m_transparamsPtr;
		m_transparamsPtr = nullptr;
	}
	if (nullptr != m_connThreadPtr && m_connThreadPtr->joinable()) {
        m_connThreadPtr->join();
		delete m_connThreadPtr;
		m_connThreadPtr = nullptr;
    }
	if(nullptr != m_rtpSessionPtr){
		delete m_rtpSessionPtr;
		m_rtpSessionPtr = nullptr;
	}
}
bool RTPUdpReceiver::Open(string channel_id)
{
	m_SipChannelId = channel_id;
	m_rtp_port = allocRtpPort();
	if (m_rtp_port < 0) {
		return false;
	}
	m_sessparamsPtr->SetUsePollThread(true);
	m_sessparamsPtr->SetMinimumRTCPTransmissionInterval(10);
	m_sessparamsPtr->SetOwnTimestampUnit(1.0/90000.0);
	m_sessparamsPtr->SetAcceptOwnPackets(true);
	
	m_transparamsPtr->SetPortbase(m_rtp_port);
	m_transparamsPtr->SetRTPReceiveBuffer(kRtpRecvBufferSize);
    LOG_INFO("[{}] port: {}", m_SipChannelId, m_rtp_port);
	
	int err = m_rtpSessionPtr->Create(*m_sessparamsPtr, m_transparamsPtr);
	if (err != 0) {	
		LOG_ERROR("[{}] Create error: {}", m_SipChannelId, err);
		return false;
	}
	m_rtpThreadPtr = new std::thread(rtp_revc_thread_, this);
	if (nullptr == m_rtpThreadPtr) {
		LOG_ERROR("[{}] Create m_rtpThreadPtr error", m_SipChannelId);
		return false;
	}
	if (InitPS() != 0) {
		return false;
	}
	// InitPS()成功就得起该线程，因为ClosePsThread是在这里完成的
	m_connThreadPtr = new std::thread(connecting_thread_, this);
	if (nullptr == m_connThreadPtr) {
		LOG_ERROR("[{}] Create m_connThreadPtr error", m_SipChannelId);
		return false;
	}
	bool bReq = RequestStream();
	if (!bReq) {
		LOG_INFO("[{}] RequestStream failed !", m_SipChannelId);
		Close();
		return false;
	}
	m_bOpened = true;
	m_bNoData = false;
    LOG_INFO("[{}] Open ok", m_SipChannelId);
	return true;
}
bool RTPUdpReceiver::IsOpened()
{
	return m_bOpened;
}
void RTPUdpReceiver::Close()
{
    m_bRtpExit = true;
}
// 收RTP包线程
int RTPUdpReceiver::OnRtpRecv()
{
	if(nullptr == m_rtpSessionPtr){
		return -1;
	}
	LOG_INFO("[{}] OnRtpRecv started.", m_SipChannelId);
	while (!m_bRtpExit)
	{
		//try
		//{
			m_rtpSessionPtr->Poll();
			m_rtpSessionPtr->BeginDataAccess();
			
			if (m_rtpSessionPtr->GotoFirstSourceWithData())
			{
				// LOG_INFO("OnRtpRecv GotoFirstSourceWithData --{}", m_SipChannelId);
    			last_recv_ts = UtilTools::get_cur_time_ms();
				m_idleCount = 0;
				m_noDataCount = 0;
				do
				{
					RTPPacket* packet;
					while ((packet = m_rtpSessionPtr->GetNextPacket()) != NULL) 
					{
						m_bNoData = false;
						// LOG_INFO("OnRtpRecv GetNextPacket --{}", m_SipChannelId);
						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_SipChannelId, ((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_SipChannelId, ((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_SipChannelId);
			//					//发送流中断
			//					//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_SipChannelId);	 
			//			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 = UtilTools::get_cur_time_ms();
   // 					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_SipChannelId);	 
			//			m_idleCount = -1;
			//			//throw GeneralException2(ERROR_REALSTREAM_INTERRUPT, "Real time video streaming interruption!");
			//		} 
			//		
			//		if (m_noDataCount < -200000)//任务开始时没收到流
			//		{
   // 					//LOG_ERROR("************m_noDataCount < -200000********{}******", m_SipChannelId);	 
			//			m_noDataCount = -1;
			//			//throw GeneralException2(ERROR_REALSTREAM_INTERRUPT, "Real time video streaming interruption2!");
			//		}
			//		
			//	}
			//}
		//}
  //      catch (GeneralException2& e)
		//{
  //  		//LOG_ERROR("---> video streaming interruption!<---{}, error: {}", m_SipChannelId, 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_SipChannelId, e.err_code(), e.err_str());
  //      			}    			
  //              }
		//		//通知网关关闭句柄
		//		if(!((VideoSession *)GetUsrParam())->streamHandle().empty())
		//		{
  //  				LOG_INFO("---->Notify hisense gateway release handle = {} !<----{}", ((VideoSession *)GetUsrParam())->streamHandle().c_str(), m_SipChannelId);
  //  				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_SipChannelId);
	return 0;
}
bool RTPUdpReceiver::RequestStream() {
	WebsocketClient* pClient = WebsocketClient::getInstance();
	if (pClient){
		if (pClient->InviteUdp(m_SipChannelId, m_rtp_port, this) < 0) {
			return false;
		}
	}
	return true;
}
int RTPUdpReceiver::IsConnecting() {
	LOG_INFO("[{}] IsConnecting started.", m_SipChannelId);
	int count = 0;
	while (!m_bRtpExit)
	{
		if (m_bNoData) {
			bool bReq = RequestStream();
			if (!bReq) {
				LOG_INFO("[{}] RequestStream failed !", m_SipChannelId);
			}
			wait_times(30); // 等待3s
			
			count++;
			if (count > 10) {
				// 30s 依然没数据过来，则关闭
				m_bRtpExit = true;
				break;
			}
		} else {
			m_bNoData = true;
			wait_times(100);	// 等待10s， 10s之内正常有数据的情况 m_bNoData 已经被置为false
		}
	}
	LOG_DEBUG("[{}] while", m_SipChannelId);
	// 结束整个任务
	WebsocketClient* pClient = WebsocketClient::getInstance();
	if (pClient){
		pClient->ByeInvite(m_SipChannelId, m_rtp_port);
	}
	LOG_DEBUG("[{}] ByeInvite", m_SipChannelId);
    
    // rtp接收线程退出
    if (nullptr != m_rtpThreadPtr && m_rtpThreadPtr->joinable())
    {
        m_rtpThreadPtr->join();
		delete m_rtpThreadPtr;
		m_rtpThreadPtr = nullptr;
    }
	LOG_DEBUG("[{}] m_rtpThreadPtr", m_SipChannelId);
    m_rtpSessionPtr->Destroy();
	LOG_DEBUG("[{}] m_rtpSessionPtr", m_SipChannelId);
	ClosePsThread();
    m_bOpened = false;
	LOG_INFO("[{}] IsConnecting exited.", m_SipChannelId);
	return 0;
}
// 对退出命令敏感的延时
bool RTPUdpReceiver::wait_times(int times) {
	int count_sleep = times;
	while (!m_bRtpExit) {
		count_sleep-- ;
		if (count_sleep <= 0) {
			count_sleep = times;
			break;
		}
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
	}
}