librtmp实时消息传输协议(RTMP)库代码浅析

1

RTMP_Init(&rtmp);

//初始化RTMP参数

2

//指定了-i 参数，直接设置URL

3

if

(RTMP_SetupURL(&rtmp, fullUrl.av_val) == FALSE) {

4

RTMP_Log(RTMP_LOGERROR,

"Couldn't parse URL: %s"

, fullUrl.av_val);

5

return

RD_FAILED;

6

}

7

rtmp.Link.timeout = timeout ;

8

/* Try to keep the stream moving if it pauses on us */

9

if

(!bLiveStream )

10

rtmp.Link.lFlags |= RTMP_LF_BUFX;

11

while

(!RTMP_ctrlC)

12

{

13

RTMP_Log(RTMP_LOGDEBUG,

"Setting buffer time to: %dms"

, DEF_BUFTIME);

14

RTMP_SetBufferMS(&rtmp, DEF_BUFTIME);

//告诉服务器帮我缓存多久

15

RTMP_LogPrintf(

"Connecting ...\n"

);

16

if

(!RTMP_Connect(&rtmp, NULL)) {

//建立连接,发送"connect"

17

nStatus = RD_NO_CONNECT;

18

break

;

19

}

20

RTMP_Log(RTMP_LOGINFO,

"Connected..."

);

21

//处理服务端返回的各种控制消息包，比如收到connect的result后就进行createStream，以及发送play(test)消息

22

if

(!RTMP_ConnectStream(&rtmp, 0)) {

//一旦返回，表示服务端开始发送数据了.可以play了

23

nStatus = RD_FAILED;

24

break

;

25

}

26

nStatus = Download(&rtmp, file, bStdoutMode, bLiveStream );

27

if

(nStatus != RD_INCOMPLETE || !RTMP_IsTimedout(&rtmp) || bLiveStream)

28

break

;

29

}

1

int

RTMP_Connect(RTMP *r, RTMPPacket *cp)

2

{

//连接对端，进行握手，并且发送"connect" 控制消息，附带一些app,tcurl等参数

3

struct

sockaddr_in service;

4

if

(!r->Link.hostname.av_len)

5

return

FALSE;

6

7

memset

(&service, 0,

sizeof

(

struct

sockaddr_in));

8

service.sin_family = AF_INET;

9

10

if

(r->Link.socksport)

11

{

12

/* Connect via SOCKS */

13

if

(!add_addr_info(&service, &r->Link.sockshost, r->Link.socksport))

14

return

FALSE;

15

}

16

else

17

{

18

/* Connect directly */

19

if

(!add_addr_info(&service, &r->Link.hostname, r->Link.port))

20

return

FALSE;

21

}

22

23

if

(!RTMP_Connect0(r, (

struct

sockaddr *)&service))

//建立一个socket连接

24

return

FALSE;

25

26

r->m_bSendCounter = TRUE;

27

28

return

RTMP_Connect1(r, cp);

//进行C0-2/S0-2协议握手，发送connect命令

29

}

1

static

int

HandShake(RTMP *r,

int

FP9HandShake)

2

{

//C0,C1 -- S0, S1, S2 -- C2 消息握手协议

3

int

i;

4

uint32_t uptime, suptime;

5

int

bMatch;

6

char

type;

7

char

clientbuf[RTMP_SIG_SIZE + 1], *clientsig = clientbuf + 1;

8

char

serversig[RTMP_SIG_SIZE];

9

10

clientbuf[0] = 0x03;

//C0, 一个字节。03代表协议版本号为3 /* not encrypted */

11

12

uptime = htonl(RTMP_GetTime());

//这是一个时间戳，放在C1消息头部

13

memcpy

(clientsig, &uptime, 4);

14

memset

(&clientsig[4], 0, 4);

//后面放4个字节的空数据然后就是随机数据

15

16

//后面是随机数据，总共1536字节的C0消息

17

#ifdef _DEBUG

18

for

(i = 8; i < RTMP_SIG_SIZE; i++)

19

clientsig[i] = 0xff;

20

#else

21

for

(i = 8; i < RTMP_SIG_SIZE; i++)

22

clientsig[i] = (

char

)(

rand

() % 256);

//发送C0，
C1消息

23

#endif

24

25

if

(!WriteN(r, clientbuf, RTMP_SIG_SIZE + 1))

26

return

FALSE;

27

28

//下面读一个字节也就是S0消息，看协议是否一样

29

if

(ReadN(r, &type, 1) != 1)

/* 0x03 or 0x06 */

30

return

FALSE;

31

RTMP_Log(RTMP_LOGDEBUG,

"%s: Type Answer   : %02X"

, __FUNCTION__, type);

32

if

(type != clientbuf[0])

//C/S版本不一致

33

RTMP_Log(RTMP_LOGWARNING,

"%s: Type mismatch: client sent %d, server answered %d"

,  __FUNCTION__, clientbuf[0], type);

34

35

//读取S1消息，里面有服务器运行时间

36

if

(ReadN(r, serversig, RTMP_SIG_SIZE) != RTMP_SIG_SIZE)

37

return

FALSE;

38

39

/* decode server response */

40

41

memcpy

(&suptime, serversig, 4);

42

suptime = ntohl(suptime);

43

44

RTMP_Log(RTMP_LOGDEBUG,

"%s: Server Uptime : %d"

, __FUNCTION__, suptime);

45

RTMP_Log(RTMP_LOGDEBUG,

"%s: FMS Version   : %d.%d.%d.%d"

, __FUNCTION__, serversig[4], serversig[5], serversig[6], serversig[7]);

46

47

/* 2nd part of handshake */

48

if

(!WriteN(r, serversig, RTMP_SIG_SIZE))

//发送C2消息，内容就等于S1消息的内容。

49

return

FALSE;

50

51

//读取S2消息

52

if

(ReadN(r, serversig, RTMP_SIG_SIZE) != RTMP_SIG_SIZE)

53

return

FALSE;

54

55

bMatch = (

memcmp

(serversig, clientsig, RTMP_SIG_SIZE) == 0);

56

if

(!bMatch)

//服务端返回的S2消息必须跟C1消息一致才行

57

{

58

RTMP_Log(RTMP_LOGWARNING,

"%s, client signature does not match!"

, __FUNCTION__);

59

}

60

return

TRUE;

61

}

1

int

RTMP_ConnectStream(RTMP *r,

int

seekTime)

2

{

//循环读取服务端发送过来的各种消息，比如window ack**, set peer bandwidth, set chunk size, _result等

3

//直到接收到了play

4

RTMPPacket packet = {0 };

5

6

/* seekTime was already set by SetupStream / SetupURL.

7

* This is only needed by ReconnectStream.

8

*/

9

if

(seekTime > 0)

10

r->Link.seekTime = seekTime;

11

12

r->m_mediaChannel = 0;

13

14

//一个个包的读取，直到服务端告诉我说可以play了为止

15

while

(!r->m_bPlaying && RTMP_IsConnected(r) && RTMP_ReadPacket(r, &packet))

16

{

17

if

(RTMPPacket_IsReady(&packet))

//是否读取完毕。((a)->m_nBytesRead == (a)->m_nBodySize)

18

{

19

if

(!packet.m_nBodySize)

20

continue

;

21

if

((packet.m_packetType == RTMP_PACKET_TYPE_AUDIO) ||

22

(packet.m_packetType == RTMP_PACKET_TYPE_VIDEO) ||

23

(packet.m_packetType == RTMP_PACKET_TYPE_INFO))

24

{

25

RTMP_Log(RTMP_LOGWARNING,

"Received FLV packet before play()! Ignoring."

);

26

RTMPPacket_Free(&packet);

27

continue

;

28

}

29

30

RTMP_ClientPacket(r, &packet);

//处理一下这个数据包，其实里面就是处理服务端发送过来的各种消息等。直到接受到了play/publish

31

RTMPPacket_Free(&packet);

32

}

33

}

34

//返回当前是否接收到了play/publish 或者stopd等

35

return

r->m_bPlaying;

36

}

1

static

int

HandleInvoke(RTMP *r,

const

char

*body, unsigned

int

nBodySize){

2

AMFObject obj;

3

AVal method;

4

double

txn;

5

int

ret = 0, nRes;

6

7

nRes = AMF_Decode(&obj, body, nBodySize, FALSE);

8

if

(nRes < 0){

9

RTMP_Log(RTMP_LOGERROR,

"%s, error decoding invoke packet"

, __FUNCTION__);

10

return

0;

11

}

12

13

AMF_Dump(&obj);

14

AMFProp_GetString(AMF_GetProp(&obj, NULL, 0), &method);

15

txn = AMFProp_GetNumber(AMF_GetProp(&obj, NULL, 1));

16

RTMP_Log(RTMP_LOGDEBUG,

"%s, server invoking <%s>"

, __FUNCTION__, method.av_val);

17

18

if

(AVMATCH(&method, &av__result))

19

{

//接收到服务端返回的一个_result包，所以我们需要找到这个包对应的那条命令，从而处理这条命令的对应事件。

20

//比如我们之前发送了个connect给服务端，服务端必然会返回_result，然后我们异步收到result后，会调用

21

//RTMP_SendServerBW,RTMP_SendCtrl,以及RTMP_SendCreateStream来创建一个stream

22

AVal methodInvoked = {0};

23

int

i;

24

25

for

(i=0;i<r->m_numCalls; i++) {

//找到这条指令对应的触发的方法

26

if

(r->m_methodCalls[i].num == (

int

)txn) {

27

methodInvoked = r->m_methodCalls[i].name;

28

AV_erase(r->m_methodCalls, &r->m_numCalls, i, FALSE);

29

break

;

30

}

31

}

1

//下面根据不同的方法进行不同的处理

2

if

(AVMATCH(&methodInvoked, &av_connect))

3

{

4

if

(r->Link.protocol & RTMP_FEATURE_WRITE)

5

{

6

SendReleaseStream(r);

7

SendFCPublish(r);

8

}

9

else

10

{

//告诉服务端，我们的期望是什么，窗口大小，等

11

RTMP_SendServerBW(r);

12

RTMP_SendCtrl(r, 3, 0, 300);

13

}

14

RTMP_SendCreateStream(r);

//因为服务端同意了我们的connect，所以这里发送createStream创建一个流

15

//创建完成后，会再次进如这个函数从而走到下面的av_createStream分支，从而发送play过去

16

if

(!(r->Link.protocol & RTMP_FEATURE_WRITE))

17

{

18

/* Authenticate on Justin.tv legacy servers before sending FCSubscribe */

19

if

(r->Link.usherToken.av_len)

20

SendUsherToken(r, &r->Link.usherToken);

21

/* Send the FCSubscribe if live stream or if subscribepath is set */

22

if

(r->Link.subscribepath.av_len)

23

SendFCSubscribe(r, &r->Link.subscribepath);

24

else

if

(r->Link.lFlags & RTMP_LF_LIVE)

25

SendFCSubscribe(r, &r->Link.playpath);

26

}

27

}

28

else

if

(AVMATCH(&methodInvoked, &av_createStream))

1

else

if

(AVMATCH(&methodInvoked, &av_createStream))

2

{

3

r->m_stream_id = (

int

)AMFProp_GetNumber(AMF_GetProp(&obj, NULL, 3));

4

if

(r->Link.protocol & RTMP_FEATURE_WRITE)

5

{

//如果是要发送，那么高尚服务端，我们要发数据

6

SendPublish(r);

7

}

8

else

9

{

//否则告诉他我们要接受数据

10

if

(r->Link.lFlags & RTMP_LF_PLST)

11

SendPlaylist(r);

12

SendPlay(r);

//发送play过去,

13

RTMP_SendCtrl(r, 3, r->m_stream_id, r->m_nBufferMS);

//以及我们的buf大小

14

}

15

}

16

else

if

(AVMATCH(&methodInvoked, &av_play) ||

17

AVMATCH(&methodInvoked, &av_publish))

18

{

//接收到了play的回复，那么标记为play

19

r->m_bPlaying = TRUE;

20

}

21

free

(methodInvoked.av_val);

22

}