#include <sys/ioctl.h>
|
#include <pthread.h>
|
#include <poll.h>
|
#include <sys/time.h>
|
#include <sys/types.h>
|
#include <unistd.h>
|
#include <sys/epoll.h>
|
#include <sys/ioctl.h>
|
#include <arpa/inet.h>
|
#include <sys/mman.h>
|
#include "TcpStreamTrans.h"
|
#include "sps_pps.h"
|
|
CTcpStreamTrans::CTcpStreamTrans()
|
{
|
ExitFlag = false;
|
RecvBuffer = NULL;
|
|
m_TcpClientFd = -1;
|
m_TcpServerFd = -1;
|
|
m_pTcpRecvBufTmp = NULL;
|
m_pTcpRecvPacketBuf = NULL;
|
m_sizeOfTcpRecvBufTmp = 0;
|
m_sizeOfTcpRecvPacketBuf = 0;
|
|
m_pHeadTmpBuff = NULL;
|
m_HeadTmplen = 0;
|
m_tcprecvssrc = 0;
|
m_listTcpStreamPacket.clear();
|
|
m_eStreamTransType = E_STREAM_TRANS_TCPACTIVE; //ÂëÁ÷´«ÊäÀàÐÍ
|
m_StreamSendIp[0] = '\0';
|
m_StreamSendPort = 0;
|
m_StreamRecvIp[0] = '\0';
|
m_StreamRecvPort = 0;
|
|
m_StreamParseHandle1 = 0;
|
m_StreamParseHandle2 = 0;
|
m_StreamParseHandle3 = 0;
|
|
m_RecvStreamType = 0;
|
m_eVideoStreamType = E_VIDEO_STREAM_NONE;
|
}
|
|
CTcpStreamTrans::~CTcpStreamTrans()
|
{
|
ThreadQuit();
|
if(m_TcpClientFd > 0)
|
{
|
close(m_TcpClientFd);
|
m_TcpClientFd = -1;
|
}
|
if (m_TcpServerFd > 0)
|
{
|
close(m_TcpServerFd);
|
m_TcpServerFd = -1;
|
}
|
if (m_StreamParseHandle1 != 0)
|
{
|
STREAMPARSE_Close(m_StreamParseHandle1);
|
m_StreamParseHandle1 = 0;
|
}
|
if (m_StreamParseHandle2 != 0)
|
{
|
STREAMPARSE_Close(m_StreamParseHandle2);
|
m_StreamParseHandle2 = 0;
|
}
|
if (m_StreamParseHandle3 != 0)
|
{
|
STREAMPARSE_Close(m_StreamParseHandle3);
|
m_StreamParseHandle3 = 0;
|
}
|
}
|
|
bool CTcpStreamTrans::SetStreamTransParam(StreamTransType_E streamtype, const char *sendip, int sendport, const char *recvip, int recvport)
|
{
|
if (streamtype != E_STREAM_TRANS_TCPACTIVE && streamtype != E_STREAM_TRANS_TCPPASSIVE)
|
{
|
return false;
|
}
|
m_eStreamTransType = streamtype;
|
strncpy2(m_StreamSendIp, sendip, IPSTR_MAX_LEN);
|
m_StreamSendPort = sendport;
|
strncpy2(m_StreamRecvIp, recvip, IPSTR_MAX_LEN);
|
m_StreamRecvPort= recvport;
|
return true;
|
}
|
|
int CTcpStreamTrans::AcceptClient(int socketfd, char *pSrcIp, UINT16 &SrcPort)
|
{
|
printf("%s:GetSockFd:%d AcceptClient enters into!!\n", __FUNCTION__, socketfd);
|
struct sockaddr_in ClientAddr;
|
socklen_t ClientLen = sizeof(struct sockaddr);
|
int NewFd = -1;
|
//UINT16 CliPort;
|
while (true)
|
{
|
struct pollfd polset;
|
|
memset(&polset,0,sizeof(struct pollfd));
|
polset.fd = socketfd;
|
polset.events = POLLIN;
|
polset.revents = 0;
|
int iRet = poll(&polset,1, 5*1000);
|
bool bOK = (iRet > 0) && ( polset.revents & POLLIN);
|
|
if(!bOK)
|
{
|
printf("%s: poll timeout 5s! close the socket!!", __FUNCTION__);
|
return -1;
|
}
|
|
ClientLen = sizeof(struct sockaddr);
|
memset(&ClientAddr,0,ClientLen);
|
|
if ( (NewFd = accept(socketfd, (struct sockaddr *)&ClientAddr, &ClientLen) ) == -1)
|
{
|
int iError = errno;
|
printf("tcp server accept error!(%d)\n",iError);
|
}
|
else
|
{
|
int tcpmaxlen = 1024*1024;
|
setsockopt(NewFd, SOL_SOCKET, SO_RCVBUF, &tcpmaxlen, sizeof(tcpmaxlen));
|
setsockopt(NewFd, SOL_SOCKET, SO_SNDBUF, &tcpmaxlen, sizeof(tcpmaxlen));
|
|
//---------------- Set socket some parameters to avoid network exception ----------------//
|
int keepalive = 1; //´ò¿ªÌ½²â
|
int keepidle = 60; //¿ªÊ¼Ì½²âÇ°µÄ¿ÕÏеȴýʱ¼ä
|
int keepintvl = 10; //·¢ËÍ̽²â·Ö½ÚµÄʱ¼ä¼ä¸ô
|
int keepcnt = 3; //·¢ËÍ̽²â·Ö½ÚµÄ´ÎÊý
|
|
if (setsockopt(NewFd, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepalive, sizeof(keepalive) ) < 0)
|
{
|
perror("fail to set SO_KEEPALIVE");
|
}
|
if (setsockopt(NewFd, SOL_TCP, TCP_KEEPIDLE, (void *)&keepidle, sizeof(keepidle) ) < 0)
|
{
|
perror("fail to set SO_KEEPIDLE");
|
}
|
if (setsockopt(NewFd, SOL_TCP, TCP_KEEPINTVL, (void *)&keepintvl, sizeof(keepintvl) ) < 0)
|
{
|
perror("fail to set SO_KEEPINTVL");
|
}
|
if (setsockopt(NewFd, SOL_TCP, TCP_KEEPCNT, (void *)&keepcnt, sizeof(keepcnt) ) < 0)
|
{
|
perror("fail to set SO_KEEPALIVE");
|
}
|
//---------------------------------------------------------------------------------------//
|
//---------------------------------------------------------------------------------------//
|
//char ClientHostIp[IPSTR_MAX_LEN+1] = {0};
|
//strncpy2(ClientHostIp, pCliIpAddr, IPSTR_MAX_LEN);
|
inet_ntop(AF_INET, (struct in_addr *)&ClientAddr.sin_addr.s_addr, pSrcIp, IPSTR_MAX_LEN); //ʹÓÃḬ̈߳²È«º¯Êý
|
SrcPort = ntohs(ClientAddr.sin_port);
|
printf("NewFd:%d#########new tcp conn:%s,%d#######\n", NewFd, pSrcIp,SrcPort);
|
|
m_TcpClientFd = NewFd;
|
}
|
|
|
return NewFd;
|
}
|
}
|
|
void CTcpStreamTrans::StickyTcpStramPacket(UINT8 *pData, int TcpRecvLen)
|
{
|
if (pData == NULL || TcpRecvLen <= 0)
|
{
|
DBGPrint(M_MRELAY, ERROR_LEVEL,"%s :Tcp Recv Data Error!!", __FUNCTION__);
|
return;
|
}
|
UINT8* pBuff = pData;
|
|
do
|
{
|
if (0 == m_sizeOfTcpRecvBufTmp)
|
{
|
//³¤¶ÈµÄ×ֶηֿª£¬ÐèÒª±£Áô×îºÃÒ»¸ö×Ö½Ú
|
//Ö±½Ó¶ªµô
|
if (TcpRecvLen < TCP_STRAM_HEADER_LEN)
|
{
|
printf("TcpRecvLen(%d) < TCP_STRAM_HEADER_LEN(2) !!!!!!!!!!pBuff[0]:0x%02x", TcpRecvLen, pBuff[0]);
|
if ((TcpRecvLen > 0) && (m_HeadTmplen <= 0))
|
{
|
memcpy(m_pHeadTmpBuff, pBuff, TcpRecvLen);
|
m_HeadTmplen = TcpRecvLen;
|
}
|
break;
|
}
|
if ((m_HeadTmplen > 0) && (m_pHeadTmpBuff != NULL))
|
{
|
memcpy(m_pHeadTmpBuff+m_HeadTmplen, pData, TcpRecvLen);
|
//Õë¶ÔbuffÖØи³Öµ;
|
pBuff = m_pHeadTmpBuff;
|
TcpRecvLen += m_HeadTmplen;
|
m_HeadTmplen = 0;
|
}
|
|
m_sizeOfTcpRecvPacketBuf = htons(*( (UINT16 *)&pBuff[0] ));
|
pBuff += TCP_STRAM_HEADER_LEN;
|
TcpRecvLen -= TCP_STRAM_HEADER_LEN;
|
|
m_sizeOfTcpRecvBufTmp = m_sizeOfTcpRecvPacketBuf;
|
#if 0
|
//Èç¹ûRTP°üµÄ³¤¶È³¬¹ý2000 ±íʾ²»Õý³£µÄ°ü£¬Ö±½Ó¶ªÆú
|
if (m_sizeOfTcpRecvBufTmp > 2000)
|
{
|
DBGPrint(M_MRELAY, ERROR_LEVEL,"vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv%d recvpacklen:%d\n", TcpRecvLen, m_sizeOfTcpRecvPacketBuf);
|
m_sizeOfTcpRecvBufTmp = 0;
|
return;
|
}
|
#endif
|
m_pTcpRecvPacketBuf = (UINT8 *)malloc(m_sizeOfTcpRecvPacketBuf*2);
|
//memset(m_pTcpRecvPacketBuf, 0, sizeof(m_pTcpRecvPacketBuf));
|
m_pTcpRecvBufTmp = m_pTcpRecvPacketBuf;
|
}
|
|
if (TcpRecvLen <= 0)
|
{
|
break;
|
}
|
//ÒѽÓÊյİüÊý¾Ý³¤¶È < °üÊý¾Ý³¤¶È = Ò»¸ö²»ÍêÕûµÄ°ü
|
if (TcpRecvLen < m_sizeOfTcpRecvBufTmp)
|
{
|
memcpy(m_pTcpRecvBufTmp, pBuff, TcpRecvLen);
|
m_pTcpRecvBufTmp += TcpRecvLen;
|
m_sizeOfTcpRecvBufTmp -= TcpRecvLen;
|
|
pBuff += TcpRecvLen;
|
TcpRecvLen -= TcpRecvLen;
|
}
|
else
|
{
|
//ÒѽÓÊյİüÊý¾Ý³¤¶È == °üÊý¾Ý³¤¶È = Ò»¸öÍêÕûµÄ°ü
|
//ÒѽÓÊյİüÊý¾Ý³¤¶È > °üÊý¾Ý³¤¶È = ÖÁÉÙÓÐÒ»¸öÍêÕûµÄ°ü + ÖÁÉÙÒ»¸öÊý¾ÝƬ¶Î
|
memcpy(m_pTcpRecvBufTmp, pBuff, m_sizeOfTcpRecvBufTmp);
|
|
pBuff += m_sizeOfTcpRecvBufTmp;
|
TcpRecvLen -= m_sizeOfTcpRecvBufTmp;
|
|
UINT32 SSrc = ntohl( *( (UINT32 *)&m_pTcpRecvPacketBuf[8]) );
|
|
if (m_tcprecvssrc <= 0)
|
{
|
m_tcprecvssrc = SSrc;
|
}
|
//Èç¹ûssrc²»Ò»Ö£¬ ±íʾ½âÎöʧ°Ü
|
if (m_tcprecvssrc != SSrc)
|
{
|
printf("vvvvvvvoldssrc:%d, newssrc:%d, %d recvpacklen:%d\n", m_tcprecvssrc, SSrc, TcpRecvLen, m_sizeOfTcpRecvPacketBuf);
|
//free
|
if (m_pTcpRecvPacketBuf != NULL)
|
{
|
free(m_pTcpRecvPacketBuf);
|
m_pTcpRecvPacketBuf = NULL;
|
}
|
m_sizeOfTcpRecvBufTmp = 0;
|
m_tcprecvssrc = 0;
|
break;
|
}
|
|
TcpStreamPacket tcppacket;
|
tcppacket.nLength = m_sizeOfTcpRecvPacketBuf;
|
tcppacket.pData = (UINT8*)new char[m_sizeOfTcpRecvPacketBuf+1];
|
memcpy(tcppacket.pData, m_pTcpRecvPacketBuf, tcppacket.nLength);
|
if (tcppacket.nLength > 0)
|
{
|
tcppacket.pData[tcppacket.nLength] = '\0';
|
}
|
m_listTcpStreamPacket.push_back(tcppacket);
|
#if 1
|
//free
|
if (m_pTcpRecvPacketBuf != NULL)
|
{
|
free(m_pTcpRecvPacketBuf);
|
m_pTcpRecvPacketBuf = NULL;
|
}
|
m_sizeOfTcpRecvBufTmp = 0;
|
#else
|
//memset(m_pTcpRecvBufTmp, 0, sizeof(m_pTcpRecvBufTmp));
|
m_pTcpRecvBufTmp = NULL;
|
m_sizeOfTcpRecvBufTmp = 0;
|
#endif
|
}
|
}while(TcpRecvLen > 0);
|
}
|
|
bool CTcpStreamTrans::ReadSelect(int scoketfd, UINT32 timeout)
|
{
|
int res;
|
struct timeval tv;
|
fd_set wrset;
|
int valopt;
|
socklen_t sock_len;
|
|
if (timeout == 0)
|
{
|
tv.tv_sec=tv.tv_usec=0;
|
}
|
else
|
{
|
tv.tv_sec = timeout / 1000000;
|
tv.tv_usec = timeout % 1000000;
|
}
|
|
FD_ZERO(&wrset);
|
FD_SET(scoketfd, &wrset);
|
|
res = select(scoketfd + 1, NULL, &wrset, NULL, &tv);
|
if (res > 0)
|
{
|
sock_len = sizeof(int);
|
if(getsockopt(scoketfd, SOL_SOCKET, SO_ERROR, (void *) (&valopt), &sock_len) == 0)
|
{
|
if(!valopt)
|
return true;
|
}
|
}
|
return false;
|
}
|
|
int CTcpStreamTrans::Connect(void)
|
{
|
if ((strlen(m_StreamSendIp) <= 0) || (m_StreamSendPort <= 0) || (m_StreamRecvPort <= 0))
|
{
|
//wait conenct to server
|
return -1;
|
}
|
|
int socketfd = -1;
|
//Create the TCP socket
|
socketfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
|
if (-1 == socketfd)
|
{
|
perror("CTcpTransport socket");
|
return -1;
|
}
|
//----------------------------------°ó¶¨Ô´IPºÍ¶Ë¿Ú---------------
|
struct sockaddr_in MyAddr;
|
bzero(&MyAddr, sizeof(struct sockaddr_in) );
|
MyAddr.sin_family = AF_INET;
|
MyAddr.sin_addr.s_addr = inet_addr(m_StreamRecvIp);
|
MyAddr.sin_port = htons(m_StreamRecvPort);
|
|
if (bind(socketfd,(struct sockaddr *)&MyAddr, sizeof(struct sockaddr_in) ) == -1)
|
{
|
printf("socket:%d bind %s %d failed!\r\n", socketfd, m_StreamRecvIp,m_StreamRecvPort);
|
perror("bind");
|
//return -1;
|
}
|
else
|
{
|
printf("socket:%d bind local:%s %d sucess!\r\n", socketfd, m_StreamRecvIp, m_StreamRecvPort);
|
}
|
//--------------------------------------------------------------------
|
struct sockaddr_in SvrAddr;
|
bzero(&SvrAddr, sizeof(struct sockaddr_in) );
|
SvrAddr.sin_family = AF_INET;
|
SvrAddr.sin_addr.s_addr = inet_addr(m_StreamSendIp);
|
SvrAddr.sin_port = htons(m_StreamSendPort);
|
|
if (connect(socketfd, (struct sockaddr *)&SvrAddr, sizeof(struct sockaddr_in) ) < 0)
|
{
|
perror("connect");
|
printf("%s:socket:%d connect tcp server:%s %d failed!\r\n",__FUNCTION__, socketfd, m_StreamSendIp,m_StreamSendPort);
|
close(socketfd);
|
socketfd = -1;
|
return -1;
|
}
|
printf("%s: connect tcp server:%s %d sucess by tcp client:%s %d!\r\n",__FUNCTION__,m_StreamSendIp,m_StreamSendPort, m_StreamRecvIp,m_StreamRecvPort);
|
return socketfd;
|
}
|
|
int CTcpStreamTrans::CreateTcpServer(int port)
|
{
|
if (port <= 0)
|
{
|
//wait conenct to server
|
return -1;
|
}
|
//Create the TCP socket
|
m_TcpServerFd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
|
if (-1 == m_TcpServerFd)
|
{
|
perror("CTcpTransport socket");
|
return -1;
|
}
|
|
struct sockaddr_in MyAddr;
|
struct linger linger;
|
int optval;
|
|
bzero(&MyAddr, sizeof(struct sockaddr_in) );
|
MyAddr.sin_family = AF_INET;
|
MyAddr.sin_addr.s_addr = htonl(INADDR_ANY);
|
MyAddr.sin_port = htons(port);
|
|
linger.l_onoff = 1;
|
linger.l_linger = 0;
|
if (setsockopt(m_TcpServerFd, SOL_SOCKET, SO_LINGER, (char *)&linger, sizeof(struct linger) ) == -1)
|
{
|
close(m_TcpServerFd);
|
perror("CTcpTransport set tcp sockopt linger error");
|
return -1;
|
}
|
|
optval = 1;
|
if (setsockopt(m_TcpServerFd, SOL_SOCKET, SO_REUSEADDR, (char *)&optval, sizeof(optval) ) == -1)
|
{
|
close(m_TcpServerFd);
|
perror("CTcpTransport set tcp sockopt reuse address error");
|
return -1;
|
}
|
|
if (bind(m_TcpServerFd, (struct sockaddr *)&MyAddr, sizeof(struct sockaddr_in) ) == -1)
|
{
|
close(m_TcpServerFd);
|
perror("CTcpTransport tcp bind error");
|
return -1;
|
}
|
|
if (listen(m_TcpServerFd, 20) == -1)
|
{
|
close(m_TcpServerFd);
|
return -1;
|
}
|
m_eStreamTransType = E_STREAM_TRANS_TCPPASSIVE;
|
printf("%s: create tcp server port:%d sucess\n", __FUNCTION__, port);
|
return m_TcpServerFd;
|
}
|
|
int CTcpStreamTrans::IsMark( UINT8* pBuffer, unsigned int BufSize)
|
{
|
bool mark = false;
|
if (BufSize > 12) //video
|
{
|
mark= pBuffer[1] & 0x80;
|
}
|
return mark;
|
}
|
|
|
int CTcpStreamTrans::GetRtpHeaderSize(UINT8* _frame, int _frameLen)
|
{
|
int size;
|
size = 12;
|
if (_frameLen < 12)
|
return 0;
|
|
size += (_frame[0] & 0x0f) * 4;
|
|
//ÅжÏÓÐûÓÐÉèÖÃRTPÍ·²¿À©Õ¹
|
if (!(_frame[0] & 0x10))/*ûÓÐÀ©Õ¹*/
|
return size;
|
|
/*Í·²¿À©Õ¹ÁË£¬À©Õ¹Í·¸ñʽ¼ûrfc3550µÄ5.3.1*/
|
//if ((size + 4) < _frameLen)
|
//return (size + 4 + (_frame[size + 2] << 8) + _frame[size + 3]);
|
|
if ((size + 4) < _frameLen)
|
return (size + 4 + 4* ((_frame[size + 2] << 8) + _frame[size + 3]) );
|
|
return 0;
|
}
|
|
int CTcpStreamTrans::ParseTcpStream(UINT8* pData, int len)
|
{
|
RTP_FIXED_HEADER *rtpheader;
|
rtpheader = (RTP_FIXED_HEADER *)pData;
|
int rtpplaytype = rtpheader->payload;
|
int rtpheadlen = GetRtpHeaderSize(pData, len);
|
if (IsVideoPayloadType(rtpplaytype) )
|
{
|
if(len-rtpheadlen > 0)
|
{
|
HandleRecvStream(pData+rtpheadlen, len-rtpheadlen, rtpplaytype, IsMark(pData,len));
|
}
|
}
|
return 0;
|
}
|
|
int CTcpStreamTrans::HandleRecvStream(UINT8* pAFrame, UINT32 FrameSize, int PayLoadType, bool bMark)
|
{
|
do
|
{
|
if (0 == m_StreamParseHandle1)
|
{
|
m_StreamParseHandle1 = STREAMPARSE_Open();
|
if (m_StreamParseHandle1 == 0)
|
{
|
//fail
|
}
|
}
|
|
StreamParseParamIn In;
|
In.pBufIn = pAFrame;
|
In.nSizeIn = FrameSize;
|
In.bMark = bMark;
|
In.PayLoadType = (StreamParseType_E)PayLoadType;
|
StreamParseParamOut Out;
|
|
STREAMPARSE_Parse(m_StreamParseHandle1, &In, &Out);
|
if (Out.nSizeOut <= 0)
|
{
|
break;
|
}
|
if (Out.streamtype==0xC0)
|
{
|
//
|
break;
|
}
|
if((m_RecvStreamType == 0) && (Out.streamtype != 0))
|
{
|
m_RecvStreamType = Out.streamtype;
|
printf("%s:1. >>>>>>>>>Detecing Stream Type... StreamType:%02x!!", __FUNCTION__, Out.streamtype);
|
}
|
|
if(m_RecvStreamType == 0)
|
{
|
printf("%s:2. Detecing Stream Type... StreamType:%d!!", __FUNCTION__, Out.streamtype);
|
break;
|
}
|
|
int pts = Out.pts/90;
|
|
StreamParseParamIn StreamIn;
|
if(m_RecvStreamType == 0xE2)//MPEG4
|
{
|
StreamIn.PayLoadType = STREAMPARSE_MERGEAMPEG4;
|
}
|
else if(m_RecvStreamType == 0xE3)//MPEG2
|
{
|
StreamIn.PayLoadType = STREAMPARSE_MERGEAMPEG2;
|
}
|
else//H264
|
{
|
StreamIn.PayLoadType = STREAMPARSE_ESTOES;
|
}
|
if (0 == m_StreamParseHandle2)
|
{
|
m_StreamParseHandle2 = STREAMPARSE_Open();
|
if (m_StreamParseHandle2 == 0)
|
{
|
//fail
|
}
|
}
|
StreamIn.pBufIn = Out.pBufOut;
|
StreamIn.nSizeIn = Out.nSizeOut;
|
StreamIn.bMark = false;
|
StreamParseParamOut StreamOut;
|
|
STREAMPARSE_Parse(m_StreamParseHandle2, &StreamIn, &StreamOut);
|
if (StreamOut.nSizeOut > 0)
|
{
|
int FrameType = GB_VIDEO_FRAME_P;
|
VideoStreamType_E eVideoStreamType = E_VIDEO_STREAM_H264;
|
if(m_RecvStreamType == 0xE2)
|
{
|
eVideoStreamType = E_VIDEO_STREAM_MPEG4;
|
mpeg_header_t* VOL = (mpeg_header_t*)StreamOut.pBufOut;
|
if(is_VOL_header(VOL))
|
{
|
FrameType = GB_VIDEO_FRAME_I;
|
}
|
}
|
else if(m_RecvStreamType == 0xE3)
|
{
|
eVideoStreamType = E_VIDEO_STREAM_MPEG2;
|
mpeg_header_t* SH = (mpeg_header_t*)StreamOut.pBufOut;
|
if(is_sequence_header(SH))
|
{
|
FrameType = GB_VIDEO_FRAME_I;
|
}
|
}
|
else if(m_RecvStreamType == 0xE1)
|
{
|
eVideoStreamType = E_VIDEO_STREAM_H265;
|
if (is_h265_IFrame_Data(StreamOut.pBufOut, StreamOut.nSizeOut))
|
{
|
FrameType = GB_VIDEO_FRAME_I;
|
}
|
}
|
else
|
{
|
eVideoStreamType = E_VIDEO_STREAM_H264;
|
if (is_H264_IFrame_Data(StreamOut.pBufOut, StreamOut.nSizeOut))
|
{
|
FrameType = GB_VIDEO_FRAME_I;
|
}
|
}
|
|
if (m_eVideoStreamType == E_VIDEO_STREAM_NONE)
|
{
|
m_eVideoStreamType = eVideoStreamType;
|
printf("%s:>>>>>>>>>Detecing Stream Type:%d!!!!!!!!", __FUNCTION__, (int)m_eVideoStreamType);
|
}
|
if (MediaCBFunc != NULL)
|
{
|
MediaCBFunc((int)m_eVideoStreamType, FrameType, StreamOut.pBufOut, StreamOut.nSizeOut, UserData);
|
}
|
}
|
break;
|
}while(0);
|
}
|
|
void* CTcpStreamTrans::Thread()
|
{
|
ThreadStarted();
|
|
int RecvLen = 0;
|
if (RecvBuffer == NULL)
|
{
|
RecvBuffer = new unsigned char[UDP_PACKET_RECV_MAXSIZE+1];
|
RecvBuffer[0] = '\0';
|
}
|
|
if (m_eStreamTransType == E_STREAM_TRANS_TCPPASSIVE && m_TcpServerFd <= 0)
|
{
|
m_TcpServerFd = CreateTcpServer(m_StreamRecvPort);
|
if (m_TcpServerFd <= 0)
|
{
|
printf("%s: create tcp server listen failed!!!!\n", __FUNCTION__);
|
return NULL;
|
}
|
}
|
|
m_pHeadTmpBuff = (UINT8 *)malloc(TCP_STREAM_BUFFER_MAX_LEN);
|
m_HeadTmplen = 0;
|
m_listTcpStreamPacket.clear();
|
TcpStreamPacket TcpRecvPacket;
|
TcpPacketIt it;
|
|
int m_DataIncomingTime = GetClockTime();
|
int WaitForTimeout = 10 * 1000;
|
struct pollfd FdSet[1] = { {0, 0} };
|
while(!ExitFlag)
|
{
|
if (m_eStreamTransType == E_STREAM_TRANS_TCPPASSIVE)
|
{
|
if (m_TcpClientFd <= 0)
|
{
|
char sendip[IPSTR_MAX_LEN+1] = {0};
|
UINT16 sendport = 0;
|
m_TcpClientFd = AcceptClient(m_TcpServerFd, sendip, sendport);
|
if (m_TcpClientFd <= 0)
|
{
|
//usleep(100*1000);
|
continue;
|
}
|
else
|
{
|
//ÊÇ·ñÆ¥ÅäÔ´ipºÍport
|
//ÔÝʱĬÈÏÓÐÁ¬½Ó¾ÍËãÊǽÓÊÕµ½ÂëÁ÷
|
//continue;
|
FdSet[0].fd = m_TcpClientFd;
|
FdSet[0].events = POLLIN | POLLPRI;
|
}
|
}
|
}
|
else
|
{
|
if (m_TcpClientFd <= 0)
|
{
|
m_TcpClientFd = Connect();
|
//tcpclient
|
if (m_TcpClientFd < 0)
|
{
|
printf("+++++++++++++++++++++m_TcpClientFd:%d\n", m_TcpClientFd);
|
usleep(1000*1000);
|
continue;
|
}
|
FdSet[0].fd = m_TcpClientFd;
|
FdSet[0].events = POLLIN | POLLPRI;
|
}
|
}
|
//ÂÖѯÍøÂçFD£¬ÓÐÊý¾Ý»òÕßÒì³£Á¢¼´·µ»Ø£¬ÎÞÊý¾ÝµÈ´ý³¬Ê±¡£
|
if ( (poll(FdSet, 1, WaitForTimeout) < 0) && (errno != EINTR) )
|
{
|
printf("%s: #####%s:%d poll error: SockFd<%d>!######", __FUNCTION__, __FILE__,__LINE__,m_TcpClientFd);
|
return NULL;
|
}
|
|
//ÅжÏÊý¾Ý¿É¿¿ÐÔ£¿
|
if ( ( (FdSet[0].revents & POLLIN) == POLLIN) || ( (FdSet[0].revents & POLLPRI) == POLLPRI) )
|
{
|
RecvLen = recv(m_TcpClientFd, RecvBuffer, TCP_STREAM_BUFFER_MAX_LEN, 0);
|
if (RecvLen > 0)
|
{
|
StickyTcpStramPacket(RecvBuffer, RecvLen);
|
while (m_listTcpStreamPacket.size() > 0)
|
{
|
it = m_listTcpStreamPacket.begin();
|
TcpRecvPacket = *it;
|
m_listTcpStreamPacket.erase(it);
|
|
if (TcpRecvPacket.nLength > 0)
|
{
|
//»ñȡÿһ°üµÄÊý¾Ý
|
ParseTcpStream(TcpRecvPacket.pData, TcpRecvPacket.nLength);
|
}
|
if (TcpRecvPacket.pData != NULL)
|
{
|
delete []((UINT8*)TcpRecvPacket.pData);
|
TcpRecvPacket.pData = NULL;
|
TcpRecvPacket.nLength = 0;
|
}
|
}
|
}
|
else
|
{
|
printf("%s:Recv Data Len<%d> From socket<%d>, Connect Failed!", __FILE__, RecvLen, m_TcpClientFd);
|
break;
|
}
|
}
|
|
// ³¬¹ý5ÃëûÓнÓÊÕµ½ÂëÁ÷´òÓ¡ÐÅÏ¢
|
int now = GetClockTime();
|
if (now - m_DataIncomingTime > 5)
|
{
|
printf("%s:>>>>>>>tcp sendinfo<%s:%d> recvinfo<%s:%d> over 5 seconds has not stream!!!!!", __FUNCTION__, m_StreamSendIp, m_StreamSendPort, m_StreamRecvIp, m_StreamRecvPort);
|
}
|
m_DataIncomingTime = now;
|
}
|
|
printf("%s: Stream Thread Exit %p, Jstream::ThreadQuit Sucess!!!\n", __FUNCTION__, this);
|
return NULL;
|
}
|
|
void CTcpStreamTrans::ThreadQuit(void)
|
{
|
ExitFlag = true;
|
usleep(100000);
|
|
if (m_pTcpRecvPacketBuf != NULL)
|
{
|
free(m_pTcpRecvPacketBuf);
|
m_pTcpRecvPacketBuf = NULL;
|
}
|
|
if(RecvBuffer != NULL)
|
{
|
delete [] RecvBuffer;
|
RecvBuffer = NULL;
|
}
|
|
if (m_pHeadTmpBuff != NULL)
|
{
|
free(m_pHeadTmpBuff);
|
m_pHeadTmpBuff = NULL;
|
}
|
m_HeadTmplen = 0;
|
|
TcpPacketIt it;
|
while (!m_listTcpStreamPacket.empty())
|
{
|
it = m_listTcpStreamPacket.begin();
|
delete []((UINT8*)it->pData);
|
it->pData = NULL;
|
m_listTcpStreamPacket.erase(it);
|
}
|
}
|
|
|
