流媒体-RTMP协议-librtmp库学习（二）
生活随笔
收集整理的這篇文章主要介紹了
流媒体-RTMP协议-librtmp库学习（二）
小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.
文章目錄
- librtmp 庫相關(guān)結(jié)構(gòu)體
- RTMP 結(jié)構(gòu)體
- RTMP_LNK 結(jié)構(gòu)體
- RTMPPacket 結(jié)構(gòu)體：描述實時消息協(xié)議的分塊
- AMF(Action Message Format): 動作消息格式
- AVal結(jié)構(gòu)體：自定義字符串
- 推流
- InitSockets() 初始化Socket
- RTMP_Alloc() 用于創(chuàng)建一個RTMP的結(jié)構(gòu)體
- RTMP_Init() 初始化結(jié)構(gòu)體
- RTMP_SetupURL() 設(shè)置推拉流的URL
- RTMP_ParseURL() 解析輸入URL
- RTMP_EnableWrite() 設(shè)置為推流狀態(tài)
- RTMP_Connect() 建立NetConnection
- RTMP_Connect0() 第0次連接，建立socket連接
- RTMP_Connect1() 第1次連接，建立真正的rtmp連接
- HandShake() 握手過程實現(xiàn)(C0+C1， S0+S1+S2， C2)
- SendConnectPacket() 發(fā)送命令請求
- RTMP_ConnectStream() 建立NetStream
- RTMP_ClientPacket() 處理收到的消息
- RTMP_SendPacket() 消息流發(fā)送，內(nèi)含分塊處理
- RTMP_Close() 關(guān)閉連接
- CleanupSockets() 清理Socket
- 拉流
- RTMP_ReadPacket() 讀取RTMPPacket包
- Download() 下載函數(shù)
- 參考文獻
librtmp 庫相關(guān)結(jié)構(gòu)體
RTMP 結(jié)構(gòu)體
表示一個rtmp連接
typedef struct RTMP{int m_inChunkSize; //拉流流分塊大小：初始化默認(rèn)128字節(jié) RTMP_DEFAULT_CHUNKSIZE 128int m_outChunkSize; //推流分塊大小：初始化默認(rèn)128字節(jié) RTMP_DEFAULT_CHUNKSIZE 128int m_nBWCheckCounter;int m_nBytesIn;int m_nBytesInSent;int m_nBufferMS;int m_stream_id; /* returned in _result from createStream */int m_mediaChannel;uint32_t m_mediaStamp;uint32_t m_pauseStamp;int m_pausing;int m_nServerBW;int m_nClientBW;uint8_t m_nClientBW2;uint8_t m_bPlaying;uint8_t m_bSendEncoding;uint8_t m_bSendCounter;int m_numInvokes;int m_numCalls;RTMP_METHOD *m_methodCalls; /* remote method calls queue */RTMPPacket *m_vecChannelsIn[RTMP_CHANNELS];RTMPPacket *m_vecChannelsOut[RTMP_CHANNELS];int m_channelTimestamp[RTMP_CHANNELS]; /* abs timestamp of last packet */double m_fAudioCodecs; /* audioCodecs for the connect packet */double m_fVideoCodecs; /* videoCodecs for the connect packet */double m_fEncoding; /* AMF0 or AMF3 */double m_fDuration; /* duration of stream in seconds */int m_msgCounter; /* RTMPT stuff */int m_polling;int m_resplen;int m_unackd;AVal m_clientID;RTMP_READ m_read;RTMPPacket m_write;RTMPSockBuf m_sb;RTMP_LNK Link;} RTMP;RTMP_LNK 結(jié)構(gòu)體
typedef struct RTMP_LNK{AVal hostname;AVal sockshost;AVal playpath0; /* parsed from URL */AVal playpath; /* passed in explicitly */AVal tcUrl;AVal swfUrl;AVal pageUrl;AVal app;AVal auth;AVal flashVer;AVal subscribepath;AVal token;AMFObject extras;int edepth;int seekTime;int stopTime;#define RTMP_LF_AUTH 0x0001 /* using auth param */ #define RTMP_LF_LIVE 0x0002 /* stream is live */ #define RTMP_LF_SWFV 0x0004 /* do SWF verification */ #define RTMP_LF_PLST 0x0008 /* send playlist before play */ #define RTMP_LF_BUFX 0x0010 /* toggle stream on BufferEmpty msg */ #define RTMP_LF_FTCU 0x0020 /* free tcUrl on close */int lFlags;int swfAge;int protocol;int timeout; /* connection timeout in seconds */unsigned short socksport;unsigned short port;#ifdef CRYPTO #define RTMP_SWF_HASHLEN 32void *dh; /* for encryption */void *rc4keyIn;void *rc4keyOut;uint32_t SWFSize;uint8_t SWFHash[RTMP_SWF_HASHLEN];char SWFVerificationResponse[RTMP_SWF_HASHLEN+10]; #endif} RTMP_LNK;RTMPPacket 結(jié)構(gòu)體：描述實時消息協(xié)議的分塊
//塊由頭和數(shù)據(jù)組成，原始的rtmp消息塊表示其中的數(shù)據(jù)都是沒有經(jīng)過解析的，是原始的字節(jié)流#define RTMP_MAX_HEADER_SIZE 18typedef struct RTMPChunk{int c_headerSize; //頭部的長度int c_chunkSize; //chunk的大小char *c_chunk; //數(shù)據(jù)char c_header[RTMP_MAX_HEADER_SIZE]; //chunk頭部} RTMPChunk;typedef struct RTMPPacket{uint8_t m_headerType; //basic header 中的type頭字節(jié)，值為(0,1,2,3)表示ChunkMsgHeader的類型（4種）uint8_t m_packetType; //Chunk Msg Header中msg type 1字節(jié)：消息類型id（8: audio；9:video；18:AMF0編碼的元數(shù)據(jù)）uint8_t m_hasAbsTimestamp; //bool值，是否是絕對時間戳(類型1時為true)int m_nChannel; //塊流ID ，通過設(shè)置ChannelID來設(shè)置Basic stream id的長度和值uint32_t m_nTimeStamp; //時間戳，消息頭前三字節(jié)int32_t m_nInfoField2; //Chunk Msg Header中msg StreamID 4字節(jié)：消息流iduint32_t m_nBodySize; //Chunk Msg Header中msg length 4字節(jié)：消息長度uint32_t m_nBytesRead; //已讀取的數(shù)據(jù)RTMPChunk *m_chunk; //raw chunk結(jié)構(gòu)體指針，把RTMPPacket的真實頭部和數(shù)據(jù)段拷貝進來char *m_body; //數(shù)據(jù)段指針} RTMPPacket;RTMPPacket_Alloc() RTMPPacket_Alloc() RTMPPacket_Free() RTMPPacket_Dump()
void RTMPPacket_Reset(RTMPPacket *p) {p->m_headerType = 0;p->m_packetType = 0;p->m_nChannel = 0;p->m_nTimeStamp = 0;p->m_nInfoField2 = 0;p->m_hasAbsTimestamp = FALSE;p->m_nBodySize = 0;p->m_nBytesRead = 0; }int RTMPPacket_Alloc(RTMPPacket *p, int nSize) {char *ptr = calloc(1, nSize + RTMP_MAX_HEADER_SIZE);if (!ptr)return FALSE;p->m_body = ptr + RTMP_MAX_HEADER_SIZE;p->m_nBytesRead = 0;return TRUE; }void RTMPPacket_Free(RTMPPacket *p) {if (p->m_body){free(p->m_body - RTMP_MAX_HEADER_SIZE);p->m_body = NULL;} }void RTMPPacket_Dump(RTMPPacket *p) {RTMP_Log(RTMP_LOGDEBUG,"RTMP PACKET: packet type: 0x%02x. channel: 0x%02x. info 1: %d info 2: %d. Body size: %lu. body: 0x%02x",p->m_packetType, p->m_nChannel, p->m_nTimeStamp, p->m_nInfoField2,p->m_nBodySize, p->m_body ? (unsigned char)p->m_body[0] : 0); }AMF(Action Message Format): 動作消息格式
它是一種二進制的數(shù)據(jù)格式，++AMF數(shù)據(jù)采用Big-Endian（大端模式，高字節(jié)存儲在內(nèi)存的低地址端）++。它的設(shè)計是為了把actionscript里面的數(shù)據(jù)(包括Object,Array,Boolean,Number等)序列化成二進制數(shù)據(jù)，然后把這段數(shù)據(jù)隨意發(fā)送給其他接收方程序，比如發(fā)給遠(yuǎn)程的服務(wù)器，在遠(yuǎn)程服務(wù)器那邊，可以把這段數(shù)據(jù)給還原出來，以此達到一個數(shù)據(jù)傳輸?shù)淖饔谩?/p>
AVal結(jié)構(gòu)體：自定義字符串
// AMF自定義的字符串; typedef struct AVal {char *av_val; //字符串指針int av_len; //字符串長度 } AVal;// AVal的快速初始化; #define AVC(str) {str, sizeof(str)-1}// 比較AVal字符串; #define AVMATCH(a1,a2) ((a1)->av_len == (a2)->av_len && !memcmp((a1)->av_val,(a2)->av_val,(a1)->av_len))推流
InitSockets() 初始化Socket
int InitSockets() { #ifdef WIN32WORD version;WSADATA wsaData;version = MAKEWORD(1, 1);return (WSAStartup(version, &wsaData) == 0); #elsereturn TRUE; #endif }RTMP_Alloc() 用于創(chuàng)建一個RTMP的結(jié)構(gòu)體
RTMP* RTMP_Alloc() {return calloc(1, sizeof(RTMP)); }RTMP_Init() 初始化結(jié)構(gòu)體
void RTMP_Init(RTMP *r) { #ifdef CRYPTOif (!RTMP_TLS_ctx)RTMP_TLS_Init(); #endifmemset(r, 0, sizeof(RTMP));r->m_sb.sb_socket = -1;r->m_inChunkSize = RTMP_DEFAULT_CHUNKSIZE;r->m_outChunkSize = RTMP_DEFAULT_CHUNKSIZE;r->m_nBufferMS = 30000;r->m_nClientBW = 2500000;r->m_nClientBW2 = 2;r->m_nServerBW = 2500000;r->m_fAudioCodecs = 3191.0;r->m_fVideoCodecs = 252.0;r->Link.timeout = 30;r->Link.swfAge = 30; }RTMP_SetupURL() 設(shè)置推拉流的URL
內(nèi)部調(diào)用了RTMP_ParseURL主要用于解析url
RTMP_ParseURL() 解析輸入URL
RTMP_EnableWrite() 設(shè)置為推流狀態(tài)
//是否推流，默認(rèn)不推流 void RTMP_EnableWrite(RTMP *r) {r->Link.protocol |= RTMP_FEATURE_WRITE; }RTMP_Connect() 建立NetConnection
主要分為RTMP_Connect0()函數(shù)(第0次連接，建立socket連接)+ RTMP_Connect1函數(shù)(第1次連接，建立真正的rtmp連接。主要分為HandShake握手、SendConnectPacket發(fā)送命令請求)
RTMP_Connect0() 第0次連接，建立socket連接
int RTMP_Connect0(RTMP *r, struct sockaddr * service) {int on = 1;r->m_sb.sb_timedout = FALSE;r->m_pausing = 0;r->m_fDuration = 0.0;r->m_sb.sb_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);if (r->m_sb.sb_socket != -1){if (connect(r->m_sb.sb_socket, service, sizeof(struct sockaddr)) < 0){int err = GetSockError();RTMP_Log(RTMP_LOGERROR, "%s, failed to connect socket. %d (%s)",__FUNCTION__, err, strerror(err));RTMP_Close(r);return FALSE;}if (r->Link.socksport){RTMP_Log(RTMP_LOGDEBUG, "%s ... SOCKS negotiation", __FUNCTION__);if (!SocksNegotiate(r)){RTMP_Log(RTMP_LOGERROR, "%s, SOCKS negotiation failed.", __FUNCTION__);RTMP_Close(r);return FALSE;}}}else{RTMP_Log(RTMP_LOGERROR, "%s, failed to create socket. Error: %d", __FUNCTION__,GetSockError());return FALSE;}/* set timeout */{SET_RCVTIMEO(tv, r->Link.timeout);if (setsockopt(r->m_sb.sb_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv, sizeof(tv))){RTMP_Log(RTMP_LOGERROR, "%s, Setting socket timeout to %ds failed!",__FUNCTION__, r->Link.timeout);}}setsockopt(r->m_sb.sb_socket, IPPROTO_TCP, TCP_NODELAY, (char *) &on, sizeof(on));return TRUE; }RTMP_Connect1() 第1次連接，建立真正的rtmp連接
主要分為HandShake握手、SendConnectPacket發(fā)送命令請求)
HandShake() 握手過程實現(xiàn)(C0+C1， S0+S1+S2， C2)
在C0中這個字段表示客戶端要求的RTMP版本 。在S0中這個字段表示服務(wù)器選擇的RTMP版本。本規(guī)范所定義的版本是3；0-2是早期產(chǎn)品所用的，已被丟棄；4-31保留在未來使用 ；32-255不允許使用 （為了區(qū)分其他以某一字符開始的文本協(xié)議）。如果服務(wù)無法識別客戶端請求的版本，應(yīng)該返回3 。客戶端可以選擇減到版本3或選擇取消握手
- 時間：4字節(jié)：本字段包含時間戳。該時間戳應(yīng)該是發(fā)送這個數(shù)據(jù)塊的端點的后續(xù)塊的時間起始點。可以是0，或其他的任何值。為了同步多個流，端點可能發(fā)送其塊流的當(dāng)前值。
- 零：4字節(jié)：本字段必須是全零。
- 隨機數(shù)據(jù)：1528字節(jié)。本字段可以包含任何值。因為每個端點必須用自己初始化的握手和對端初始化的握手來區(qū)分身份，所以這個數(shù)據(jù)應(yīng)有充分的隨機性。但是并不需要加密安全的隨機值，或者動態(tài)值
- 時間：4字節(jié)：本字段必須包含對等段發(fā)送的時間（對C2來說是S1，對S2來說是C1）。
- 時間2：4字節(jié)：本字段必須包含先前發(fā)送的并被對端讀取的包的時間戳。
- 隨機回復(fù)：1528字節(jié)：本字段必須包含對端發(fā)送的隨機數(shù)據(jù)字段（對C2來說是S1，對S2來說是C1）。每個對等端可以用時間和時間2字段中的時間戳來快速地估計帶寬和延遲。但這樣做可能并不實用。
SendConnectPacket() 發(fā)送命令請求
發(fā)送"connect"等各種字符串到服務(wù)端
#define SAVC(x) static const AVal av_##x = AVC(#x) //可知av_connect==》SAVC(connect) #define AVC(str) {str,sizeof(str)-1}SAVC(app); SAVC(connect); SAVC(flashVer); SAVC(swfUrl); SAVC(pageUrl); SAVC(tcUrl); SAVC(fpad); SAVC(capabilities); SAVC(audioCodecs); SAVC(videoCodecs); SAVC(videoFunction); SAVC(objectEncoding); SAVC(secureToken); SAVC(secureTokenResponse); SAVC(type); SAVC(nonprivate); static int SendConnectPacket(RTMP *r, RTMPPacket *cp) {RTMPPacket packet;char pbuf[4096], *pend = pbuf + sizeof(pbuf);char *enc;if (cp)return RTMP_SendPacket(r, cp, TRUE);//塊流IDpacket.m_nChannel = 0x03; /* control channel (invoke) */packet.m_headerType = RTMP_PACKET_SIZE_LARGE;//消息類型為20的用AMF0編碼，這些消息用于在遠(yuǎn)端實現(xiàn)連接，創(chuàng)建流，發(fā)布，播放和暫停等操作packet.m_packetType = 0x14; /* INVOKE */packet.m_nTimeStamp = 0;//流ID需要設(shè)置為0packet.m_nInfoField2 = 0;packet.m_hasAbsTimestamp = 0;packet.m_body = pbuf + RTMP_MAX_HEADER_SIZE;enc = packet.m_body;//connect使用##進行字符串化連接，此處編碼connect字符串enc = AMF_EncodeString(enc, pend, &av_connect);enc = AMF_EncodeNumber(enc, pend, ++r->m_numInvokes);*enc++ = AMF_OBJECT;enc = AMF_EncodeNamedString(enc, pend, &av_app, &r->Link.app);if (!enc)return FALSE;if (r->Link.protocol & RTMP_FEATURE_WRITE){enc = AMF_EncodeNamedString(enc, pend, &av_type, &av_nonprivate);if (!enc)return FALSE;}if (r->Link.flashVer.av_len){enc = AMF_EncodeNamedString(enc, pend, &av_flashVer, &r->Link.flashVer);if (!enc)return FALSE;}if (r->Link.swfUrl.av_len){enc = AMF_EncodeNamedString(enc, pend, &av_swfUrl, &r->Link.swfUrl);if (!enc)return FALSE;}if (r->Link.tcUrl.av_len){enc = AMF_EncodeNamedString(enc, pend, &av_tcUrl, &r->Link.tcUrl);if (!enc)return FALSE;}if (!(r->Link.protocol & RTMP_FEATURE_WRITE)){enc = AMF_EncodeNamedBoolean(enc, pend, &av_fpad, FALSE);if (!enc)return FALSE;enc = AMF_EncodeNamedNumber(enc, pend, &av_capabilities, 15.0);if (!enc)return FALSE;enc = AMF_EncodeNamedNumber(enc, pend, &av_audioCodecs, r->m_fAudioCodecs);if (!enc)return FALSE;enc = AMF_EncodeNamedNumber(enc, pend, &av_videoCodecs, r->m_fVideoCodecs);if (!enc)return FALSE;enc = AMF_EncodeNamedNumber(enc, pend, &av_videoFunction, 1.0);if (!enc)return FALSE;if (r->Link.pageUrl.av_len){enc = AMF_EncodeNamedString(enc, pend, &av_pageUrl, &r->Link.pageUrl);if (!enc)return FALSE;}}if (r->m_fEncoding != 0.0 || r->m_bSendEncoding){ /* AMF0, AMF3 not fully supported yet */enc = AMF_EncodeNamedNumber(enc, pend, &av_objectEncoding, r->m_fEncoding);if (!enc)return FALSE;}if (enc + 3 >= pend)return FALSE;*enc++ = 0;*enc++ = 0; /* end of object - 0x00 0x00 0x09 */*enc++ = AMF_OBJECT_END;/* add auth string */if (r->Link.auth.av_len){enc = AMF_EncodeBoolean(enc, pend, r->Link.lFlags & RTMP_LF_AUTH);if (!enc)return FALSE;enc = AMF_EncodeString(enc, pend, &r->Link.auth);if (!enc)return FALSE;}if (r->Link.extras.o_num){int i;for (i = 0; i < r->Link.extras.o_num; i++){enc = AMFProp_Encode(&r->Link.extras.o_props[i], enc, pend);if (!enc)return FALSE;}}packet.m_nBodySize = enc - packet.m_body;return RTMP_SendPacket(r, &packet, TRUE); }RTMP_ConnectStream() 建立NetStream
RTMP_ClientPacket() 處理收到的消息
消息類型解析
int RTMP_ClientPacket(RTMP *r, RTMPPacket *packet) {int bHasMediaPacket = 0;switch (packet->m_packetType){/* 協(xié)議控制消息：RTMP 塊流使用類型為 1、2、3、5 和 6 的消息用于協(xié)議控制消息。這些消息包含 RTMP塊流協(xié)議需要的信息。這個協(xié)議控制消息必須（MUST）使用 ID 為 0 消息流并且用 ID 為 2 塊流發(fā)送。協(xié)議控制消息在接收到時盡快處理。*/case 0x01:/* chunk size */HandleChangeChunkSize(r, packet);break;case 0x03:/* bytes read report */RTMP_Log(RTMP_LOGDEBUG, "%s, received: bytes read report", __FUNCTION__);break;/*用戶控制消息：RTMP 使用類型 ID 為 4 的消息做為用戶控制消息，用戶控制消息應(yīng)該（SHOULD）消息流 ID 0（稱為控制流），并且通過 RTMP 塊流發(fā)送，使用塊流 ID 為 2。*/case 0x04:/* ctrl */HandleCtrl(r, packet);break;case 0x05:/* server bw */HandleServerBW(r, packet);break;case 0x06:/* client bw */HandleClientBW(r, packet);break;//類型為 8 的消息保留給音頻消息case 0x08:/* audio data *//*RTMP_Log(RTMP_LOGDEBUG, "%s, received: audio %lu bytes", __FUNCTION__, packet.m_nBodySize); */HandleAudio(r, packet);bHasMediaPacket = 1;if (!r->m_mediaChannel)r->m_mediaChannel = packet->m_nChannel;if (!r->m_pausing)r->m_mediaStamp = packet->m_nTimeStamp;break;//類型為 9 的消息保留給視頻消息case 0x09:/* video data *//*RTMP_Log(RTMP_LOGDEBUG, "%s, received: video %lu bytes", __FUNCTION__, packet.m_nBodySize); */HandleVideo(r, packet);bHasMediaPacket = 1;if (!r->m_mediaChannel)r->m_mediaChannel = packet->m_nChannel;if (!r->m_pausing)r->m_mediaStamp = packet->m_nTimeStamp;break;case 0x0F: /* flex stream send */RTMP_Log(RTMP_LOGDEBUG,"%s, flex stream send, size %lu bytes, not supported, ignoring",__FUNCTION__, packet->m_nBodySize);break;case 0x10: /* flex shared object */RTMP_Log(RTMP_LOGDEBUG,"%s, flex shared object, size %lu bytes, not supported, ignoring",__FUNCTION__, packet->m_nBodySize);break;case 0x11: /* flex message */{RTMP_Log(RTMP_LOGDEBUG,"%s, flex message, size %lu bytes, not fully supported",__FUNCTION__, packet->m_nBodySize);/*RTMP_LogHex(packet.m_body, packet.m_nBodySize); *//* some DEBUG code */ #if 0RTMP_LIB_AMFObject obj;int nRes = obj.Decode(packet.m_body+1, packet.m_nBodySize-1);if(nRes < 0) {RTMP_Log(RTMP_LOGERROR, "%s, error decoding AMF3 packet", __FUNCTION__);/*return; */}obj.Dump(); #endifif (HandleInvoke(r, packet->m_body + 1, packet->m_nBodySize - 1) == 1)bHasMediaPacket = 2;break;}/*元數(shù)據(jù)消息：客戶端用這個消息向?qū)Χ税l(fā)送 Metadata 或者任意的用戶數(shù)據(jù)。Metadata 包函了數(shù)據(jù)（音頻、視頻）的詳細(xì)信息，像創(chuàng)建時間，時長，主題等等。這些消息使用消息類型 18 表示 AMF0，用消息類型 15 來表示 AMF3。*/case 0x12:/* metadata (notify) */RTMP_Log(RTMP_LOGDEBUG, "%s, received: notify %lu bytes", __FUNCTION__,packet->m_nBodySize);if (HandleMetadata(r, packet->m_body, packet->m_nBodySize))bHasMediaPacket = 1;break;/*共享對象：共享對象是一個 Flash 對象（一個鍵值對的集合），用來同步多個客戶端，應(yīng)用實例等等。消息類型為 19 表示使用 AMF0，16 保留用作 AMF3 編碼共享事件。*/case 0x13:RTMP_Log(RTMP_LOGDEBUG, "%s, shared object, not supported, ignoring",__FUNCTION__);break;/*命令消息，消息類型為17或20)發(fā)送端發(fā)送時會帶有命令的名字，如connect，TransactionID表示此次命令的標(biāo)識，Command Object表示相關(guān)參數(shù)。*/case 0x14:/* invoke */RTMP_Log(RTMP_LOGDEBUG, "%s, received: invoke %lu bytes", __FUNCTION__,packet->m_nBodySize);/*RTMP_LogHex(packet.m_body, packet.m_nBodySize); */if (HandleInvoke(r, packet->m_body, packet->m_nBodySize) == 1)bHasMediaPacket = 2;break;//處理flv元數(shù)據(jù)case 0x16:{/* go through FLV packets and handle metadata packets */unsigned int pos = 0;uint32_t nTimeStamp = packet->m_nTimeStamp;while (pos + 11 < packet->m_nBodySize){//從flv頭部獲取數(shù)據(jù)長度uint32_t dataSize = AMF_DecodeInt24(packet->m_body + pos + 1); /* size without header (11) and prevTagSize (4) */if (pos + 11 + dataSize + 4 > packet->m_nBodySize){RTMP_Log(RTMP_LOGWARNING, "Stream corrupt?!");break;}//類型為18，處理元數(shù)據(jù)if (packet->m_body[pos] == 0x12){HandleMetadata(r, packet->m_body + pos + 11, dataSize);}//類型為8表示音頻數(shù)據(jù)，9表示視頻數(shù)據(jù)else if (packet->m_body[pos] == 8 || packet->m_body[pos] == 9){nTimeStamp = AMF_DecodeInt24(packet->m_body + pos + 4);nTimeStamp |= (packet->m_body[pos + 7] << 24);}pos += (11 + dataSize + 4);//指向下一個flv數(shù)據(jù)塊頭部}if (!r->m_pausing)r->m_mediaStamp = nTimeStamp;/* FLV tag(s) *//*RTMP_Log(RTMP_LOGDEBUG, "%s, received: FLV tag(s) %lu bytes", __FUNCTION__, packet.m_nBodySize); */bHasMediaPacket = 1;break;}default:RTMP_Log(RTMP_LOGDEBUG, "%s, unknown packet type received: 0x%02x", __FUNCTION__,packet->m_packetType); #ifdef _DEBUGRTMP_LogHex(RTMP_LOGDEBUG, packet->m_body, packet->m_nBodySize); #endif}return bHasMediaPacket; }RTMP_SendPacket() 消息流發(fā)送，內(nèi)含分塊處理
int RTMP_SendPacket(RTMP *r, RTMPPacket *packet, int queue) {/*獲取m_vecChannelsOut數(shù)組接收對應(yīng)塊流的包并保存待使用；#define RTMP_CHANNELS 65600本協(xié)議支持65597種塊流，ID從3-65599.ID 0,1,2作為保留。0表示ID范圍64-319（第二字節(jié)+64）1表示范圍是64-65599（第三字節(jié)*256 + 第二字節(jié)+64）2表示低層協(xié)議消息3-63#define RTMP_CHANNELS 65600RTMPPacket *m_vecChannelsIn[RTMP_CHANNELS];RTMPPacket *m_vecChannelsOut[RTMP_CHANNELS];*/const RTMPPacket *prevPacket = r->m_vecChannelsOut[packet->m_nChannel];uint32_t last = 0;//上一次相對時間戳int nSize;//實際頭部大小int hSize, cSize;//header頭指針指向頭部，hend塊尾指針指向body頭部//hbuf表示頭部最大18（3字節(jié)最大塊基本頭+11字節(jié)最大快消息頭+4字節(jié)擴展時間戳）緩沖數(shù)組char *header, *hptr, *hend, hbuf[RTMP_MAX_HEADER_SIZE], c;uint32_t t;//char *buffer, *tbuf = NULL, *toff = NULL;int nChunkSize;int tlen;//根據(jù)m_headerType值判斷消息類型（0,1,2,3）塊基本頭的高兩位if (prevPacket && packet->m_headerType != RTMP_PACKET_SIZE_LARGE){/* compress a bit by using the prev packet's attributes */if (prevPacket->m_nBodySize == packet->m_nBodySize&& prevPacket->m_packetType == packet->m_packetType&& packet->m_headerType == RTMP_PACKET_SIZE_MEDIUM)packet->m_headerType = RTMP_PACKET_SIZE_SMALL;//塊類型判斷為2if (prevPacket->m_nTimeStamp == packet->m_nTimeStamp&& packet->m_headerType == RTMP_PACKET_SIZE_SMALL)packet->m_headerType = RTMP_PACKET_SIZE_MINIMUM;//塊類型判斷為2last = prevPacket->m_nTimeStamp;}if (packet->m_headerType > 3) /* sanity */{RTMP_Log(RTMP_LOGERROR, "sanity failed!! trying to send header of type: 0x%02x.",(unsigned char)packet->m_headerType);return FALSE;}//static const int packetSize[] = { 12, 8, 4, 1 };塊消息頭加上一字節(jié)塊基本頭nSize = packetSize[packet->m_headerType];hSize = nSize; cSize = 0;t = packet->m_nTimeStamp - last;//相對時間if (packet->m_body){header = packet->m_body - nSize;hend = packet->m_body;}else //消息體為空{(diào)header = hbuf + 6;//header指向消息長度，最長塊基本頭3+塊消息頭3字節(jié)時間戳hend = hbuf + sizeof(hbuf);}//由塊流ID判斷塊類型if (packet->m_nChannel > 319)cSize = 2;//塊流id所占字節(jié)else if (packet->m_nChannel > 63)cSize = 1;if (cSize){header -= cSize;hSize += cSize;}//時間大于0xffffff增加4字節(jié)擴展時間戳大小if (nSize > 1 && t >= 0xffffff){header -= 4;hSize += 4;}hptr = header;c = packet->m_headerType << 6;//塊類型switch (cSize){case 0:c |= packet->m_nChannel;break;case 1:break;case 2:c |= 1;break;}*hptr++ = c;if (cSize){int tmp = packet->m_nChannel - 64;*hptr++ = tmp & 0xff;if (cSize == 2)*hptr++ = tmp >> 8;}if (nSize > 1){hptr = AMF_EncodeInt24(hptr, hend, t > 0xffffff ? 0xffffff : t);}if (nSize > 4){hptr = AMF_EncodeInt24(hptr, hend, packet->m_nBodySize);*hptr++ = packet->m_packetType;}if (nSize > 8)hptr += EncodeInt32LE(hptr, packet->m_nInfoField2);if (nSize > 1 && t >= 0xffffff)hptr = AMF_EncodeInt32(hptr, hend, t);// 到此為止 已經(jīng)將塊頭填寫好了;// 此時nSize表示負(fù)載數(shù)據(jù)的長度 buffer是指向負(fù)載數(shù)據(jù)區(qū)的指針;nSize = packet->m_nBodySize; //數(shù)據(jù)大小buffer = packet->m_body; //數(shù)據(jù)頭部指針nChunkSize = r->m_outChunkSize; //輸出塊大小，默認(rèn)是128字節(jié)RTMP_Log(RTMP_LOGDEBUG2, "%s: fd=%d, size=%d", __FUNCTION__, r->m_sb.sb_socket,nSize);/* send all chunks in one HTTP request */if (r->Link.protocol & RTMP_FEATURE_HTTP){int chunks = (nSize+nChunkSize-1) / nChunkSize;//分塊數(shù)if (chunks > 1){tlen = chunks * (cSize + 1) + nSize + hSize;tbuf = malloc(tlen);if (!tbuf)return FALSE;toff = tbuf;}}//while (nSize + hSize){int wrote;if (nSize < nChunkSize)//最后一個塊大小nChunkSize = nSize;RTMP_LogHexString(RTMP_LOGDEBUG2, (uint8_t *)header, hSize);RTMP_LogHexString(RTMP_LOGDEBUG2, (uint8_t *)buffer, nChunkSize);if (tbuf){// 將從Chunk頭開始的nChunkSize + hSize個字節(jié)拷貝至toff中;// 這些拷貝的數(shù)據(jù)包括塊頭數(shù)據(jù)(hSize字節(jié))和nChunkSize個負(fù)載數(shù)據(jù);memcpy(toff, header, nChunkSize + hSize);toff += nChunkSize + hSize;}else{wrote = WriteN(r, header, nChunkSize + hSize);if (!wrote)return FALSE;}nSize -= nChunkSize;buffer += nChunkSize;hSize = 0;if (nSize > 0){header = buffer - 1;hSize = 1;if (cSize){header -= cSize;hSize += cSize;}*header = (0xc0 | c);if (cSize){int tmp = packet->m_nChannel - 64;header[1] = tmp & 0xff;if (cSize == 2)header[2] = tmp >> 8;}}}if (tbuf){int wrote = WriteN(r, tbuf, toff-tbuf);free(tbuf);tbuf = NULL;if (!wrote)return FALSE;}/* we invoked a remote method */if (packet->m_packetType == 0x14)//消息流類型：0x14（20）{AVal method;char *ptr;ptr = packet->m_body + 1;AMF_DecodeString(ptr, &method);RTMP_Log(RTMP_LOGDEBUG, "Invoking %s", method.av_val);/* keep it in call queue till result arrives */if (queue) {int txn;ptr += 3 + method.av_len;txn = (int)AMF_DecodeNumber(ptr);AV_queue(&r->m_methodCalls, &r->m_numCalls, &method, txn);}}//m_vecChannelsOut數(shù)組添加傳入的packetif (!r->m_vecChannelsOut[packet->m_nChannel])r->m_vecChannelsOut[packet->m_nChannel] = malloc(sizeof(RTMPPacket));memcpy(r->m_vecChannelsOut[packet->m_nChannel], packet, sizeof(RTMPPacket));return TRUE; }RTMP_Close() 關(guān)閉連接
CleanupSockets() 清理Socket
inline void CleanupSockets() { #ifdef WIN32WSACleanup(); #endif }拉流
RTMP_ReadPacket() 讀取RTMPPacket包
int RTMP_ReadPacket(RTMP *r, RTMPPacket *packet) {uint8_t hbuf[RTMP_MAX_HEADER_SIZE] = { 0 };//按最大字節(jié)初始化char *header = (char *)hbuf;int nSize, hSize, nToRead, nChunk;int didAlloc = FALSE;RTMP_Log(RTMP_LOGDEBUG2, "%s: fd=%d", __FUNCTION__, r->m_sb.sb_socket);//解析塊基本頭(1-3字節(jié))if (ReadN(r, (char *)hbuf, 1) == 0){RTMP_Log(RTMP_LOGERROR, "%s, failed to read RTMP packet header", __FUNCTION__);return FALSE;}packet->m_headerType = (hbuf[0] & 0xc0) >> 6;//獲取塊類型//3-63 之間的值表示完整的流IDpacket->m_nChannel = (hbuf[0] & 0x3f);//獲取塊流id：0x3f首字節(jié)后6比特位header++;//ID 0、1作為保留：if (packet->m_nChannel == 0){//0，表示ID 的范圍是64-319（需要2個字節(jié)得出塊id：第二個字節(jié)+64）if (ReadN(r, (char *)&hbuf[1], 1) != 1){RTMP_Log(RTMP_LOGERROR, "%s, failed to read RTMP packet header 2nd byte",__FUNCTION__);return FALSE;}packet->m_nChannel = hbuf[1];packet->m_nChannel += 64;header++;}else if (packet->m_nChannel == 1){//1，表示ID 范圍是64-65599（需要3個字節(jié)得出塊id：第三個字節(jié)*256+第二個字節(jié)+64）int tmp;if (ReadN(r, (char *)&hbuf[1], 2) != 2){RTMP_Log(RTMP_LOGERROR, "%s, failed to read RTMP packet header 3nd byte",__FUNCTION__);return FALSE;}tmp = (hbuf[2] << 8) + hbuf[1];packet->m_nChannel = tmp + 64;RTMP_Log(RTMP_LOGDEBUG, "%s, m_nChannel: %0x", __FUNCTION__, packet->m_nChannel);header += 2;}//basic header 中的type頭字節(jié)，值為（0，1，2，3）表示ChunkMsgHeader的類型（11,7,3,0）字節(jié)長度//static const int packetSize[] = { 12, 8, 4, 1 };nSize = packetSize[packet->m_headerType];//獲取塊消息頭字節(jié)大小//判斷是否為絕對時間戳：對于0 類型的塊，消息的絕對時間戳在這里發(fā)送if (nSize == RTMP_LARGE_HEADER_SIZE) /* if we get a full header the timestamp is absolute */packet->m_hasAbsTimestamp = TRUE;else if (nSize < RTMP_LARGE_HEADER_SIZE){ /* using values from the last message of this channel */ //使用此通道上一條消息中的值//根據(jù)塊流id號從RTMPPacket *的m_vecChannelsIn數(shù)組中獲取目標(biāo)分塊（用塊流id對應(yīng)m_vecChannelsIn數(shù)組下標(biāo)）if (r->m_vecChannelsIn[packet->m_nChannel])memcpy(packet, r->m_vecChannelsIn[packet->m_nChannel],sizeof(RTMPPacket));}//塊消息頭值，(塊基本頭最少一個字節(jié))nSize--;//解析塊消息頭//塊基本頭大于1字節(jié)時，讀取塊消息頭，header指向消息頭地址if (nSize > 0 && ReadN(r, header, nSize) != nSize){RTMP_Log(RTMP_LOGERROR, "%s, failed to read RTMP packet header. type: %x",__FUNCTION__, (unsigned int)hbuf[0]);return FALSE;}//hSize大小指向數(shù)據(jù)段地址起始hSize = nSize + (header - (char *)hbuf);//非3類型if (nSize >= 3){//前3字節(jié)填充時間戳，塊消息頭前三字節(jié)packet->m_nTimeStamp = AMF_DecodeInt24(header);/*RTMP_Log(RTMP_LOGDEBUG, "%s, reading RTMP packet chunk on channel %x, headersz %i, timestamp %i, abs timestamp %i", __FUNCTION__, packet.m_nChannel, nSize, packet.m_nTimeStamp, packet.m_hasAbsTimestamp); */if (nSize >= 6){//第4-6字節(jié)填充塊消息長度packet->m_nBodySize = AMF_DecodeInt24(header + 3);packet->m_nBytesRead = 0;RTMPPacket_Free(packet);if (nSize > 6){//第7字節(jié)填充消息類型packet->m_packetType = header[6];//第8-11字節(jié)填充消息idif (nSize == 11)packet->m_nInfoField2 = DecodeInt32LE(header + 7);}}if (packet->m_nTimeStamp == 0xffffff)//時間戳滿，啟用擴展時間戳{if (ReadN(r, header + nSize, 4) != 4){RTMP_Log(RTMP_LOGERROR, "%s, failed to read extended timestamp",__FUNCTION__);return FALSE;}packet->m_nTimeStamp = AMF_DecodeInt32(header + nSize);hSize += 4;}}RTMP_LogHexString(RTMP_LOGDEBUG2, (uint8_t *)hbuf, hSize);//分配數(shù)據(jù)空間if (packet->m_nBodySize > 0 && packet->m_body == NULL){if (!RTMPPacket_Alloc(packet, packet->m_nBodySize)){RTMP_Log(RTMP_LOGDEBUG, "%s, failed to allocate packet", __FUNCTION__);return FALSE;}didAlloc = TRUE;packet->m_headerType = (hbuf[0] & 0xc0) >> 6;}//將要讀取的數(shù)據(jù)nToRead = packet->m_nBodySize - packet->m_nBytesRead;//rtmp流輸入塊大小nChunk = r->m_inChunkSize;// 剩下的消息數(shù)據(jù)長度如果比塊尺寸大，則需要分塊,否則塊尺寸就等于剩下的消息數(shù)據(jù)長度if (nToRead < nChunk)nChunk = nToRead;/* Does the caller want the raw chunk? *///根據(jù)解析的數(shù)據(jù)重新封裝一個RTMPChunk *對象，即獲得raw chunkif (packet->m_chunk){packet->m_chunk->c_headerSize = hSize;memcpy(packet->m_chunk->c_header, hbuf, hSize);packet->m_chunk->c_chunk = packet->m_body + packet->m_nBytesRead;packet->m_chunk->c_chunkSize = nChunk;}if (ReadN(r, packet->m_body + packet->m_nBytesRead, nChunk) != nChunk){RTMP_Log(RTMP_LOGERROR, "%s, failed to read RTMP packet body. len: %lu",__FUNCTION__, packet->m_nBodySize);return FALSE;}RTMP_LogHexString(RTMP_LOGDEBUG2, (uint8_t *)packet->m_body + packet->m_nBytesRead, nChunk);// 更新已讀數(shù)據(jù)字節(jié)個數(shù)packet->m_nBytesRead += nChunk;/* keep the packet as ref for other packets on this channel */if (!r->m_vecChannelsIn[packet->m_nChannel])r->m_vecChannelsIn[packet->m_nChannel] = malloc(sizeof(RTMPPacket));memcpy(r->m_vecChannelsIn[packet->m_nChannel], packet, sizeof(RTMPPacket));//數(shù)據(jù)準(zhǔn)備完畢，清除r內(nèi)存if (RTMPPacket_IsReady(packet)){/* make packet's timestamp absolute *//*絕對時間戳 = 上一次絕對時間戳 + 時間戳增量 */if (!packet->m_hasAbsTimestamp)packet->m_nTimeStamp += r->m_channelTimestamp[packet->m_nChannel]; /* timestamps seem to be always relative!! */// 當(dāng)前絕對時間戳保存起來，供下一個包轉(zhuǎn)換時間戳使用r->m_channelTimestamp[packet->m_nChannel] = packet->m_nTimeStamp;/* reset the data from the stored packet. we keep the header since we may use it later if a new packet for this channel *//* arrives and requests to re-use some info (small packet header) *///重置保存的包。保留塊頭數(shù)據(jù)，因為通道中新到來的包(更短的塊頭)可能需要使用前面塊頭的信息.r->m_vecChannelsIn[packet->m_nChannel]->m_body = NULL;r->m_vecChannelsIn[packet->m_nChannel]->m_nBytesRead = 0;r->m_vecChannelsIn[packet->m_nChannel]->m_hasAbsTimestamp = FALSE; /* can only be false if we reuse header */}else{packet->m_body = NULL; /* so it won't be erased on free */}return TRUE; }Download() 下載函數(shù)
int Download(RTMP * rtmp, // connected RTMP objectFILE * file, uint32_t dSeek, uint32_t dStopOffset, double duration, int bResume, char *metaHeader, uint32_t nMetaHeaderSize, char *initialFrame, int initialFrameType, uint32_t nInitialFrameSize, int nSkipKeyFrames, int bStdoutMode, int bLiveStream, int bHashes, int bOverrideBufferTime, uint32_t bufferTime, double *percent) // percentage downloaded [out] {int32_t now, lastUpdate;int bufferSize = 64 * 1024;char *buffer = (char *) malloc(bufferSize);int nRead = 0;//long ftell(FILE *stream);//返回當(dāng)前文件指針RTMP_LogPrintf("開始下載！\n");off_t size = ftello(file);unsigned long lastPercent = 0;//時間戳rtmp->m_read.timestamp = dSeek;*percent = 0.0;if (rtmp->m_read.timestamp){RTMP_Log(RTMP_LOGDEBUG, "Continuing at TS: %d ms\n", rtmp->m_read.timestamp);}//是直播if (bLiveStream){RTMP_LogPrintf("直播流\n");}else{// print initial status// Workaround to exit with 0 if the file is fully (> 99.9%) downloadedif (duration > 0){if ((double) rtmp->m_read.timestamp >= (double) duration * 999.0){RTMP_LogPrintf("Already Completed at: %.3f sec Duration=%.3f sec\n",(double) rtmp->m_read.timestamp / 1000.0,(double) duration / 1000.0);return RD_SUCCESS;}else{*percent = ((double) rtmp->m_read.timestamp) / (duration * 1000.0) * 100.0;*percent = ((double) (int) (*percent * 10.0)) / 10.0;RTMP_LogPrintf("%s download at: %.3f kB / %.3f sec (%.1f%%)\n",bResume ? "Resuming" : "Starting",(double) size / 1024.0, (double) rtmp->m_read.timestamp / 1000.0,*percent);}}else{RTMP_LogPrintf("%s download at: %.3f kB\n",bResume ? "Resuming" : "Starting",(double) size / 1024.0);}}if (dStopOffset > 0)RTMP_LogPrintf("For duration: %.3f sec\n", (double) (dStopOffset - dSeek) / 1000.0);//各種設(shè)置參數(shù)到rtmp連接if (bResume && nInitialFrameSize > 0)rtmp->m_read.flags |= RTMP_READ_RESUME;rtmp->m_read.initialFrameType = initialFrameType;rtmp->m_read.nResumeTS = dSeek;rtmp->m_read.metaHeader = metaHeader;rtmp->m_read.initialFrame = initialFrame;rtmp->m_read.nMetaHeaderSize = nMetaHeaderSize;rtmp->m_read.nInitialFrameSize = nInitialFrameSize;now = RTMP_GetTime();lastUpdate = now - 1000;do{//從rtmp中把bufferSize（64k）個數(shù)據(jù)讀入buffernRead = RTMP_Read(rtmp, buffer, bufferSize);//RTMP_LogPrintf("nRead: %d\n", nRead);if (nRead > 0){//函數(shù)：size_t fwrite(const void* buffer,size_t size,size_t count,FILE* stream);//向文件讀入寫入一個數(shù)據(jù)塊。返回值：返回實際寫入的數(shù)據(jù)塊數(shù)目//（1）buffer：是一個指針，對fwrite來說，是要輸出數(shù)據(jù)的地址。//（2）size：要寫入內(nèi)容的單字節(jié)數(shù)； //（3）count:要進行寫入size字節(jié)的數(shù)據(jù)項的個數(shù)； //（4）stream:目標(biāo)文件指針。 //（5）返回實際寫入的數(shù)據(jù)項個數(shù)count。//關(guān)鍵。把buffer里面的數(shù)據(jù)寫成文件if (fwrite(buffer, sizeof(unsigned char), nRead, file) !=(size_t) nRead){RTMP_Log(RTMP_LOGERROR, "%s: Failed writing, exiting!", __FUNCTION__);free(buffer);return RD_FAILED;}//記錄已經(jīng)寫入的字節(jié)數(shù)size += nRead;//RTMP_LogPrintf("write %dbytes (%.1f kB)\n", nRead, nRead/1024.0);if (duration <= 0) // if duration unknown try to get it from the stream (onMetaData)duration = RTMP_GetDuration(rtmp);if (duration > 0){// make sure we claim to have enough buffer time!if (!bOverrideBufferTime && bufferTime < (duration * 1000.0)){bufferTime = (uint32_t) (duration * 1000.0) + 5000; // 再加5s以確保buffertime足夠長RTMP_Log(RTMP_LOGDEBUG,"Detected that buffer time is less than duration, resetting to: %dms",bufferTime);//重設(shè)Buffer長度RTMP_SetBufferMS(rtmp, bufferTime);//給服務(wù)器發(fā)送UserControl消息通知Buffer改變RTMP_UpdateBufferMS(rtmp);}//計算百分比*percent = ((double) rtmp->m_read.timestamp) / (duration * 1000.0) * 100.0;*percent = ((double) (int) (*percent * 10.0)) / 10.0;if (bHashes){if (lastPercent + 1 <= *percent){RTMP_LogStatus("#");lastPercent = (unsigned long) *percent;}}else{//設(shè)置顯示數(shù)據(jù)的更新間隔200msnow = RTMP_GetTime();if (abs(now - lastUpdate) > 200){RTMP_LogStatus("\r%.3f kB / %.2f sec (%.1f%%)",(double) size / 1024.0,(double) (rtmp->m_read.timestamp) / 1000.0, *percent);lastUpdate = now;}}}else{//現(xiàn)在距離開機的毫秒數(shù)now = RTMP_GetTime();//每間隔200ms刷新一次數(shù)據(jù)if (abs(now - lastUpdate) > 200){if (bHashes)RTMP_LogStatus("#");else//size為已寫入文件的字節(jié)數(shù)RTMP_LogStatus("\r%.3f kB / %.2f sec", (double) size / 1024.0,(double) (rtmp->m_read.timestamp) / 1000.0);lastUpdate = now;}}} #ifdef _DEBUGelse{RTMP_Log(RTMP_LOGDEBUG, "zero read!");} #endif}while (!RTMP_ctrlC && nRead > -1 && RTMP_IsConnected(rtmp) && !RTMP_IsTimedout(rtmp));free(buffer);if (nRead < 0)//nRead是讀取情況nRead = rtmp->m_read.status;/* Final status update */if (!bHashes){if (duration > 0){*percent = ((double) rtmp->m_read.timestamp) / (duration * 1000.0) * 100.0;*percent = ((double) (int) (*percent * 10.0)) / 10.0;//輸出RTMP_LogStatus("\r%.3f kB / %.2f sec (%.1f%%)",(double) size / 1024.0,(double) (rtmp->m_read.timestamp) / 1000.0, *percent);}else{RTMP_LogStatus("\r%.3f kB / %.2f sec", (double) size / 1024.0,(double) (rtmp->m_read.timestamp) / 1000.0);}}RTMP_Log(RTMP_LOGDEBUG, "RTMP_Read returned: %d", nRead);//讀取錯誤if (bResume && nRead == -2){RTMP_LogPrintf("Couldn't resume FLV file, try --skip %d\n\n",nSkipKeyFrames + 1);return RD_FAILED;}//讀取正確if (nRead == -3)return RD_SUCCESS;//沒讀完...if ((duration > 0 && *percent < 99.9) || RTMP_ctrlC || nRead < 0|| RTMP_IsTimedout(rtmp)){return RD_INCOMPLETE;}return RD_SUCCESS; }參考文獻
總結(jié)
以上是生活随笔為你收集整理的流媒体-RTMP协议-librtmp库学习（二）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 利用ExcelJS读取Excel文件
- 下一篇： 计算机视觉有哪些商业用途