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

Added proper logging.

This commit is contained in:
scruzin 2019-01-07 23:59:41 +10:30
parent c4bc4f7c61
commit cbe6149c8f
4 changed files with 86 additions and 116 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 sess *rtmp.Session
var err error var err error
for n := 0; n < retries; n++ { for n := 0; n < retries; n++ {
sess = rtmp.NewSession(url, timeout) sess = rtmp.NewSession(url, timeout, log)
err = sess.Open() err = sess.Open()
if err == nil { if err == nil {
break break
@ -310,7 +310,7 @@ func (s *rtmpSender) restart() error {
return err return err
} }
for n := 0; n < s.retries; n++ { 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() err = s.sess.Open()
if err == nil { if err == nil {
break break

51
rtmp/packet.go
View File

@ -36,7 +36,6 @@ package rtmp
import ( import (
"encoding/binary" "encoding/binary"
"log"
) )
const ( const (
@ -101,7 +100,7 @@ func readPacket(s *Session, pkt *packet) error {
err := readN(s, header[:1]) err := readN(s, header[:1])
if err != nil { 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())
return err return err
} }
pkt.headerType = (header[0] & 0xc0) >> 6 pkt.headerType = (header[0] & 0xc0) >> 6
@ -112,7 +111,7 @@ func readPacket(s *Session, pkt *packet) error {
case pkt.channel == 0: case pkt.channel == 0:
err = readN(s, header[:1]) err = readN(s, header[:1])
if err != nil { 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 return err
} }
header = header[1:] header = header[1:]
@ -121,12 +120,11 @@ func readPacket(s *Session, pkt *packet) error {
case pkt.channel == 1: case pkt.channel == 1:
err = readN(s, header[:2]) err = readN(s, header[:2])
if err != nil { 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 return err
} }
header = header[2:] header = header[2:]
pkt.channel = int32(binary.BigEndian.Uint16(header[:2])) + 64 pkt.channel = int32(binary.BigEndian.Uint16(header[:2])) + 64
} }
if pkt.channel >= s.channelsAllocatedIn { if pkt.channel >= s.channelsAllocatedIn {
@ -161,22 +159,19 @@ func readPacket(s *Session, pkt *packet) error {
*pkt = *(s.vecChannelsIn[pkt.channel]) *pkt = *(s.vecChannelsIn[pkt.channel])
} }
} }
size-- size--
if size > 0 { if size > 0 {
err = readN(s, header[:size]) err = readN(s, header[:size])
if err != nil { 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 return err
} }
} }
hSize := len(hbuf) - len(header) + size hSize := len(hbuf) - len(header) + size
if size >= 3 { if size >= 3 {
pkt.timestamp = C_AMF_DecodeInt24(header[:3]) pkt.timestamp = C_AMF_DecodeInt24(header[:3])
if size >= 6 { if size >= 6 {
pkt.bodySize = C_AMF_DecodeInt24(header[3:6]) pkt.bodySize = C_AMF_DecodeInt24(header[3:6])
pkt.bytesRead = 0 pkt.bytesRead = 0
@ -195,7 +190,7 @@ func readPacket(s *Session, pkt *packet) error {
if extendedTimestamp { if extendedTimestamp {
err = readN(s, header[size:size+4]) err = readN(s, header[size:size+4])
if err != nil { 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 return err
} }
// TODO: port this // TODO: port this
@ -222,7 +217,7 @@ func readPacket(s *Session, pkt *packet) error {
err = readN(s, pkt.body[pkt.bytesRead:][:chunkSize]) err = readN(s, pkt.body[pkt.bytesRead:][:chunkSize])
if err != nil { 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 return err
} }
@ -254,7 +249,7 @@ func readPacket(s *Session, pkt *packet) error {
return nil 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) { func resizePacket(pkt *packet, size uint32, ht uint8) {
buf := make([]byte, RTMP_MAX_HEADER_SIZE+size) buf := make([]byte, RTMP_MAX_HEADER_SIZE+size)
pkt.headerType = ht pkt.headerType = ht
@ -300,8 +295,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
} }
if pkt.headerType > 3 { if pkt.headerType > 3 {
log.Printf("Sanity failed! trying to send header of type: 0x%02x.", s.log(WarnLevel, pkg+"unexpected header type", "type", pkt.headerType)
pkt.headerType)
return errInvalidHeader return errInvalidHeader
} }
@ -339,7 +333,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
if ts >= 0xffffff { if ts >= 0xffffff {
origIdx -= 4 origIdx -= 4
hSize += 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 headerIdx := origIdx
@ -396,39 +390,37 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
size := int(pkt.bodySize) size := int(pkt.bodySize)
chunkSize := int(s.outChunkSize) chunkSize := int(s.outChunkSize)
if debugMode { s.log(DebugLevel, pkg+"sending packet", "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr(), "size", size)
log.Printf("sendPacket: %v->%v, size=%v", s.link.conn.LocalAddr(), s.link.conn.RemoteAddr(), size)
}
// Send the previously deferred packet if combining it with the next packet would exceed the chunk size. if s.deferred != nil && len(s.deferred)+size+hSize > chunkSize {
if s.defered != nil && len(s.defered)+size+hSize > chunkSize { err := writeN(s, s.deferred)
err := writeN(s, s.defered)
if err != nil { if err != nil {
return err return err
} }
s.defered = nil s.deferred = nil
} }
// TODO(kortschak): Rewrite this horrific peice of premature optimisation. // 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. // NB: RTMP wants packets in chunks which are 128 bytes by default, but the server may request a different size.
for size+hSize != 0 { for size+hSize != 0 {
if s.defered == nil && pkt.packetType == RTMP_PACKET_TYPE_AUDIO && size < chunkSize { if s.deferred == nil && pkt.packetType == RTMP_PACKET_TYPE_AUDIO && size < chunkSize {
s.defered = headBytes[origIdx:][:size+hSize] s.deferred = headBytes[origIdx:][:size+hSize]
s.log(DebugLevel, pkg+"deferred sending packet")
break break
} }
if chunkSize > size { if chunkSize > size {
chunkSize = size chunkSize = size
} }
bytes := headBytes[origIdx:][:chunkSize+hSize] 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. // 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) err := writeN(s, bytes)
if err != nil { if err != nil {
return err return err
} }
s.defered = nil s.deferred = nil
size -= chunkSize size -= chunkSize
origIdx += chunkSize + hSize origIdx += chunkSize + hSize
@ -464,10 +456,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
if pkt.packetType == RTMP_PACKET_TYPE_INVOKE { if pkt.packetType == RTMP_PACKET_TYPE_INVOKE {
buf := pkt.body[1:] buf := pkt.body[1:]
meth := C_AMF_DecodeString(buf) meth := C_AMF_DecodeString(buf)
s.log(DebugLevel, "invoking method", "method", meth)
if debugMode {
log.Printf("invoking %v", meth)
}
// keep it in call queue till result arrives // keep it in call queue till result arrives
if queue { if queue {
buf = buf[3+len(meth):] buf = buf[3+len(meth):]

121
rtmp/rtmp.go
View File

@ -38,9 +38,7 @@ import (
"bytes" "bytes"
"encoding/binary" "encoding/binary"
"errors" "errors"
"fmt"
"io" "io"
"log"
"math/rand" "math/rand"
"net" "net"
"strconv" "strconv"
@ -48,9 +46,8 @@ import (
) )
const ( const (
pkg = "rtmp:"
minDataSize = 11 minDataSize = 11
debugMode = false
length = 512
) )
const ( const (
@ -123,6 +120,7 @@ var rtmpProtocolStrings = [...]string{
// RTMP errors. // RTMP errors.
var ( var (
errUnknownScheme = errors.New("rtmp: unknown scheme") errUnknownScheme = errors.New("rtmp: unknown scheme")
errConnected = errors.New("rtmp: already connected")
errNotConnected = errors.New("rtmp: not connected") errNotConnected = errors.New("rtmp: not connected")
errHandshake = errors.New("rtmp: handshake failed") errHandshake = errors.New("rtmp: handshake failed")
errConnSend = errors.New("rtmp: connection send error") 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.tcUrl == "" {
if s.link.app != "" { if s.link.app != "" {
s.link.tcUrl = fmt.Sprintf("%v://%v:%v/%v", s.link.tcUrl = rtmpProtocolStrings[s.link.protocol] + "://" + s.link.host + ":" + strconv.Itoa(int(s.link.port)) + "/" + s.link.app
rtmpProtocolStrings[s.link.protocol], s.link.host, s.link.port, s.link.app)
s.link.lFlags |= RTMP_LF_FTCU s.link.lFlags |= RTMP_LF_FTCU
} else { } else {
s.link.tcUrl = addr s.link.tcUrl = addr
@ -156,6 +153,7 @@ func setupURL(s *Session, addr string) (err error) {
switch { switch {
case (s.link.protocol & RTMP_FEATURE_SSL) != 0: case (s.link.protocol & RTMP_FEATURE_SSL) != 0:
s.link.port = 433 s.link.port = 433
s.log(FatalLevel, pkg+"SSL not supported")
case (s.link.protocol & RTMP_FEATURE_HTTP) != 0: case (s.link.protocol & RTMP_FEATURE_HTTP) != 0:
s.link.port = 80 s.link.port = 80
default: default:
@ -173,22 +171,19 @@ func connect(s *Session) error {
} }
s.link.conn, err = net.DialTCP("tcp4", nil, addr) s.link.conn, err = net.DialTCP("tcp4", nil, addr)
if err != nil { if err != nil {
s.log(WarnLevel, pkg+"dial failed", "error", err.Error())
return err return err
} }
if debugMode { s.log(DebugLevel, pkg+"connected")
log.Println("... connected, handshaking...")
}
err = handshake(s) err = handshake(s)
if err != nil { if err != nil {
log.Println("connect: handshake failed") s.log(WarnLevel, pkg+"handshake failed", "error", err.Error())
return errHandshake return errHandshake
} }
if debugMode { s.log(DebugLevel, pkg+"handshaked")
log.Println("... handshaked...")
}
err = sendConnectPacket(s) err = sendConnectPacket(s)
if err != nil { if err != nil {
log.Println("connect: sendConnect failed") s.log(WarnLevel, pkg+"sendConnect failed", "error", err.Error())
return errConnSend return errConnSend
} }
return nil return nil
@ -211,7 +206,7 @@ func connectStream(s *Session) error {
if pkt.packetType == RTMP_PACKET_TYPE_AUDIO || if pkt.packetType == RTMP_PACKET_TYPE_AUDIO ||
pkt.packetType == RTMP_PACKET_TYPE_VIDEO || pkt.packetType == RTMP_PACKET_TYPE_VIDEO ||
pkt.packetType == RTMP_PACKET_TYPE_INFO { pkt.packetType == RTMP_PACKET_TYPE_INFO {
log.Println("connectStream: got packet before play()! Ignoring.") s.log(DebugLevel, pkg+"got packet before play; ignoring")
pkt.body = nil pkt.body = nil
continue continue
} }
@ -241,7 +236,7 @@ func handlePacket(s *Session, pkt *packet) int32 {
s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4])) s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
case RTMP_PACKET_TYPE_CONTROL: case RTMP_PACKET_TYPE_CONTROL:
panic("Unsupported packet type RTMP_PACKET_TYPE_CONTROL") s.log(FatalLevel, "unsupported packet type RTMP_PACKET_TYPE_CONTROL")
case RTMP_PACKET_TYPE_SERVER_BW: case RTMP_PACKET_TYPE_SERVER_BW:
s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4])) s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
@ -255,34 +250,30 @@ func handlePacket(s *Session, pkt *packet) int32 {
} }
case RTMP_PACKET_TYPE_AUDIO: case RTMP_PACKET_TYPE_AUDIO:
panic("Unsupported packet type RTMP_PACKET_TYPE_AUDIO") s.log(FatalLevel, "unsupported packet type RTMP_PACKET_TYPE_AUDIO")
case RTMP_PACKET_TYPE_VIDEO: case RTMP_PACKET_TYPE_VIDEO:
panic("Unsupported packet type RTMP_PACKET_TYPE_VIDEO") s.log(FatalLevel, "unsupported packet type RTMP_PACKET_TYPE_VIDEO")
case RTMP_PACKET_TYPE_FLEX_MESSAGE: case RTMP_PACKET_TYPE_FLEX_MESSAGE:
panic("Unsupported packet type RTMP_PACKET_TYPE_FLEX_MESSAGE") s.log(FatalLevel, "unsupported packet type RTMP_PACKET_TYPE_FLEX_MESSAGE")
case RTMP_PACKET_TYPE_INFO: case RTMP_PACKET_TYPE_INFO:
panic("Unsupported packet type RTMP_PACKET_TYPE_INFO") s.log(FatalLevel, "unsupported packet type RTMP_PACKET_TYPE_INFO")
case RTMP_PACKET_TYPE_INVOKE: case RTMP_PACKET_TYPE_INVOKE:
if debugMode {
log.Println("RTMP_PACKET_TYPE_INVOKE:")
}
err := handleInvoke(s, pkt.body[:pkt.bodySize]) err := handleInvoke(s, pkt.body[:pkt.bodySize])
if err != nil { if err != nil {
// This will never happen with the methods we implement. // This will never happen with the methods we implement.
log.Println("HasMediaPacket") s.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error())
hasMediaPacket = 2 hasMediaPacket = 2
} }
case RTMP_PACKET_TYPE_FLASH_VIDEO: case RTMP_PACKET_TYPE_FLASH_VIDEO:
panic("Unsupported packet type RTMP_PACKET_TYPE_FLASH_VIDEO") s.log(FatalLevel, "unsupported packet type RTMP_PACKET_TYPE_FLASH_VIDEO")
default: default:
// TODO use new logger here s.log(WarnLevel, pkg+"unknown packet type", "type", pkt.packetType)
// RTMP_Log(RTMP_LOGDEBUG, "%s, unknown packet type received: 0x%02x", __FUNCTION__,pkt.packetType);
} }
return hasMediaPacket return hasMediaPacket
} }
@ -294,9 +285,7 @@ func readN(s *Session, buf []byte) error {
} }
n, err := io.ReadFull(s.link.conn, buf) n, err := io.ReadFull(s.link.conn, buf)
if err != nil { if err != nil {
if debugMode { s.log(WarnLevel, pkg+"read failed", "error", err.Error())
log.Printf("readN error: %v\n", err)
}
s.close() s.close()
return err return err
} }
@ -318,9 +307,7 @@ func writeN(s *Session, buf []byte) error {
} }
_, err = s.link.conn.Write(buf) _, err = s.link.conn.Write(buf)
if err != nil { if err != nil {
if debugMode { s.log(WarnLevel, pkg+"write failed", "error", err.Error())
log.Printf("writeN, RTMP send error: %v\n", err)
}
s.close() s.close()
return err return err
} }
@ -706,9 +693,8 @@ func handleInvoke(s *Session, body []byte) error {
} }
meth := C_AMFProp_GetString(C_AMF_GetProp(&obj, "", 0)) 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)) 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 { switch meth {
case av__result: case av__result:
@ -721,30 +707,27 @@ func handleInvoke(s *Session, body []byte) error {
} }
} }
if methodInvoked == "" { if methodInvoked == "" {
// TODO use new logger here s.log(WarnLevel, pkg+"received result without matching request", "id", txn)
//RTMP_Log(RTMP_LOGDEBUG, "%s, received result id %f without matching request",
//__FUNCTION__, txn);
goto leave goto leave
} }
// TODO use new logger here s.log(DebugLevel, pkg+"received result for method", "id", txn)
//RTMP_Log(RTMP_LOGDEBUG, "%s, received result for method call <%s>", __FUNCTION__,
//methodInvoked.av_val);
switch methodInvoked { switch methodInvoked {
case av_connect: case av_connect:
if s.link.token != "" { if s.link.token != "" {
panic("No support for link token") s.log(FatalLevel, "no support for link token")
} }
if (s.link.protocol & RTMP_FEATURE_WRITE) != 0 { if (s.link.protocol & RTMP_FEATURE_WRITE) != 0 {
sendReleaseStream(s) sendReleaseStream(s)
sendFCPublish(s) sendFCPublish(s)
} else { } else {
panic("Link protocol has no RTMP_FEATURE_WRITE") s.log(FatalLevel, "link protocol has no RTMP_FEATURE_WRITE")
} }
sendCreateStream(s) sendCreateStream(s)
if (s.link.protocol & RTMP_FEATURE_WRITE) == 0 { if (s.link.protocol & RTMP_FEATURE_WRITE) == 0 {
panic("Link protocol has no RTMP_FEATURE_WRITE") s.log(FatalLevel, "link protocol has no RTMP_FEATURE_WRITE")
} }
case av_createStream: case av_createStream:
@ -753,13 +736,12 @@ func handleInvoke(s *Session, body []byte) error {
if s.link.protocol&RTMP_FEATURE_WRITE != 0 { if s.link.protocol&RTMP_FEATURE_WRITE != 0 {
sendPublish(s) sendPublish(s)
} else { } else {
panic("Link protocol has no RTMP_FEATURE_WRITE") s.log(FatalLevel, "link protocol has no RTMP_FEATURE_WRITE")
} }
case av_play, av_publish: case av_play, av_publish:
panic("Unsupported method av_play/av_publish") s.log(FatalLevel, "unsupported method av_play/av_publish")
} }
//C.free(unsafe.Pointer(methodInvoked.av_val))
case av_onBWDone: case av_onBWDone:
if s.bwCheckCounter == 0 { if s.bwCheckCounter == 0 {
@ -767,38 +749,34 @@ func handleInvoke(s *Session, body []byte) error {
} }
case av_onFCUnsubscribe, av_onFCSubscribe: case av_onFCUnsubscribe, av_onFCSubscribe:
panic("Unsupported method av_onFCUnsubscribe/av_onFCSubscribe") s.log(FatalLevel, "unsupported method av_onFCUnsubscribe/av_onFCSubscribe")
case av_ping: case av_ping:
panic("Unsupported method av_ping") s.log(FatalLevel, "unsupported method av_ping")
case av__onbwcheck: case av__onbwcheck:
panic("Unsupported method av_onbwcheck") s.log(FatalLevel, "unsupported method av_onbwcheck")
case av__onbwdone: case av__onbwdone:
panic("Unsupported method av_onbwdone") s.log(FatalLevel, "unsupported method av_onbwdone")
case av_close: case av_close:
panic("Unsupported method av_close") s.log(FatalLevel, "unsupported method av_close")
case av_onStatus: case av_onStatus:
var obj2 C_AMFObject var obj2 C_AMFObject
C_AMFProp_GetObject(C_AMF_GetProp(&obj, "", 3), &obj2) C_AMFProp_GetObject(C_AMF_GetProp(&obj, "", 3), &obj2)
code := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_code, -1)) 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 := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_level, -1)) // Not used.
_ = level s.log(DebugLevel, pkg+"onStatus", "code", code, "level", level)
// TODO use new logger
// RTMP_Log(RTMP_LOGDEBUG, "%s, onStatus: %s", __FUNCTION__, code.av_val);
switch code { switch code {
case av_NetStream_Failed, av_NetStream_Play_Failed, case av_NetStream_Failed, av_NetStream_Play_Failed,
av_NetStream_Play_StreamNotFound, av_NetConnection_Connect_InvalidApp: 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, "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: case av_NetStream_Play_Start, av_NetStream_Play_PublishNotify:
panic("Unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify") s.log(FatalLevel, "unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify")
case av_NetStream_Publish_Start: case av_NetStream_Publish_Start:
s.isPlaying = true s.isPlaying = true
@ -810,20 +788,20 @@ func handleInvoke(s *Session, body []byte) error {
} }
case av_NetStream_Play_Complete, av_NetStream_Play_Stop, av_NetStream_Play_UnpublishNotify: 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, "unsupported method av_NetStream_Play_Complete/av_NetStream_Play_Stop/av_NetStream_Play_UnpublishNotify")
case av_NetStream_Seek_Notify: case av_NetStream_Seek_Notify:
panic("Unsupported method av_netStream_Seek_Notify") s.log(FatalLevel, "unsupported method av_netStream_Seek_Notify")
case av_NetStream_Pause_Notify: case av_NetStream_Pause_Notify:
panic("Unsupported method av_NetStream_Pause_Notify") s.log(FatalLevel, "unsupported method av_NetStream_Pause_Notify")
} }
case av_playlist_ready: case av_playlist_ready:
panic("Unsupported method av_playlist_ready") s.log(FatalLevel, "unsupported method av_playlist_ready")
default: default:
panic(fmt.Sprintf("unknown method: %q", meth)) s.log(FatalLevel, "unknown method", "method", meth)
} }
leave: leave:
C_AMF_Reset(&obj) C_AMF_Reset(&obj)
@ -854,12 +832,9 @@ func handshake(s *Session) error {
return err return err
} }
if debugMode { s.log(DebugLevel, pkg+"handshake", "received", typ[0])
log.Printf("handshake: Type answer: %v\n", typ[0])
}
if typ[0] != clientbuf[0] { if typ[0] != clientbuf[0] {
log.Printf("handshake: type mismatch: client sent %v, server sent: %v\n", s.log(WarnLevel, pkg+"handshake type mismatch", "sent", clientbuf[0], "received", typ)
clientbuf[0], typ)
} }
err = readN(s, serversig[:]) err = readN(s, serversig[:])
if err != nil { if err != nil {
@ -868,10 +843,7 @@ func handshake(s *Session) error {
// decode server response // decode server response
suptime := binary.BigEndian.Uint32(serversig[:4]) suptime := binary.BigEndian.Uint32(serversig[:4])
_ = suptime s.log(DebugLevel, pkg+"server uptime", "uptime", 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 // 2nd part of handshake
err = writeN(s, serversig[:]) err = writeN(s, serversig[:])
@ -885,8 +857,7 @@ func handshake(s *Session) error {
} }
if !bytes.Equal(serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1]) { if !bytes.Equal(serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1]) {
log.Printf("Client signature does not match: %q != %q", s.log(WarnLevel, pkg+"signature mismatch", "serversig", serversig[:RTMP_SIG_SIZE], "clientsig", clientbuf[1:RTMP_SIG_SIZE+1])
serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1])
} }
return nil return nil
} }

26
rtmp/session.go
View File

@ -33,10 +33,6 @@ LICENSE
*/ */
package rtmp package rtmp
import (
"errors"
)
// Session holds the state for an RTMP session. // Session holds the state for an RTMP session.
type Session struct { type Session struct {
url string url string
@ -62,22 +58,36 @@ type Session struct {
audioCodecs float64 audioCodecs float64
videoCodecs float64 videoCodecs float64
encoding float64 encoding float64
defered []byte deferred []byte
link link 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. // NewSession returns a new Session.
func NewSession(url string, connectTimeout uint) *Session { func NewSession(url string, timeout uint, log Log) *Session {
return &Session{ return &Session{
url: url, url: url,
timeout: connectTimeout, timeout: timeout,
log: log,
} }
} }
// Open establishes an rtmp connection with the url passed into the constructor. // Open establishes an rtmp connection with the url passed into the constructor.
func (s *Session) Open() error { func (s *Session) Open() error {
if s.isConnected() { if s.isConnected() {
return errors.New("rtmp: attempt to start already running session") return errConnected
} }
err := s.start() err := s.start()
if err != nil { if err != nil {