diff --git a/revid/senders.go b/revid/senders.go index 9c9abca1..f09d34fa 100644 --- a/revid/senders.go +++ b/revid/senders.go @@ -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 diff --git a/rtmp/packet.go b/rtmp/packet.go index 95efb658..d97d99ea 100644 --- a/rtmp/packet.go +++ b/rtmp/packet.go @@ -36,7 +36,6 @@ package rtmp import ( "encoding/binary" - "log" ) const ( @@ -101,7 +100,7 @@ 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()) return err } pkt.headerType = (header[0] & 0xc0) >> 6 @@ -112,7 +111,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 +120,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 { @@ -161,22 +159,19 @@ func readPacket(s *Session, pkt *packet) error { *pkt = *(s.vecChannelsIn[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 +190,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,7 +217,7 @@ 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 } @@ -254,7 +249,7 @@ func readPacket(s *Session, pkt *packet) error { 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 @@ -300,8 +295,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 } @@ -339,7 +333,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 @@ -396,39 +390,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", "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr(), "size", size) - // 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,10 +456,7 @@ 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) - } + s.log(DebugLevel, "invoking method", "method", meth) // keep it in call queue till result arrives if queue { buf = buf[3+len(meth):] diff --git a/rtmp/rtmp.go b/rtmp/rtmp.go index d73c2fc2..2574a785 100644 --- a/rtmp/rtmp.go +++ b/rtmp/rtmp.go @@ -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 @@ -211,7 +206,7 @@ func connectStream(s *Session) error { 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.") + s.log(DebugLevel, pkg+"got packet before play; ignoring") pkt.body = nil continue } @@ -241,7 +236,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, "unsupported packet type RTMP_PACKET_TYPE_CONTROL") case RTMP_PACKET_TYPE_SERVER_BW: 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: - panic("Unsupported packet type RTMP_PACKET_TYPE_AUDIO") + s.log(FatalLevel, "unsupported packet type RTMP_PACKET_TYPE_AUDIO") 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: - 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: - panic("Unsupported packet type RTMP_PACKET_TYPE_INFO") + s.log(FatalLevel, "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") + s.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error()) hasMediaPacket = 2 } 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: - // 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 } @@ -294,9 +285,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.log(WarnLevel, pkg+"read failed", "error", err.Error()) s.close() return err } @@ -318,9 +307,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.log(WarnLevel, pkg+"write failed", "error", err.Error()) s.close() return err } @@ -706,9 +693,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 +707,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 method", "id", txn) + switch methodInvoked { case av_connect: 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 { sendReleaseStream(s) sendFCPublish(s) } else { - panic("Link protocol has no RTMP_FEATURE_WRITE") + s.log(FatalLevel, "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, "link protocol has no RTMP_FEATURE_WRITE") } case av_createStream: @@ -753,13 +736,12 @@ 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, "link protocol has no RTMP_FEATURE_WRITE") } 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: if s.bwCheckCounter == 0 { @@ -767,38 +749,34 @@ func handleInvoke(s *Session, body []byte) error { } case av_onFCUnsubscribe, av_onFCSubscribe: - panic("Unsupported method av_onFCUnsubscribe/av_onFCSubscribe") + s.log(FatalLevel, "unsupported method av_onFCUnsubscribe/av_onFCSubscribe") case av_ping: - panic("Unsupported method av_ping") + s.log(FatalLevel, "unsupported method av_ping") case av__onbwcheck: - panic("Unsupported method av_onbwcheck") + s.log(FatalLevel, "unsupported method av_onbwcheck") case av__onbwdone: - panic("Unsupported method av_onbwdone") + s.log(FatalLevel, "unsupported method av_onbwdone") case av_close: - panic("Unsupported method av_close") + s.log(FatalLevel, "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); + 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, "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, "unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify") case av_NetStream_Publish_Start: 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: - 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: - panic("Unsupported method av_netStream_Seek_Notify") + s.log(FatalLevel, "unsupported method av_netStream_Seek_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: - panic("Unsupported method av_playlist_ready") + s.log(FatalLevel, "unsupported method av_playlist_ready") default: - panic(fmt.Sprintf("unknown method: %q", meth)) + s.log(FatalLevel, "unknown method", "method", meth) } leave: C_AMF_Reset(&obj) @@ -854,12 +832,9 @@ func handshake(s *Session) error { return err } - if debugMode { - log.Printf("handshake: Type answer: %v\n", typ[0]) - } + s.log(DebugLevel, pkg+"handshake", "received", 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 +843,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 +857,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 } diff --git a/rtmp/session.go b/rtmp/session.go index 6d1fad40..a04f39d5 100644 --- a/rtmp/session.go +++ b/rtmp/session.go @@ -33,10 +33,6 @@ LICENSE */ package rtmp -import ( - "errors" -) - // Session holds the state for an RTMP session. type Session struct { url string @@ -62,22 +58,36 @@ type Session struct { 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, + timeout: timeout, + log: log, } } // 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 {