當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

流媒体-RTMP协议-librtmp库学习（二）

發布時間：2023/12/31 编程问答 33 豆豆

生活随笔收集整理的這篇文章主要介紹了流媒体-RTMP协议-librtmp库学习（二）小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

流媒體-RTMP協議-rtmpdump-flv封裝解析（一）

流媒體-RTMP協議-librtmp庫學習（二）

流媒體-RTMP協議-librtmp庫學習-c++多線程實現rtmp推流flv文件（三）

文章目錄

- librtmp 庫相關結構體
- - RTMP 結構體
  - RTMP_LNK 結構體
  - RTMPPacket 結構體：描述實時消息協議的分塊
- AMF(Action Message Format): 動作消息格式
- - AVal結構體：自定義字符串
- 推流
- - InitSockets() 初始化Socket
  - RTMP_Alloc() 用于創建一個RTMP的結構體
  - RTMP_Init() 初始化結構體
  - RTMP_SetupURL() 設置推拉流的URL
  - - RTMP_ParseURL() 解析輸入URL
  - RTMP_EnableWrite() 設置為推流狀態
  - RTMP_Connect() 建立NetConnection
  - - RTMP_Connect0() 第0次連接，建立socket連接
    - RTMP_Connect1() 第1次連接，建立真正的rtmp連接
    - - HandShake() 握手過程實現(C0+C1， S0+S1+S2， C2)
      - SendConnectPacket() 發送命令請求
  - RTMP_ConnectStream() 建立NetStream
  - - RTMP_ClientPacket() 處理收到的消息
  - RTMP_SendPacket() 消息流發送，內含分塊處理
  - RTMP_Close() 關閉連接
  - CleanupSockets() 清理Socket
- 拉流
- - RTMP_ReadPacket() 讀取RTMPPacket包
  - Download() 下載函數
- 參考文獻

librtmp 庫相關結構體

RTMP 結構體

表示一個rtmp連接

typedef struct RTMP{int m_inChunkSize; //拉流流分塊大小：初始化默認128字節 RTMP_DEFAULT_CHUNKSIZE 128int m_outChunkSize; //推流分塊大小：初始化默認128字節 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 結構體

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 結構體：描述實時消息協議的分塊

//塊由頭和數據組成，原始的rtmp消息塊表示其中的數據都是沒有經過解析的，是原始的字節流#define RTMP_MAX_HEADER_SIZE 18typedef struct RTMPChunk{int c_headerSize; //頭部的長度int c_chunkSize; //chunk的大小char *c_chunk; //數據char c_header[RTMP_MAX_HEADER_SIZE]; //chunk頭部} RTMPChunk;typedef struct RTMPPacket{uint8_t m_headerType; //basic header 中的type頭字節，值為(0,1,2,3)表示ChunkMsgHeader的類型（4種）uint8_t m_packetType; //Chunk Msg Header中msg type 1字節：消息類型id（8: audio；9:video；18:AMF0編碼的元數據）uint8_t m_hasAbsTimestamp; //bool值，是否是絕對時間戳(類型1時為true)int m_nChannel; //塊流ID ，通過設置ChannelID來設置Basic stream id的長度和值uint32_t m_nTimeStamp; //時間戳，消息頭前三字節int32_t m_nInfoField2; //Chunk Msg Header中msg StreamID 4字節：消息流iduint32_t m_nBodySize; //Chunk Msg Header中msg length 4字節：消息長度uint32_t m_nBytesRead; //已讀取的數據RTMPChunk *m_chunk; //raw chunk結構體指針，把RTMPPacket的真實頭部和數據段拷貝進來char *m_body; //數據段指針} 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): 動作消息格式

它是一種二進制的數據格式，++AMF數據采用Big-Endian（大端模式，高字節存儲在內存的低地址端）++。它的設計是為了把actionscript里面的數據(包括Object,Array,Boolean,Number等)序列化成二進制數據，然后把這段數據隨意發送給其他接收方程序，比如發給遠程的服務器，在遠程服務器那邊，可以把這段數據給還原出來，以此達到一個數據傳輸的作用。

AVal結構體：自定義字符串

// 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() 用于創建一個RTMP的結構體

RTMP* RTMP_Alloc() {return calloc(1, sizeof(RTMP)); }

RTMP_Init() 初始化結構體

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() 設置推拉流的URL

內部調用了RTMP_ParseURL主要用于解析url

RTMP_ParseURL() 解析輸入URL

RTMP_EnableWrite() 設置為推流狀態

//是否推流，默認不推流 void RTMP_EnableWrite(RTMP *r) {r->Link.protocol |= RTMP_FEATURE_WRITE; }

RTMP_Connect() 建立NetConnection

主要分為RTMP_Connect0()函數(第0次連接，建立socket連接)+ RTMP_Connect1函數(第1次連接，建立真正的rtmp連接。主要分為HandShake握手、SendConnectPacket發送命令請求)

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發送命令請求)

HandShake() 握手過程實現(C0+C1， S0+S1+S2， C2)

C0 和 S0消息格式：C0和S0是單獨的一個字節，表示版本信息

在C0中這個字段表示客戶端要求的RTMP版本。在S0中這個字段表示服務器選擇的RTMP版本。本規范所定義的版本是3；0-2是早期產品所用的，已被丟棄；4-31保留在未來使用；32-255不允許使用（為了區分其他以某一字符開始的文本協議）。如果服務無法識別客戶端請求的版本，應該返回3 ?？蛻舳丝梢赃x擇減到版本3或選擇取消握手

C1 和 S1消息格式：C1和S1消息有1536字節長

時間：4字節：本字段包含時間戳。該時間戳應該是發送這個數據塊的端點的后續塊的時間起始點。可以是0，或其他的任何值。為了同步多個流，端點可能發送其塊流的當前值。
零：4字節：本字段必須是全零。
隨機數據：1528字節。本字段可以包含任何值。因為每個端點必須用自己初始化的握手和對端初始化的握手來區分身份，所以這個數據應有充分的隨機性。但是并不需要加密安全的隨機值，或者動態值

C2 和 S2 消息格式：C2和S2消息有1536字節長，只是S1和C1的回復

時間：4字節：本字段必須包含對等段發送的時間（對C2來說是S1，對S2來說是C1）。
時間2：4字節：本字段必須包含先前發送的并被對端讀取的包的時間戳。
隨機回復：1528字節：本字段必須包含對端發送的隨機數據字段（對C2來說是S1，對S2來說是C1）。每個對等端可以用時間和時間2字段中的時間戳來快速地估計帶寬和延遲。但這樣做可能并不實用。

//不使用加密 #ifndef CRYPTO static int HandShake(RTMP *r, int FP9HandShake) {int i;uint32_t uptime, suptime;int bMatch;char type;//#define RTMP_SIG_SIZE 1536:C1和S1消息有1536字節長char clientbuf[RTMP_SIG_SIZE + 1], *clientsig = clientbuf + 1;//C0和C1同時發送，因此多一字節C0char serversig[RTMP_SIG_SIZE];//0x03代表RTMP協議的版本（客戶端要求的）clientbuf[0] = 0x03; /* not encrypted *///當前時間，填充C1前4字節uptime = htonl(RTMP_GetTime());memcpy(clientsig, &uptime, 4);//4字節0填充memset(&clientsig[4], 0, 4);#ifdef _DEBUGfor (i = 8; i < RTMP_SIG_SIZE; i++)clientsig[i] = 0xff; #else//隨機數填充for (i = 8; i < RTMP_SIG_SIZE; i++)clientsig[i] = (char)(rand() % 256); #endif//同時發送C0+C1if (!WriteN(r, clientbuf, RTMP_SIG_SIZE + 1))return FALSE;//讀取1字節服務端返回值S0填充到typeif (ReadN(r, &type, 1) != 1) /* 0x03 or 0x06 */return FALSE;RTMP_Log(RTMP_LOGDEBUG, "%s: Type Answer : %02X", __FUNCTION__, type);if (type != clientbuf[0])RTMP_Log(RTMP_LOGWARNING, "%s: Type mismatch: client sent %d, server answered %d",__FUNCTION__, clientbuf[0], type);//讀取1536字節的S1到serversigif (ReadN(r, serversig, RTMP_SIG_SIZE) != RTMP_SIG_SIZE)return FALSE;/* decode server response *///獲取服務端時間memcpy(&suptime, serversig, 4);suptime = ntohl(suptime);RTMP_Log(RTMP_LOGDEBUG, "%s: Server Uptime : %d", __FUNCTION__, suptime);RTMP_Log(RTMP_LOGDEBUG, "%s: FMS Version : %d.%d.%d.%d", __FUNCTION__,serversig[4], serversig[5], serversig[6], serversig[7]);/* 2nd part of handshake */if (!WriteN(r, serversig, RTMP_SIG_SIZE))return FALSE;if (ReadN(r, serversig, RTMP_SIG_SIZE) != RTMP_SIG_SIZE)return FALSE;bMatch = (memcmp(serversig, clientsig, RTMP_SIG_SIZE) == 0);if (!bMatch){RTMP_Log(RTMP_LOGWARNING, "%s, client signature does not match!", __FUNCTION__);}return TRUE; }

SendConnectPacket() 發送命令請求

發送"connect"等各種字符串到服務端

#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編碼，這些消息用于在遠端實現連接，創建流，發布，播放和暫停等操作packet.m_packetType = 0x14; /* INVOKE */packet.m_nTimeStamp = 0;//流ID需要設置為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){/* 協議控制消息：RTMP 塊流使用類型為 1、2、3、5 和 6 的消息用于協議控制消息。這些消息包含 RTMP塊流協議需要的信息。這個協議控制消息必須（MUST）使用 ID 為 0 消息流并且用 ID 為 2 塊流發送。協議控制消息在接收到時盡快處理。*/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 的消息做為用戶控制消息，用戶控制消息應該（SHOULD）消息流 ID 0（稱為控制流），并且通過 RTMP 塊流發送，使用塊流 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;}/*元數據消息：客戶端用這個消息向對端發送 Metadata 或者任意的用戶數據。Metadata 包函了數據（音頻、視頻）的詳細信息，像創建時間，時長，主題等等。這些消息使用消息類型 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 對象（一個鍵值對的集合），用來同步多個客戶端，應用實例等等。消息類型為 19 表示使用 AMF0，16 保留用作 AMF3 編碼共享事件。*/case 0x13:RTMP_Log(RTMP_LOGDEBUG, "%s, shared object, not supported, ignoring",__FUNCTION__);break;/*命令消息，消息類型為17或20)發送端發送時會帶有命令的名字，如connect，TransactionID表示此次命令的標識，Command Object表示相關參數。*/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元數據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頭部獲取數據長度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，處理元數據if (packet->m_body[pos] == 0x12){HandleMetadata(r, packet->m_body + pos + 11, dataSize);}//類型為8表示音頻數據，9表示視頻數據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數據塊頭部}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() 消息流發送，內含分塊處理

int RTMP_SendPacket(RTMP *r, RTMPPacket *packet, int queue) {/*獲取m_vecChannelsOut數組接收對應塊流的包并保存待使用；#define RTMP_CHANNELS 65600本協議支持65597種塊流，ID從3-65599.ID 0,1,2作為保留。0表示ID范圍64-319（第二字節+64）1表示范圍是64-65599（第三字節*256 + 第二字節+64）2表示低層協議消息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字節最大塊基本頭+11字節最大快消息頭+4字節擴展時間戳）緩沖數組char *header, *hptr, *hend, hbuf[RTMP_MAX_HEADER_SIZE], c;uint32_t t;//char *buffer, *tbuf = NULL, *toff = NULL;int nChunkSize;int tlen;//根據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 };塊消息頭加上一字節塊基本頭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 //消息體為空{header = hbuf + 6;//header指向消息長度，最長塊基本頭3+塊消息頭3字節時間戳hend = hbuf + sizeof(hbuf);}//由塊流ID判斷塊類型if (packet->m_nChannel > 319)cSize = 2;//塊流id所占字節else if (packet->m_nChannel > 63)cSize = 1;if (cSize){header -= cSize;hSize += cSize;}//時間大于0xffffff增加4字節擴展時間戳大小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);// 到此為止已經將塊頭填寫好了;// 此時nSize表示負載數據的長度 buffer是指向負載數據區的指針;nSize = packet->m_nBodySize; //數據大小buffer = packet->m_body; //數據頭部指針nChunkSize = r->m_outChunkSize; //輸出塊大小，默認是128字節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;//分塊數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個字節拷貝至toff中;// 這些拷貝的數據包括塊頭數據(hSize字節)和nChunkSize個負載數據;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數組添加傳入的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() 關閉連接

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 };//按最大字節初始化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字節)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首字節后6比特位header++;//ID 0、1作為保留：if (packet->m_nChannel == 0){//0，表示ID 的范圍是64-319（需要2個字節得出塊id：第二個字節+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個字節得出塊id：第三個字節*256+第二個字節+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頭字節，值為（0，1，2，3）表示ChunkMsgHeader的類型（11,7,3,0）字節長度//static const int packetSize[] = { 12, 8, 4, 1 };nSize = packetSize[packet->m_headerType];//獲取塊消息頭字節大小//判斷是否為絕對時間戳：對于0 類型的塊，消息的絕對時間戳在這里發送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 */ //使用此通道上一條消息中的值//根據塊流id號從RTMPPacket *的m_vecChannelsIn數組中獲取目標分塊（用塊流id對應m_vecChannelsIn數組下標）if (r->m_vecChannelsIn[packet->m_nChannel])memcpy(packet, r->m_vecChannelsIn[packet->m_nChannel],sizeof(RTMPPacket));}//塊消息頭值，(塊基本頭最少一個字節)nSize--;//解析塊消息頭//塊基本頭大于1字節時，讀取塊消息頭，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大小指向數據段地址起始hSize = nSize + (header - (char *)hbuf);//非3類型if (nSize >= 3){//前3字節填充時間戳，塊消息頭前三字節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字節填充塊消息長度packet->m_nBodySize = AMF_DecodeInt24(header + 3);packet->m_nBytesRead = 0;RTMPPacket_Free(packet);if (nSize > 6){//第7字節填充消息類型packet->m_packetType = header[6];//第8-11字節填充消息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);//分配數據空間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;}//將要讀取的數據nToRead = packet->m_nBodySize - packet->m_nBytesRead;//rtmp流輸入塊大小nChunk = r->m_inChunkSize;// 剩下的消息數據長度如果比塊尺寸大，則需要分塊,否則塊尺寸就等于剩下的消息數據長度if (nToRead < nChunk)nChunk = nToRead;/* Does the caller want the raw chunk? *///根據解析的數據重新封裝一個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);// 更新已讀數據字節個數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));//數據準備完畢，清除r內存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!! */// 當前絕對時間戳保存起來，供下一個包轉換時間戳使用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) *///重置保存的包。保留塊頭數據，因為通道中新到來的包(更短的塊頭)可能需要使用前面塊頭的信息.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() 下載函數

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);//返回當前文件指針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);//各種設置參數到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）個數據讀入buffernRead = RTMP_Read(rtmp, buffer, bufferSize);//RTMP_LogPrintf("nRead: %d\n", nRead);if (nRead > 0){//函數：size_t fwrite(const void* buffer,size_t size,size_t count,FILE* stream);//向文件讀入寫入一個數據塊。返回值：返回實際寫入的數據塊數目//（1）buffer：是一個指針，對fwrite來說，是要輸出數據的地址。//（2）size：要寫入內容的單字節數；　　//（3）count:要進行寫入size字節的數據項的個數；　　//（4）stream:目標文件指針。　　//（5）返回實際寫入的數據項個數count。//關鍵。把buffer里面的數據寫成文件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;}//記錄已經寫入的字節數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);//重設Buffer長度RTMP_SetBufferMS(rtmp, bufferTime);//給服務器發送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{//設置顯示數據的更新間隔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{//現在距離開機的毫秒數now = RTMP_GetTime();//每間隔200ms刷新一次數據if (abs(now - lastUpdate) > 200){if (bHashes)RTMP_LogStatus("#");else//size為已寫入文件的字節數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; }

參考文獻

【原】librtmp源碼詳解

雷霄驊-libRTMP

總結

以上是生活随笔為你收集整理的流媒体-RTMP协议-librtmp库学习（二）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：利用ExcelJS读取Excel文件
下一篇：计算机视觉有哪些商业用途