RTP发送(TCP UDP两种方式)

// RTPReceiver.cpp #include "RTPReceiver.h" #include<iostream> using namespace std; #pragma comment(lib, "pthreadVC2.lib") #pragma comment(lib,"ws2_32.lib") #pragma warning(disable:4996) #define HOST_IP "192.168.2.100" #define HOST_PORT 1430 #define RECEIVED_STR "RECEIVED" void Callback_RtpPacket(char *buffer, int len)//函数 { //解析 RTPReceiver::rtp_msg_t *prtp = (RTPReceiver::rtp_msg_t*)buffer; if (prtp->header.version != 2) { len = 0; return; } printf("RTP Header Analysis -------------!!! \n"); printf("[Callback_RtpPacket] cc-----%d \r\n", prtp->header.cc); printf("[Callback_RtpPacket] x-----%d \r\n", prtp->header.x); printf("[Callback_RtpPacket] m-----%d \r\n", prtp->header.m); printf("[Callback_RtpPacket] pt-----%x \r\n", prtp->header.pt); printf("[Callback_RtpPacket] p-----%d \r\n", prtp->header.p); printf("[Callback_RtpPacket] version-----%d\r\n ", prtp->header.version); printf("[Callback_RtpPacket] seq-----%d \r\n", prtp->header.seq); printf("[Callback_RtpPacket] ts-----%d \r\n ", prtp->header.ts); printf("[Callback_RtpPacket] ssrc-----%d \r\n", prtp->header.ssrc); printf("[Callback_RtpPacket] payload=%s\r\n", prtp->body); } typedef void(*pCallback_RtpPacket)(char *buffer, int len); pCallback_RtpPacket m_RtpDataFunc = Callback_RtpPacket; //回调函数/函数指针 BOOL thread_running; using namespace std; void * thr_fn(void * arg) { thread_running = true; RTPReceiver *pthis = (RTPReceiver*)arg; pthis->thread_recv_func(); return (void *)0; } int RTPReceiver::thread_create() { pthread_attr_t attr; pthread_mutex_init(&m_real_rtp_mutex, NULL); int ret = -1; pthread_attr_init(&attr); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); ret = pthread_create(&thread, &attr, thr_fn, this); if (0 != ret) { printf("receive thread_create err ret =%d ", ret); return -1; } else { printf("receive thread_create success!\n "); return 0; } } void RTPReceiver::thread_destory() { thread_running = false; //thread 退出代码 pthread_join(thread, NULL); printf("thread destoryed!\n"); } /*********************************************TCP************************************************/ int RTPReceiver::thread_recv_func() { WSADATA wsd = { 0 }; int status_Code = WSAStartup(MAKEWORD(1, 1), &wsd); //返回0成功 cout << "WSAStartup status_Code :" << status_Code << endl; if (status_Code != 0) { cout << "WSAStartup failed" << endl; return status_Code; } SOCKET m_sockServer; sockaddr_in server_addr; sockaddr_in server_addr_from; SOCKET m_server[20]; server_addr.sin_family = AF_INET; server_addr.sin_port = htons(9999); server_addr.sin_addr.S_un.S_addr = inet_addr("127.0.0.1"); m_sockServer = socket(AF_INET, SOCK_STREAM, 0); int iConnect = 0; const int bufSize = 65535; char packetbuffer[bufSize]; char sendBuf[1024]="had received!"; int i = bind(m_sockServer, (sockaddr*)&server_addr, sizeof(server_addr)); if (i != SOCKET_ERROR) { cout << "bind success i=" << i << endl; int iMaxConnect = 20; int lis_code; int len = sizeof(sockaddr); //sockaddr 是系统变量 while (thread_running) { lis_code = listen(m_sockServer, 0); //进行监听 监听等待队列100 //if (lis_code == -1) //{ // cout << "listen failed!!!! lis_code-->" << lis_code << endl; // return lis_code; //} m_server[iConnect] = accept(m_sockServer, (sockaddr*)&server_addr_from, &len);//同意建立连接 if (m_server[iConnect] != INVALID_SOCKET) { iConnect++; if (iConnect>iMaxConnect) { cout << "more than max connect !!! status_Code---" << status_Code << endl; return status_Code; } else { cout << "iConnect--->" << iConnect << endl; --iConnect; while (thread_running) { int status_code = recv(m_server[iConnect], packetbuffer, bufSize, 0); if (status_code != -1) { pthread_mutex_lock(&m_real_rtp_mutex); m_RtpDataFunc(packetbuffer, status_code); pthread_mutex_unlock(&m_real_rtp_mutex); int status_code = send(m_server[iConnect], sendBuf, sizeof(sendBuf), 0); if (status_code == SOCKET_ERROR) { cout << "send Notice failed!!! status_code:" << status_code << endl; } } else { cout << " receive failed !!! status_code--->" << status_code << endl; return status_code; } } } } }//while } else { cout << "bind failed...i=" << i << endl; return i; } closesocket(m_server[iConnect]); WSACleanup(); return TRUE; } /************************************************************************************************/ //int RTPReceiver::thread_recv_func() //{ // //the version bit of Winsock // int version_a = 1;//low bit // int version_b = 1;//high bit // //makeword // WORD versionRequest = MAKEWORD(version_a, version_b); // WSAData wsaData; // int err; // //wsa startup // err = WSAStartup(versionRequest, &wsaData); // if (err != 0) { // printf("ERROR!"); // return -1; // } // //check whether the version is 1.1, if not print the error and cleanup wsa? // if (LOBYTE(wsaData.wVersion) != 1 || HIBYTE(wsaData.wVersion) != 1) // { // printf("WRONG WINSOCK VERSION!"); // WSACleanup(); // return -1; // } // // //build the socket // m_real_rtp_sock = socket(AF_INET, SOCK_DGRAM, 0); // SOCKADDR_IN addr_Srv; // addr_Srv.sin_addr.S_un.S_addr = inet_addr(HOST_IP); // addr_Srv.sin_family = AF_INET; // addr_Srv.sin_port = htons(HOST_PORT); // // if (1) // { // int opt = SO_REUSEADDR; // const char *popt = "1"; // err = setsockopt(m_real_rtp_sock, SOL_SOCKET, SO_REUSEADDR, popt, sizeof(opt)); // if (err == SOCKET_ERROR)//设置失败。 // { // printf("set SO_REUSEADDR error %d\r\n", WSAGetLastError()); // } // } // //bind socket to the host // int resbind = bind(m_real_rtp_sock, (SOCKADDR*)&addr_Srv, sizeof(SOCKADDR)); // if (resbind == SOCKET_ERROR) // { // printf("bind error %d\r\n", WSAGetLastError()); // } // Sleep(1000); // // //address of a client // SOCKADDR_IN addr_Clt; // const int bufSize = 65535; // char packetbuffer[bufSize]; // // int fromlen = sizeof(SOCKADDR); // while (thread_running) // { // unsigned long len = 1; // //FIONREAD：确定套接口s自动读入的数据量，udp返回套接口上排队的第一个数据报大小 // //len=0设置成阻塞模式,len=1设置成非阻塞模式。 // ioctlsocket(m_real_rtp_sock, FIONREAD, &len); // if (len <= 0) // { // Sleep(1000); // continue; // } // while (thread_running && len > 0) // { // memset(packetbuffer, 0, bufSize); // //receive data function of server // int recvlen = recvfrom(m_real_rtp_sock, packetbuffer, bufSize, 0, (SOCKADDR*)&addr_Clt, &fromlen); // if (recvlen != -1) // { // //cout << "recevie buffer is " << packetbuffer << endl; // cout << "recevied buffer" << endl; // } // if (recvlen > 0 && m_RtpDataFunc) // { // pthread_mutex_lock(&m_real_rtp_mutex); // m_RtpDataFunc(packetbuffer, recvlen); // pthread_mutex_unlock(&m_real_rtp_mutex); // } // //send "OK" to the from_client to indicates that data have been received. // sendto(m_real_rtp_sock, RECEIVED_STR, strlen(RECEIVED_STR) + 1, 0, (SOCKADDR*)&addr_Clt, sizeof(SOCKADDR)); // Sleep(10); // } // // } // closesocket(m_real_rtp_sock); // WSACleanup(); // return TRUE; //} int main() { RTPReceiver *self = new RTPReceiver; self->thread_create(); while (1) { Sleep(1); } self->thread_destory(); return 0; }