mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merged in rtmp-logging (pull request #102) 

Added proper logging and made more names idiomatic (although not finished yet.)

Approved-by: Saxon Milton <saxon.milton@gmail.com>
Approved-by: Alan Noble <anoble@gmail.com>
This commit is contained in:
Alan Noble 2019-01-10 05:17:06 +00:00
parent c4bc4f7c61 a318f9c3eb
commit 6e034fd264
5 changed files with 377 additions and 189 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
revid/senders.go
View File

@ -264,7 +264,7 @@ func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg
var sess *rtmp.Session
var err error
for n := 0; n < retries; n++ {
sess = rtmp.NewSession(url, timeout)
sess = rtmp.NewSession(url, timeout, log)
err = sess.Open()
if err == nil {
break
@ -310,7 +310,7 @@ func (s *rtmpSender) restart() error {
return err
}
for n := 0; n < s.retries; n++ {
s.sess = rtmp.NewSession(s.url, s.timeout)
s.sess = rtmp.NewSession(s.url, s.timeout, s.log)
err = s.sess.Open()
if err == nil {
break

117
rtmp/packet.go
View File

@ -35,8 +35,8 @@ LICENSE
package rtmp
import (
"io"
"encoding/binary"
"log"
)
const (
@ -68,8 +68,12 @@ const (
RTMP_CHANNEL_SOURCE = 0x04
)
// packetSize defines valid packet sizes.
var packetSize = [...]int{12, 8, 4, 1}
// headerSizes defines header sizes for header types 0, 1, 2 and 3 respectively:
// 0: full header (12 bytes)
// 1: header without message ID (8 bytes)
// 2: basic header + timestamp (4 byes)
// 3: basic header (chunk type and stream ID) (1 byte)
var headerSizes = [...]int{12, 8, 4, 1}
// packet defines an RTMP packet.
type packet struct {
@ -101,7 +105,10 @@ func readPacket(s *Session, pkt *packet) error {
err := readN(s, header[:1])
if err != nil {
log.Println("readPacket: failed to read RTMP packet header!")
s.log(DebugLevel, pkg+"failed to read packet header 1st byte", "error", err.Error())
if err == io.EOF {
s.log(WarnLevel, pkg+"EOF error; connection likely terminated")
}
return err
}
pkt.headerType = (header[0] & 0xc0) >> 6
@ -112,7 +119,7 @@ func readPacket(s *Session, pkt *packet) error {
case pkt.channel == 0:
err = readN(s, header[:1])
if err != nil {
log.Println("readPacket: failed to read rtmp packet header 2nd byte.")
s.log(DebugLevel, pkg+"failed to read packet header 2nd byte", "error", err.Error())
return err
}
header = header[1:]
@ -121,12 +128,11 @@ func readPacket(s *Session, pkt *packet) error {
case pkt.channel == 1:
err = readN(s, header[:2])
if err != nil {
log.Println("readPacket: failed to read RTMP packet 3rd byte")
s.log(DebugLevel, pkg+"failed to read packet header 3rd byte", "error", err.Error())
return err
}
header = header[2:]
pkt.channel = int32(binary.BigEndian.Uint16(header[:2])) + 64
}
if pkt.channel >= s.channelsAllocatedIn {
@ -134,49 +140,46 @@ func readPacket(s *Session, pkt *packet) error {
timestamp := append(s.channelTimestamp, make([]int32, 10)...)
var pkts []*packet
if s.vecChannelsIn == nil {
if s.channelsIn == nil {
pkts = make([]*packet, n)
} else {
pkts = append(s.vecChannelsIn[:pkt.channel:pkt.channel], make([]*packet, 10)...)
pkts = append(s.channelsIn[:pkt.channel:pkt.channel], make([]*packet, 10)...)
}
s.channelTimestamp = timestamp
s.vecChannelsIn = pkts
s.channelsIn = pkts
for i := int(s.channelsAllocatedIn); i < len(s.channelTimestamp); i++ {
s.channelTimestamp[i] = 0
}
for i := int(s.channelsAllocatedIn); i < int(n); i++ {
s.vecChannelsIn[i] = nil
s.channelsIn[i] = nil
}
s.channelsAllocatedIn = n
}
size := packetSize[pkt.headerType]
size := headerSizes[pkt.headerType]
switch {
case size == RTMP_LARGE_HEADER_SIZE:
pkt.hasAbsTimestamp = true
case size < RTMP_LARGE_HEADER_SIZE:
if s.vecChannelsIn[pkt.channel] != nil {
*pkt = *(s.vecChannelsIn[pkt.channel])
if s.channelsIn[pkt.channel] != nil {
*pkt = *(s.channelsIn[pkt.channel])
}
}
size--
if size > 0 {
err = readN(s, header[:size])
if err != nil {
log.Println("readPacket: failed to read rtmp packet heades.")
s.log(DebugLevel, pkg+"failed to read packet header", "error", err.Error())
return err
}
}
hSize := len(hbuf) - len(header) + size
if size >= 3 {
pkt.timestamp = C_AMF_DecodeInt24(header[:3])
if size >= 6 {
pkt.bodySize = C_AMF_DecodeInt24(header[3:6])
pkt.bytesRead = 0
@ -195,7 +198,7 @@ func readPacket(s *Session, pkt *packet) error {
if extendedTimestamp {
err = readN(s, header[size:size+4])
if err != nil {
log.Println("readPacket: Failed to read extended timestamp")
s.log(DebugLevel, pkg+"failed to read extended timestamp", "error", err.Error())
return err
}
// TODO: port this
@ -222,20 +225,20 @@ func readPacket(s *Session, pkt *packet) error {
err = readN(s, pkt.body[pkt.bytesRead:][:chunkSize])
if err != nil {
log.Println("readPacket: failed to read RTMP packet body")
s.log(DebugLevel, pkg+"failed to read packet body", "error", err.Error())
return err
}
pkt.bytesRead += uint32(chunkSize)
// keep the packet as ref for other packets on this channel
if s.vecChannelsIn[pkt.channel] == nil {
s.vecChannelsIn[pkt.channel] = &packet{}
if s.channelsIn[pkt.channel] == nil {
s.channelsIn[pkt.channel] = &packet{}
}
*(s.vecChannelsIn[pkt.channel]) = *pkt
*(s.channelsIn[pkt.channel]) = *pkt
if extendedTimestamp {
s.vecChannelsIn[pkt.channel].timestamp = 0xffffff
s.channelsIn[pkt.channel].timestamp = 0xffffff
}
if pkt.bytesRead != pkt.bodySize {
@ -248,13 +251,13 @@ func readPacket(s *Session, pkt *packet) error {
}
s.channelTimestamp[pkt.channel] = int32(pkt.timestamp)
s.vecChannelsIn[pkt.channel].body = nil
s.vecChannelsIn[pkt.channel].bytesRead = 0
s.vecChannelsIn[pkt.channel].hasAbsTimestamp = false
s.channelsIn[pkt.channel].body = nil
s.channelsIn[pkt.channel].bytesRead = 0
s.channelsIn[pkt.channel].hasAbsTimestamp = false
return nil
}
// resizePacket adjust the packet's storage to accommodate a body of the given size.
// resizePacket adjusts the packet's storage to accommodate a body of the given size.
func resizePacket(pkt *packet, size uint32, ht uint8) {
buf := make([]byte, RTMP_MAX_HEADER_SIZE+size)
pkt.headerType = ht
@ -271,20 +274,20 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
n := int(pkt.channel + 10)
var pkts []*packet
if s.vecChannelsOut == nil {
if s.channelsOut == nil {
pkts = make([]*packet, n)
} else {
pkts = append(s.vecChannelsOut[:pkt.channel:pkt.channel], make([]*packet, 10)...)
pkts = append(s.channelsOut[:pkt.channel:pkt.channel], make([]*packet, 10)...)
}
s.vecChannelsOut = pkts
s.channelsOut = pkts
for i := int(s.channelsAllocatedOut); i < n; i++ {
s.vecChannelsOut[i] = nil
s.channelsOut[i] = nil
}
s.channelsAllocatedOut = int32(n)
}
prevPkt = s.vecChannelsOut[pkt.channel]
prevPkt = s.channelsOut[pkt.channel]
if prevPkt != nil && pkt.headerType != RTMP_PACKET_SIZE_LARGE {
// compress a bit by using the prev packet's attributes
@ -300,8 +303,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
}
if pkt.headerType > 3 {
log.Printf("Sanity failed! trying to send header of type: 0x%02x.",
pkt.headerType)
s.log(WarnLevel, pkg+"unexpected header type", "type", pkt.headerType)
return errInvalidHeader
}
@ -310,7 +312,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
if pkt.body != nil {
// Span from -packetsize for the type to the start of the body.
headBytes = pkt.header
origIdx = RTMP_MAX_HEADER_SIZE - packetSize[pkt.headerType]
origIdx = RTMP_MAX_HEADER_SIZE - headerSizes[pkt.headerType]
} else {
// Allocate a new header and allow 6 bytes of movement backward.
var hbuf [RTMP_MAX_HEADER_SIZE]byte
@ -326,7 +328,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
cSize = 1
}
hSize := packetSize[pkt.headerType]
hSize := headerSizes[pkt.headerType]
if cSize != 0 {
origIdx -= cSize
hSize += cSize
@ -339,7 +341,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
if ts >= 0xffffff {
origIdx -= 4
hSize += 4
log.Printf("Larger timestamp than 24-bit: 0x%v", ts)
s.log(DebugLevel, pkg+"larger timestamp than 24 bits", "timestamp", ts)
}
headerIdx := origIdx
@ -367,7 +369,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
}
}
if packetSize[pkt.headerType] > 1 {
if headerSizes[pkt.headerType] > 1 {
res := ts
if ts > 0xffffff {
res = 0xffffff
@ -376,14 +378,14 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
headerIdx += 3 // 24bits
}
if packetSize[pkt.headerType] > 4 {
if headerSizes[pkt.headerType] > 4 {
C_AMF_EncodeInt24(headBytes[headerIdx:], int32(pkt.bodySize))
headerIdx += 3 // 24bits
headBytes[headerIdx] = pkt.packetType
headerIdx++
}
if packetSize[pkt.headerType] > 8 {
if headerSizes[pkt.headerType] > 8 {
n := int(encodeInt32LE(headBytes[headerIdx:headerIdx+4], pkt.info))
headerIdx += n
}
@ -396,39 +398,37 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
size := int(pkt.bodySize)
chunkSize := int(s.outChunkSize)
if debugMode {
log.Printf("sendPacket: %v->%v, size=%v", s.link.conn.LocalAddr(), s.link.conn.RemoteAddr(), size)
}
s.log(DebugLevel, pkg+"sending packet", "size", size, "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr())
// Send the previously deferred packet if combining it with the next packet would exceed the chunk size.
if s.defered != nil && len(s.defered)+size+hSize > chunkSize {
err := writeN(s, s.defered)
if s.deferred != nil && len(s.deferred)+size+hSize > chunkSize {
err := writeN(s, s.deferred)
if err != nil {
return err
}
s.defered = nil
s.deferred = nil
}
// TODO(kortschak): Rewrite this horrific peice of premature optimisation.
// NB: RTMP wants packets in chunks which are 128 bytes by default, but the server may request a different size.
for size+hSize != 0 {
if s.defered == nil && pkt.packetType == RTMP_PACKET_TYPE_AUDIO && size < chunkSize {
s.defered = headBytes[origIdx:][:size+hSize]
if s.deferred == nil && pkt.packetType == RTMP_PACKET_TYPE_AUDIO && size < chunkSize {
s.deferred = headBytes[origIdx:][:size+hSize]
s.log(DebugLevel, pkg+"deferred sending packet")
break
}
if chunkSize > size {
chunkSize = size
}
bytes := headBytes[origIdx:][:chunkSize+hSize]
if s.defered != nil {
if s.deferred != nil {
// Prepend the previously deferred packet and write it with the current one.
bytes = append(s.defered, bytes...)
bytes = append(s.deferred, bytes...)
}
err := writeN(s, bytes)
if err != nil {
return err
}
s.defered = nil
s.deferred = nil
size -= chunkSize
origIdx += chunkSize + hSize
@ -464,22 +464,19 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
if pkt.packetType == RTMP_PACKET_TYPE_INVOKE {
buf := pkt.body[1:]
meth := C_AMF_DecodeString(buf)
if debugMode {
log.Printf("invoking %v", meth)
}
// keep it in call queue till result arrives
if queue {
s.log(DebugLevel, pkg+"queuing method "+meth)
buf = buf[3+len(meth):]
txn := int32(C_AMF_DecodeNumber(buf[:8]))
s.methodCalls = append(s.methodCalls, method{name: meth, num: txn})
}
}
if s.vecChannelsOut[pkt.channel] == nil {
s.vecChannelsOut[pkt.channel] = &packet{}
if s.channelsOut[pkt.channel] == nil {
s.channelsOut[pkt.channel] = &packet{}
}
*(s.vecChannelsOut[pkt.channel]) = *pkt
*(s.channelsOut[pkt.channel]) = *pkt
return nil
}

166
rtmp/rtmp.go
View File

@ -38,9 +38,7 @@ import (
"bytes"
"encoding/binary"
"errors"
"fmt"
"io"
"log"
"math/rand"
"net"
"strconv"
@ -48,9 +46,8 @@ import (
)
const (
pkg = "rtmp:"
minDataSize = 11
debugMode = false
length = 512
)
const (
@ -123,6 +120,7 @@ var rtmpProtocolStrings = [...]string{
// RTMP errors.
var (
errUnknownScheme = errors.New("rtmp: unknown scheme")
errConnected = errors.New("rtmp: already connected")
errNotConnected = errors.New("rtmp: not connected")
errHandshake = errors.New("rtmp: handshake failed")
errConnSend = errors.New("rtmp: connection send error")
@ -144,8 +142,7 @@ func setupURL(s *Session, addr string) (err error) {
if s.link.tcUrl == "" {
if s.link.app != "" {
s.link.tcUrl = fmt.Sprintf("%v://%v:%v/%v",
rtmpProtocolStrings[s.link.protocol], s.link.host, s.link.port, s.link.app)
s.link.tcUrl = rtmpProtocolStrings[s.link.protocol] + "://" + s.link.host + ":" + strconv.Itoa(int(s.link.port)) + "/" + s.link.app
s.link.lFlags |= RTMP_LF_FTCU
} else {
s.link.tcUrl = addr
@ -156,6 +153,7 @@ func setupURL(s *Session, addr string) (err error) {
switch {
case (s.link.protocol & RTMP_FEATURE_SSL) != 0:
s.link.port = 433
s.log(FatalLevel, pkg+"SSL not supported")
case (s.link.protocol & RTMP_FEATURE_HTTP) != 0:
s.link.port = 80
default:
@ -173,22 +171,19 @@ func connect(s *Session) error {
}
s.link.conn, err = net.DialTCP("tcp4", nil, addr)
if err != nil {
s.log(WarnLevel, pkg+"dial failed", "error", err.Error())
return err
}
if debugMode {
log.Println("... connected, handshaking...")
}
s.log(DebugLevel, pkg+"connected")
err = handshake(s)
if err != nil {
log.Println("connect: handshake failed")
s.log(WarnLevel, pkg+"handshake failed", "error", err.Error())
return errHandshake
}
if debugMode {
log.Println("... handshaked...")
}
s.log(DebugLevel, pkg+"handshaked")
err = sendConnectPacket(s)
if err != nil {
log.Println("connect: sendConnect failed")
s.log(WarnLevel, pkg+"sendConnect failed", "error", err.Error())
return errConnSend
}
return nil
@ -196,40 +191,38 @@ func connect(s *Session) error {
// connectStream reads a packet and handles it
func connectStream(s *Session) error {
var pkt packet
var err error
for !s.isPlaying && s.isConnected() {
err := readPacket(s, &pkt)
pkt := packet{}
err = readPacket(s, &pkt)
if err != nil {
break
}
if pkt.bodySize == 0 {
continue
}
if pkt.packetType == RTMP_PACKET_TYPE_AUDIO ||
pkt.packetType == RTMP_PACKET_TYPE_VIDEO ||
pkt.packetType == RTMP_PACKET_TYPE_INFO {
log.Println("connectStream: got packet before play()! Ignoring.")
pkt.body = nil
s.log(DebugLevel, pkg+"got packet before play; ignoring")
continue
}
handlePacket(s, &pkt)
pkt.body = nil
err = handlePacket(s, &pkt)
if err != nil {
break
}
}
if !s.isPlaying {
return errConnStream
return err
}
return nil
}
// handlePacket handles a packet that the client has received.
// NB: cases have been commented out that are not currently used by AusOcean
func handlePacket(s *Session, pkt *packet) int32 {
var hasMediaPacket int32
func handlePacket(s *Session, pkt *packet) error {
switch pkt.packetType {
case RTMP_PACKET_TYPE_CHUNK_SIZE:
@ -241,7 +234,7 @@ func handlePacket(s *Session, pkt *packet) int32 {
s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
case RTMP_PACKET_TYPE_CONTROL:
panic("Unsupported packet type RTMP_PACKET_TYPE_CONTROL")
s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_CONTROL")
case RTMP_PACKET_TYPE_SERVER_BW:
s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
@ -255,36 +248,32 @@ func handlePacket(s *Session, pkt *packet) int32 {
}
case RTMP_PACKET_TYPE_AUDIO:
panic("Unsupported packet type RTMP_PACKET_TYPE_AUDIO")
s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_AUDIO")
case RTMP_PACKET_TYPE_VIDEO:
panic("Unsupported packet type RTMP_PACKET_TYPE_VIDEO")
s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_VIDEO")
case RTMP_PACKET_TYPE_FLEX_MESSAGE:
panic("Unsupported packet type RTMP_PACKET_TYPE_FLEX_MESSAGE")
s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_FLEX_MESSAGE")
case RTMP_PACKET_TYPE_INFO:
panic("Unsupported packet type RTMP_PACKET_TYPE_INFO")
s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_INFO")
case RTMP_PACKET_TYPE_INVOKE:
if debugMode {
log.Println("RTMP_PACKET_TYPE_INVOKE:")
}
err := handleInvoke(s, pkt.body[:pkt.bodySize])
if err != nil {
// This will never happen with the methods we implement.
log.Println("HasMediaPacket")
hasMediaPacket = 2
s.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error())
return err
}
case RTMP_PACKET_TYPE_FLASH_VIDEO:
panic("Unsupported packet type RTMP_PACKET_TYPE_FLASH_VIDEO")
s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_FLASH_VIDEO")
default:
// TODO use new logger here
// RTMP_Log(RTMP_LOGDEBUG, "%s, unknown packet type received: 0x%02x", __FUNCTION__,pkt.packetType);
s.log(WarnLevel, pkg+"unknown packet type", "type", pkt.packetType)
}
return hasMediaPacket
return nil
}
func readN(s *Session, buf []byte) error {
@ -294,10 +283,7 @@ func readN(s *Session, buf []byte) error {
}
n, err := io.ReadFull(s.link.conn, buf)
if err != nil {
if debugMode {
log.Printf("readN error: %v\n", err)
}
s.close()
s.log(DebugLevel, pkg+"read failed", "error", err.Error())
return err
}
s.nBytesIn += int32(n)
@ -318,10 +304,7 @@ func writeN(s *Session, buf []byte) error {
}
_, err = s.link.conn.Write(buf)
if err != nil {
if debugMode {
log.Printf("writeN, RTMP send error: %v\n", err)
}
s.close()
s.log(WarnLevel, pkg+"write failed", "error", err.Error())
return err
}
return nil
@ -417,12 +400,10 @@ func sendConnectPacket(s *Session) error {
}
}
if copy(enc, []byte{0, 0, AMF_OBJECT_END}) != 3 {
return errCopying // TODO: is this even possible?
}
copy(enc, []byte{0, 0, AMF_OBJECT_END})
enc = enc[3:]
/* add auth string */
// add auth string
if s.link.auth != "" {
enc = C_AMF_EncodeBoolean(enc, s.link.lFlags&RTMP_LF_AUTH != 0)
if enc == nil {
@ -443,7 +424,7 @@ func sendConnectPacket(s *Session) error {
pkt.bodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return sendPacket(s, &pkt, true)
return sendPacket(s, &pkt, true) // response expected
}
func sendCreateStream(s *Session) error {
@ -471,7 +452,7 @@ func sendCreateStream(s *Session) error {
pkt.bodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return sendPacket(s, &pkt, true)
return sendPacket(s, &pkt, true) // response expected
}
func sendReleaseStream(s *Session) error {
@ -602,7 +583,7 @@ func sendPublish(s *Session) error {
pkt.bodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return sendPacket(s, &pkt, true)
return sendPacket(s, &pkt, true) // response expected
}
func sendDeleteStream(s *Session, dStreamId float64) error {
@ -706,9 +687,8 @@ func handleInvoke(s *Session, body []byte) error {
}
meth := C_AMFProp_GetString(C_AMF_GetProp(&obj, "", 0))
s.log(DebugLevel, pkg+"invoking method "+meth)
txn := C_AMFProp_GetNumber(C_AMF_GetProp(&obj, "", 1))
// TODO use new logger here
// RTMP_Log(RTMP_LOGDEBUG, "%s, server invoking <%s>", __FUNCTION__, method.av_val);
switch meth {
case av__result:
@ -721,30 +701,27 @@ func handleInvoke(s *Session, body []byte) error {
}
}
if methodInvoked == "" {
// TODO use new logger here
//RTMP_Log(RTMP_LOGDEBUG, "%s, received result id %f without matching request",
//__FUNCTION__, txn);
s.log(WarnLevel, pkg+"received result without matching request", "id", txn)
goto leave
}
// TODO use new logger here
//RTMP_Log(RTMP_LOGDEBUG, "%s, received result for method call <%s>", __FUNCTION__,
//methodInvoked.av_val);
s.log(DebugLevel, pkg+"received result for "+methodInvoked)
switch methodInvoked {
case av_connect:
if s.link.token != "" {
panic("No support for link token")
s.log(FatalLevel, pkg+"no support for link token")
}
if (s.link.protocol & RTMP_FEATURE_WRITE) != 0 {
sendReleaseStream(s)
sendFCPublish(s)
} else {
panic("Link protocol has no RTMP_FEATURE_WRITE")
s.log(FatalLevel, pkg+"link protocol has no RTMP_FEATURE_WRITE")
}
sendCreateStream(s)
if (s.link.protocol & RTMP_FEATURE_WRITE) == 0 {
panic("Link protocol has no RTMP_FEATURE_WRITE")
s.log(FatalLevel, pkg+"link protocol has no RTMP_FEATURE_WRITE")
}
case av_createStream:
@ -753,54 +730,50 @@ func handleInvoke(s *Session, body []byte) error {
if s.link.protocol&RTMP_FEATURE_WRITE != 0 {
sendPublish(s)
} else {
panic("Link protocol has no RTMP_FEATURE_WRITE")
s.log(FatalLevel, pkg+"link protocol has no RTMP_FEATURE_WRITE")
}
case av_play, av_publish:
panic("Unsupported method av_play/av_publish")
s.log(FatalLevel, pkg+"unsupported method av_play/av_publish")
}
//C.free(unsafe.Pointer(methodInvoked.av_val))
case av_onBWDone:
if s.bwCheckCounter == 0 {
if s.checkCounter == 0 { // ToDo: why is this always zero?
sendCheckBW(s)
}
case av_onFCUnsubscribe, av_onFCSubscribe:
panic("Unsupported method av_onFCUnsubscribe/av_onFCSubscribe")
s.log(FatalLevel, pkg+"unsupported method av_onFCUnsubscribe/av_onFCSubscribe")
case av_ping:
panic("Unsupported method av_ping")
s.log(FatalLevel, pkg+"unsupported method av_ping")
case av__onbwcheck:
panic("Unsupported method av_onbwcheck")
s.log(FatalLevel, pkg+"unsupported method av_onbwcheck")
case av__onbwdone:
panic("Unsupported method av_onbwdone")
s.log(FatalLevel, pkg+"unsupported method av_onbwdone")
case av_close:
panic("Unsupported method av_close")
s.log(FatalLevel, pkg+"unsupported method av_close")
case av_onStatus:
var obj2 C_AMFObject
C_AMFProp_GetObject(C_AMF_GetProp(&obj, "", 3), &obj2)
code := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_code, -1))
level := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_level, -1)) // Not used.
_ = level
// TODO use new logger
// RTMP_Log(RTMP_LOGDEBUG, "%s, onStatus: %s", __FUNCTION__, code.av_val);
level := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_level, -1))
s.log(DebugLevel, pkg+"onStatus", "code", code, "level", level)
switch code {
case av_NetStream_Failed, av_NetStream_Play_Failed,
av_NetStream_Play_StreamNotFound, av_NetConnection_Connect_InvalidApp:
panic("Unsupported method av_NetStream/av_NetStream_Play_Failed/av_netSTream_Play_StreamNotFound/av_netConnection_Connect_invalidApp")
s.log(FatalLevel, pkg+"unsupported method av_NetStream/av_NetStream_Play_Failed/av_netSTream_Play_StreamNotFound/av_netConnection_Connect_invalidApp")
case av_NetStream_Play_Start, av_NetStream_Play_PublishNotify:
panic("Unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify")
s.log(FatalLevel, pkg+"unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify")
case av_NetStream_Publish_Start:
s.log(DebugLevel, pkg+"playing")
s.isPlaying = true
for i, m := range s.methodCalls {
if m.name == av_publish {
@ -808,22 +781,23 @@ func handleInvoke(s *Session, body []byte) error {
break
}
}
// ToDo: handle case when av_publish method not found
case av_NetStream_Play_Complete, av_NetStream_Play_Stop, av_NetStream_Play_UnpublishNotify:
panic("Unsupported method av_NetStream_Play_Complete/av_NetStream_Play_Stop/av_NetStream_Play_UnpublishNotify")
s.log(FatalLevel, pkg+"unsupported method av_NetStream_Play_Complete/av_NetStream_Play_Stop/av_NetStream_Play_UnpublishNotify")
case av_NetStream_Seek_Notify:
panic("Unsupported method av_netStream_Seek_Notify")
s.log(FatalLevel, pkg+"unsupported method av_netStream_Seek_Notify")
case av_NetStream_Pause_Notify:
panic("Unsupported method av_NetStream_Pause_Notify")
s.log(FatalLevel, pkg+"unsupported method av_NetStream_Pause_Notify")
}
case av_playlist_ready:
panic("Unsupported method av_playlist_ready")
s.log(FatalLevel, pkg+"unsupported method av_playlist_ready")
default:
panic(fmt.Sprintf("unknown method: %q", meth))
s.log(FatalLevel, pkg+"unknown method "+meth)
}
leave:
C_AMF_Reset(&obj)
@ -847,19 +821,17 @@ func handshake(s *Session) error {
if err != nil {
return err
}
s.log(DebugLevel, pkg+"handshake sent")
var typ [1]byte
err = readN(s, typ[:])
if err != nil {
return err
}
s.log(DebugLevel, pkg+"handshake received")
if debugMode {
log.Printf("handshake: Type answer: %v\n", typ[0])
}
if typ[0] != clientbuf[0] {
log.Printf("handshake: type mismatch: client sent %v, server sent: %v\n",
clientbuf[0], typ)
s.log(WarnLevel, pkg+"handshake type mismatch", "sent", clientbuf[0], "received", typ)
}
err = readN(s, serversig[:])
if err != nil {
@ -868,10 +840,7 @@ func handshake(s *Session) error {
// decode server response
suptime := binary.BigEndian.Uint32(serversig[:4])
_ = 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])
s.log(DebugLevel, pkg+"server uptime", "uptime", suptime)
// 2nd part of handshake
err = writeN(s, serversig[:])
@ -885,8 +854,7 @@ func handshake(s *Session) error {
}
if !bytes.Equal(serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1]) {
log.Printf("Client signature does not match: %q != %q",
serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1])
s.log(WarnLevel, pkg+"signature mismatch", "serversig", serversig[:RTMP_SIG_SIZE], "clientsig", clientbuf[1:RTMP_SIG_SIZE+1])
}
return nil
}

217
rtmp/rtmp_test.go Normal file
View File

@ -0,0 +1,217 @@
/*
NAME
rtmp_test.go
DESCRIPTION
RTMP tests
AUTHORS
Alan Noble <alan@ausocean.org>
LICENSE
rtmp_test.go is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean)
It is free software: you can redistribute it and/or modify them
under the terms of the GNU General Public License as published by the
Free Software Foundation, either version 3 of the License, or (at your
option) any later version.
It is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
for more details.
You should have received a copy of the GNU General Public License
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
*/
package rtmp
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"runtime"
"testing"
"time"
"bitbucket.org/ausocean/av/stream/flv"
"bitbucket.org/ausocean/av/stream/lex"
)
const (
rtmpProtocol = "rtmp"
testHost = "a.rtmp.youtube.com"
testApp = "live2"
testBaseURL = rtmpProtocol + "://" + testHost + "/" + testApp + "/"
testTimeout = 30
testDataDir = "../../test/test-data/av/input"
)
// testVerbosity controls the amount of output
// NB: This is not the log level, which is DebugLevel.
// 0: suppress logging completely
// 1: log messages only
// 2: log messages with errors, if any
var testVerbosity = 1
// testKey is the RTMP key required for YouTube streaming (RTMP_TEST_KEY env var)
var testKey string
// testFile is the test video file (RTMP_TEST_FILE env var)
var testFile string
// testLog is a bare bones logger that logs to stdout.
func testLog(level int8, msg string, params ...interface{}) {
logLevels := [...]string{"Debug", "Info", "Warn", "Error", "", "", "Fatal"}
if testVerbosity == 0 {
return
}
if level < -1 || level > 5 {
panic("Invalid log level")
}
if testVerbosity == 2 && len(params) >= 2 {
// extract the params we know about, otherwise just print the message
switch params[0].(string) {
case "error":
fmt.Printf("%s: %s, error=%v\n", logLevels[level+1], msg, params[1].(string))
case "size":
fmt.Printf("%s: %s, size=%d\n", logLevels[level+1], msg, params[1].(int))
default:
fmt.Printf("%s: %s\n", logLevels[level+1], msg)
}
} else {
fmt.Printf("%s: %s\n", logLevels[level+1], msg)
}
if level == 5 {
// Fatal
buf := make([]byte, 1<<16)
size := runtime.Stack(buf, true)
fmt.Printf("%s\n", string(buf[:size]))
os.Exit(1)
}
}
// TestKey tests that the RTMP_TEST_KEY environment variable is present
func TestKey(t *testing.T) {
testLog(0, "TestKey")
testKey = os.Getenv("RTMP_TEST_KEY")
if testKey == "" {
msg := "RTMP_TEST_KEY environment variable not defined"
testLog(0, msg)
t.Skip(msg)
}
testLog(0, "Testing against URL "+testBaseURL+testKey)
}
// TestSetupURL tests URL parsing.
func TestSetupURL(t *testing.T) {
testLog(0, "TestSetupURL")
// test with just the base URL
s := NewSession(testBaseURL, testTimeout, testLog)
if s.url != testBaseURL && s.link.timeout != testTimeout {
t.Errorf("NewSession failed")
}
err := setupURL(s, s.url)
if err != nil {
t.Errorf("setupURL(testBaseURL) failed with error: %v", err)
}
// test the parts are as expected
if rtmpProtocolStrings[s.link.protocol] != rtmpProtocol {
t.Errorf("setupURL returned wrong protocol: %v", s.link.protocol)
}
if s.link.host != testHost {
t.Errorf("setupURL returned wrong host: %v", s.link.host)
}
if s.link.app != testApp {
t.Errorf("setupURL returned wrong app: %v", s.link.app)
}
}
// TestOpenClose tests opening an closing an RTMP connection.
func TestOpenClose(t *testing.T) {
testLog(0, "TestOpenClose")
if testKey == "" {
t.Skip("Skipping TestOpenClose since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
if err != nil {
t.Errorf("Open failed with error: %v", err)
return
}
err = s.Close()
if err != nil {
t.Errorf("Close failed with error: %v", err)
return
}
}
// TestFromFrame tests streaming from a single H.264 frame which is repeated.
func TestFromFrame(t *testing.T) {
testLog(0, "TestFromFrame")
if testKey == "" {
t.Skip("Skipping TestFromFrame since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
if err != nil {
t.Errorf("Session.Open failed with error: %v", err)
}
b, err := ioutil.ReadFile(filepath.Join(testDataDir, "AusOcean_logo_1080p.h264"))
if err != nil {
t.Errorf("ReadFile failed with error: %v", err)
}
// Pass RTMP session, true for audio, true for video, and 25 FPS
flvEncoder := flv.NewEncoder(s, true, true, 25)
for i := 0; i < 25; i++ {
err := flvEncoder.Encode(b)
if err != nil {
t.Errorf("Encoding failed with error: %v", err)
}
}
err = s.Close()
if err != nil {
t.Errorf("Session.Close failed with error: %v", err)
}
}
// TestFromFile tests streaming from an video file comprising raw H.264.
// The test file is supplied via the RTMP_TEST_FILE environment variable.
func TestFromFile(t *testing.T) {
testLog(0, "TestFromFile")
testFile := os.Getenv("RTMP_TEST_FILE")
if testFile == "" {
t.Skip("Skipping TestFromFile since no RTMP_TEST_FILE")
}
if testKey == "" {
t.Skip("Skipping TestFromFile since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
if err != nil {
t.Errorf("Session.Open failed with error: %v", err)
}
f, err := os.Open(testFile)
if err != nil {
t.Errorf("Open failed with error: %v", err)
}
defer f.Close()
// Pass RTMP session, true for audio, true for video, and 25 FPS
flvEncoder := flv.NewEncoder(s, true, true, 25)
err = lex.H264(flvEncoder, f, time.Second/time.Duration(25))
if err != nil {
t.Errorf("Lexing and encoding failed with error: %v", err)
}
err = s.Close()
if err != nil {
t.Errorf("Session.Close failed with error: %v", err)
}
}

62
rtmp/session.go
View File

@ -33,17 +33,12 @@ LICENSE
*/
package rtmp
import (
"errors"
)
// Session holds the state for an RTMP session.
type Session struct {
url string
timeout uint
inChunkSize int32
outChunkSize int32
bwCheckCounter int32
checkCounter int32
nBytesIn int32
nBytesInSent int32
streamID int32
@ -56,28 +51,52 @@ type Session struct {
methodCalls []method
channelsAllocatedIn int32
channelsAllocatedOut int32
vecChannelsIn []*packet
vecChannelsOut []*packet
channelsIn []*packet
channelsOut []*packet
channelTimestamp []int32
audioCodecs float64
videoCodecs float64
encoding float64
defered []byte
deferred []byte
link link
log Log
}
// Log defines the RTMP logging function.
type Log func(level int8, message string, params ...interface{})
// Log levels used by Log.
const (
DebugLevel int8 = -1
InfoLevel int8 = 0
WarnLevel int8 = 1
ErrorLevel int8 = 2
FatalLevel int8 = 5
)
// NewSession returns a new Session.
func NewSession(url string, connectTimeout uint) *Session {
func NewSession(url string, timeout uint, log Log) *Session {
return &Session{
url: url,
timeout: connectTimeout,
url: url,
inChunkSize: 128,
outChunkSize: 128,
clientBW: 2500000,
clientBW2: 2,
serverBW: 2500000,
audioCodecs: 3191.0,
videoCodecs: 252.0,
log: log,
link: link{
timeout: timeout,
swfAge: 30,
},
}
}
// Open establishes an rtmp connection with the url passed into the constructor.
func (s *Session) Open() error {
if s.isConnected() {
return errors.New("rtmp: attempt to start already running session")
return errConnected
}
err := s.start()
if err != nil {
@ -88,7 +107,7 @@ func (s *Session) Open() error {
// start does the heavylifting for Open().
func (s *Session) start() error {
s.init()
s.log(DebugLevel, pkg+"Session.start")
err := setupURL(s, s.url)
if err != nil {
s.close()
@ -110,20 +129,6 @@ func (s *Session) start() error {
return nil
}
// init initializes various RTMP defauls.
// ToDo: define consts for the magic numbers.
func (s *Session) init() {
s.inChunkSize = RTMP_DEFAULT_CHUNKSIZE
s.outChunkSize = RTMP_DEFAULT_CHUNKSIZE
s.clientBW = 2500000
s.clientBW2 = 2
s.serverBW = 2500000
s.audioCodecs = 3191.0
s.videoCodecs = 252.0
s.link.timeout = s.timeout
s.link.swfAge = 30
}
// Close terminates the rtmp connection,
func (s *Session) Close() error {
if !s.isConnected() {
@ -136,6 +141,7 @@ func (s *Session) Close() error {
// close does the heavylifting for Close().
// Any errors are ignored as it is often called in response to an earlier error.
func (s *Session) close() {
s.log(DebugLevel, pkg+"Session.close")
if s.isConnected() {
if s.streamID > 0 {
if s.link.protocol&RTMP_FEATURE_WRITE != 0 {