Hu Chunming / vpt_ascend_arm

  
  #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->CheckConnecting();

  }
  

  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;
  	}
  

  	m_bRecvExit = false;
  

  	LOG_INFO("[{}] OnRtpRecv started.", m_SipChannelId);

  	while (!m_bRecvExit)

  	{

  		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_bRecvExit = true;

  					}

  				}
  			} while (m_rtpSessionPtr->GotoNextSourceWithData());
  		}
  

  		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::CheckConnecting() {
  	LOG_INFO("[{}] CheckConnecting 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(50); // 等待5s

  			
  			count++;
  

  			if (count > 60) {
  				// 3min 依然没数据过来，则关闭

  				m_bRtpExit = true;
  				break;
  			}
  		} else {
  			m_bNoData = true;
  
  			wait_times(100);	// 等待10s， 10s之内正常有数据的情况 m_bNoData 已经被置为false
  		}
  	}
  

  	m_bRecvExit = true;

  
  	// 结束整个任务
  	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;
      }
  

      m_rtpSessionPtr->Destroy();
  

  	ClosePsThread();
  
      m_bOpened = false;
  

  	LOG_INFO("[{}] CheckConnecting 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));
  	}
  }