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

Merged in rtmp-refactoring-3 (pull request #119) 

Rtmp refactoring 3
This commit is contained in:
Alan Noble 2019-01-19 07:56:35 +00:00
parent 35d63f664f 6959243d50
commit 54d351a55e
5 changed files with 318 additions and 372 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
revid/senders.go
View File

@ -251,7 +251,7 @@ func (s *ffmpegSender) close() error {
// rtmpSender implements loadSender for a native RTMP destination.
type rtmpSender struct {
sess *rtmp.Session
conn *rtmp.Conn
url string
timeout uint
@ -264,16 +264,15 @@ type rtmpSender struct {
var _ restarter = (*rtmpSender)(nil)
func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg string, args ...interface{})) (*rtmpSender, error) {
var sess *rtmp.Session
var conn *rtmp.Conn
var err error
for n := 0; n < retries; n++ {
sess = rtmp.NewSession(url, timeout, log)
err = sess.Open()
conn, err = rtmp.Dial(url, timeout, log)
if err == nil {
break
}
log(logger.Error, err.Error())
sess.Close()
conn.Close()
if n < retries-1 {
log(logger.Info, pkg+"retry rtmp connection")
}
@ -283,7 +282,7 @@ func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg
}
s := &rtmpSender{
sess: sess,
conn: conn,
url: url,
timeout: timeout,
retries: retries,
@ -298,7 +297,7 @@ func (s *rtmpSender) load(c *ring.Chunk) error {
}
func (s *rtmpSender) send() error {
_, err := s.chunk.WriteTo(s.sess)
_, err := s.chunk.WriteTo(s.conn)
return err
}
@ -308,18 +307,17 @@ func (s *rtmpSender) release() {
}
func (s *rtmpSender) restart() error {
err := s.sess.Close()
err := s.conn.Close()
if err != nil {
return err
}
for n := 0; n < s.retries; n++ {
s.sess = rtmp.NewSession(s.url, s.timeout, s.log)
err = s.sess.Open()
s.conn, err = rtmp.Dial(s.url, s.timeout, s.log)
if err == nil {
break
}
s.log(logger.Error, err.Error())
s.sess.Close()
s.conn.Close()
if n < s.retries-1 {
s.log(logger.Info, pkg+"retry rtmp connection")
}
@ -328,7 +326,7 @@ func (s *rtmpSender) restart() error {
}
func (s *rtmpSender) close() error {
return s.sess.Close()
return s.conn.Close()
}
// udpSender implements loadSender for a native udp destination.

135
rtmp/session.go → rtmp/conn.go
View File

@ -1,9 +1,9 @@
/*
NAME
session.go
conn.go
DESCRIPTION
RTMP session functionality.
RTMP connection functionality.
AUTHORS
Saxon Nelson-Milton <saxon@ausocean.org>
@ -11,7 +11,7 @@ AUTHORS
Alan Noble <alan@ausocean.org>
LICENSE
session.go is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean)
conn.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
@ -36,19 +36,19 @@ package rtmp
import (
"io"
"net"
"strconv"
"time"
"bitbucket.org/ausocean/av/rtmp/amf"
)
// Session holds the state for an RTMP session.
type Session struct {
url string
// Conn represents an RTMP connection.
type Conn struct {
inChunkSize uint32
outChunkSize uint32
nBytesIn uint32
nBytesInSent uint32
streamID int32
streamID uint32
serverBW uint32
clientBW uint32
clientBW2 uint8
@ -60,8 +60,6 @@ type Session struct {
channelsIn []*packet
channelsOut []*packet
channelTimestamp []int32
audioCodecs float64
videoCodecs float64
deferred []byte
link link
log Log
@ -103,69 +101,71 @@ const (
// NB: We don't accept extended headers.
const flvTagheaderSize = 11
// NewSession returns a new Session.
func NewSession(url string, timeout uint, log Log) *Session {
return &Session{
url: url,
// Dial connects to RTMP server specified by the given URL and returns the connection.
func Dial(url string, timeout uint, log Log) (*Conn, error) {
log(DebugLevel, pkg+"rtmp.Dial")
c := Conn{
inChunkSize: 128,
outChunkSize: 128,
clientBW: 2500000,
clientBW2: 2,
serverBW: 2500000,
audioCodecs: 3191.0,
videoCodecs: 252.0,
log: log,
link: link{
timeout: timeout,
},
}
var err error
c.link.protocol, c.link.host, c.link.port, c.link.app, c.link.playpath, err = parseURL(url)
if err != nil {
return nil, err
}
if c.link.app == "" {
return nil, errInvalidURL
}
if c.link.port == 0 {
switch {
case (c.link.protocol & featureSSL) != 0:
c.link.port = 433
c.log(FatalLevel, pkg+"SSL not supported")
case (c.link.protocol & featureHTTP) != 0:
c.link.port = 80
default:
c.link.port = 1935
}
}
c.link.url = rtmpProtocolStrings[c.link.protocol] + "://" + c.link.host + ":" + strconv.Itoa(int(c.link.port)) + "/" + c.link.app
c.link.protocol |= featureWrite
err = connect(&c)
if err != nil {
return nil, err
}
return &c, nil
}
// Open establishes an rtmp connection with the url passed into the constructor.
func (s *Session) Open() error {
s.log(DebugLevel, pkg+"Session.Open")
if s.isConnected() {
return errConnected
}
err := s.init()
if err != nil {
return err
}
err = connect(s)
if err != nil {
s.Close()
return err
}
err = connectStream(s)
if err != nil {
s.Close()
return err
}
return nil
}
// Close terminates the rtmp connection.
// NB: Close is idempotent and the session value is cleared completely.
func (s *Session) Close() error {
s.log(DebugLevel, pkg+"Session.Close")
if !s.isConnected() {
// Close terminates the RTMP connection.
// NB: Close is idempotent and the connection value is cleared completely.
func (c *Conn) Close() error {
c.log(DebugLevel, pkg+"Conn.Close")
if !c.isConnected() {
return errNotConnected
}
if s.streamID > 0 {
if s.link.protocol&featureWrite != 0 {
sendFCUnpublish(s)
if c.streamID > 0 {
if c.link.protocol&featureWrite != 0 {
sendFCUnpublish(c)
}
sendDeleteStream(s, float64(s.streamID))
sendDeleteStream(c, float64(c.streamID))
}
s.link.conn.Close()
*s = Session{}
c.link.conn.Close()
*c = Conn{}
return nil
}
// Write writes a frame (flv tag) to the rtmp connection.
func (s *Session) Write(data []byte) (int, error) {
if !s.isConnected() {
func (c *Conn) Write(data []byte) (int, error) {
if !c.isConnected() {
return 0, errNotConnected
}
if len(data) < flvTagheaderSize {
@ -180,12 +180,12 @@ func (s *Session) Write(data []byte) (int, error) {
bodySize: amf.DecodeInt24(data[1:4]),
timestamp: amf.DecodeInt24(data[4:7]) | uint32(data[7])<<24,
channel: chanSource,
info: s.streamID,
streamID: c.streamID,
}
pkt.resize(pkt.bodySize, headerSizeAuto)
copy(pkt.body, data[flvTagheaderSize:flvTagheaderSize+pkt.bodySize])
err := pkt.writeTo(s, false)
err := pkt.writeTo(c, false)
if err != nil {
return 0, err
}
@ -197,19 +197,19 @@ func (s *Session) Write(data []byte) (int, error) {
// read from an RTMP connection. Sends a bytes received message if the
// number of bytes received (nBytesIn) is greater than the number sent
// (nBytesInSent) by 10% of the bandwidth.
func (s *Session) read(buf []byte) (int, error) {
err := s.link.conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(s.link.timeout)))
func (c *Conn) read(buf []byte) (int, error) {
err := c.link.conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(c.link.timeout)))
if err != nil {
return 0, err
}
n, err := io.ReadFull(s.link.conn, buf)
n, err := io.ReadFull(c.link.conn, buf)
if err != nil {
s.log(DebugLevel, pkg+"read failed", "error", err.Error())
c.log(DebugLevel, pkg+"read failed", "error", err.Error())
return 0, err
}
s.nBytesIn += uint32(n)
if s.nBytesIn > (s.nBytesInSent + s.clientBW/10) {
err := sendBytesReceived(s)
c.nBytesIn += uint32(n)
if c.nBytesIn > (c.nBytesInSent + c.clientBW/10) {
err := sendBytesReceived(c)
if err != nil {
return n, err // NB: we still read n bytes, even though send bytes failed
}
@ -218,22 +218,21 @@ func (s *Session) read(buf []byte) (int, error) {
}
// write to an RTMP connection.
func (s *Session) write(buf []byte) (int, error) {
func (c *Conn) write(buf []byte) (int, error) {
//ToDo: consider using a different timeout for writes than for reads
err := s.link.conn.SetWriteDeadline(time.Now().Add(time.Second * time.Duration(s.link.timeout)))
err := c.link.conn.SetWriteDeadline(time.Now().Add(time.Second * time.Duration(c.link.timeout)))
if err != nil {
return 0, err
}
n, err := s.link.conn.Write(buf)
n, err := c.link.conn.Write(buf)
if err != nil {
s.log(WarnLevel, pkg+"write failed", "error", err.Error())
c.log(WarnLevel, pkg+"write failed", "error", err.Error())
return 0, err
}
return n, nil
}
// isConnected returns true if the RTMP connection is up.
func (s *Session) isConnected() bool {
return s.link.conn != nil
func (c *Conn) isConnected() bool {
return c.link.conn != nil
}

183
rtmp/packet.go
View File

@ -88,23 +88,23 @@ type packet struct {
channel int32
hasAbsTimestamp bool
timestamp uint32
info int32
streamID uint32
bodySize uint32
bytesRead uint32
header []byte
buf []byte
body []byte
}
// readFrom reads a packet from the RTMP connection.
func (pkt *packet) readFrom(s *Session) error {
func (pkt *packet) readFrom(c *Conn) error {
var hbuf [fullHeaderSize]byte
header := hbuf[:]
_, err := s.read(header[:1])
_, err := c.read(header[:1])
if err != nil {
s.log(DebugLevel, pkg+"failed to read packet header 1st byte", "error", err.Error())
c.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")
c.log(WarnLevel, pkg+"EOF error; connection likely terminated")
}
return err
}
@ -114,45 +114,45 @@ func (pkt *packet) readFrom(s *Session) error {
switch {
case pkt.channel == 0:
_, err = s.read(header[:1])
_, err = c.read(header[:1])
if err != nil {
s.log(DebugLevel, pkg+"failed to read packet header 2nd byte", "error", err.Error())
c.log(DebugLevel, pkg+"failed to read packet header 2nd byte", "error", err.Error())
return err
}
header = header[1:]
pkt.channel = int32(header[0]) + 64
case pkt.channel == 1:
_, err = s.read(header[:2])
_, err = c.read(header[:2])
if err != nil {
s.log(DebugLevel, pkg+"failed to read packet header 3rd byte", "error", err.Error())
c.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 {
if pkt.channel >= c.channelsAllocatedIn {
n := pkt.channel + 10
timestamp := append(s.channelTimestamp, make([]int32, 10)...)
timestamp := append(c.channelTimestamp, make([]int32, 10)...)
var pkts []*packet
if s.channelsIn == nil {
if c.channelsIn == nil {
pkts = make([]*packet, n)
} else {
pkts = append(s.channelsIn[:pkt.channel:pkt.channel], make([]*packet, 10)...)
pkts = append(c.channelsIn[:pkt.channel:pkt.channel], make([]*packet, 10)...)
}
s.channelTimestamp = timestamp
s.channelsIn = pkts
c.channelTimestamp = timestamp
c.channelsIn = pkts
for i := int(s.channelsAllocatedIn); i < len(s.channelTimestamp); i++ {
s.channelTimestamp[i] = 0
for i := int(c.channelsAllocatedIn); i < len(c.channelTimestamp); i++ {
c.channelTimestamp[i] = 0
}
for i := int(s.channelsAllocatedIn); i < int(n); i++ {
s.channelsIn[i] = nil
for i := int(c.channelsAllocatedIn); i < int(n); i++ {
c.channelsIn[i] = nil
}
s.channelsAllocatedIn = n
c.channelsAllocatedIn = n
}
size := headerSizes[pkt.headerType]
@ -160,16 +160,16 @@ func (pkt *packet) readFrom(s *Session) error {
case size == fullHeaderSize:
pkt.hasAbsTimestamp = true
case size < fullHeaderSize:
if s.channelsIn[pkt.channel] != nil {
*pkt = *(s.channelsIn[pkt.channel])
if c.channelsIn[pkt.channel] != nil {
*pkt = *(c.channelsIn[pkt.channel])
}
}
size--
if size > 0 {
_, err = s.read(header[:size])
_, err = c.read(header[:size])
if err != nil {
s.log(DebugLevel, pkg+"failed to read packet header", "error", err.Error())
c.log(DebugLevel, pkg+"failed to read packet header", "error", err.Error())
return err
}
}
@ -184,7 +184,7 @@ func (pkt *packet) readFrom(s *Session) error {
if size > 6 {
pkt.packetType = header[6]
if size == 11 {
pkt.info = int32(amf.DecodeInt32LE(header[7:11]))
pkt.streamID = amf.DecodeInt32LE(header[7:11])
}
}
}
@ -192,9 +192,9 @@ func (pkt *packet) readFrom(s *Session) error {
extendedTimestamp := pkt.timestamp == 0xffffff
if extendedTimestamp {
_, err = s.read(header[size : size+4])
_, err = c.read(header[size : size+4])
if err != nil {
s.log(DebugLevel, pkg+"failed to read extended timestamp", "error", err.Error())
c.log(DebugLevel, pkg+"failed to read extended timestamp", "error", err.Error())
return err
}
pkt.timestamp = amf.DecodeInt32(header[size : size+4])
@ -206,48 +206,47 @@ func (pkt *packet) readFrom(s *Session) error {
}
toRead := pkt.bodySize - pkt.bytesRead
chunkSize := s.inChunkSize
chunkSize := c.inChunkSize
if toRead < chunkSize {
chunkSize = toRead
}
_, err = s.read(pkt.body[pkt.bytesRead:][:chunkSize])
_, err = c.read(pkt.body[pkt.bytesRead:][:chunkSize])
if err != nil {
s.log(DebugLevel, pkg+"failed to read packet body", "error", err.Error())
c.log(DebugLevel, pkg+"failed to read packet body", "error", err.Error())
return err
}
pkt.bytesRead += uint32(chunkSize)
// Keep the packet as a reference for other packets on this channel.
if s.channelsIn[pkt.channel] == nil {
s.channelsIn[pkt.channel] = &packet{}
if c.channelsIn[pkt.channel] == nil {
c.channelsIn[pkt.channel] = &packet{}
}
*(s.channelsIn[pkt.channel]) = *pkt
*(c.channelsIn[pkt.channel]) = *pkt
if extendedTimestamp {
s.channelsIn[pkt.channel].timestamp = 0xffffff
c.channelsIn[pkt.channel].timestamp = 0xffffff
}
if !pkt.hasAbsTimestamp {
// Timestamps seem to always be relative.
pkt.timestamp += uint32(s.channelTimestamp[pkt.channel])
pkt.timestamp += uint32(c.channelTimestamp[pkt.channel])
}
s.channelTimestamp[pkt.channel] = int32(pkt.timestamp)
c.channelTimestamp[pkt.channel] = int32(pkt.timestamp)
s.channelsIn[pkt.channel].body = nil
s.channelsIn[pkt.channel].bytesRead = 0
s.channelsIn[pkt.channel].hasAbsTimestamp = false
c.channelsIn[pkt.channel].body = nil
c.channelsIn[pkt.channel].bytesRead = 0
c.channelsIn[pkt.channel].hasAbsTimestamp = false
return nil
}
// resize adjusts the packet's storage to accommodate a body of the given size and header type.
// When headerSizeAuto is specified, the header type is computed based on packet type.
func (pkt *packet) resize(size uint32, ht uint8) {
buf := make([]byte, fullHeaderSize+size)
pkt.header = buf
pkt.body = buf[fullHeaderSize:]
pkt.buf = make([]byte, fullHeaderSize+size)
pkt.body = pkt.buf[fullHeaderSize:]
if ht != headerSizeAuto {
pkt.headerType = ht
return
@ -270,32 +269,32 @@ func (pkt *packet) resize(size uint32, ht uint8) {
// writeTo writes a packet to the RTMP connection.
// Packets are written in chunks which are Session.chunkSize in length (128 bytes in length).
// We defer sending small audio packets and combine consecutive small audio packets where possible to reduce I/O.
// When queue is true, we expect a response to this request and cache the method on s.methodCalls.
func (pkt *packet) writeTo(s *Session, queue bool) error {
// When queue is true, we expect a response to this request and cache the method on c.methodCalls.
func (pkt *packet) writeTo(c *Conn, queue bool) error {
if pkt.body == nil || pkt.bodySize == 0 {
return errInvalidBody
}
if pkt.channel >= s.channelsAllocatedOut {
s.log(DebugLevel, pkg+"growing channelsOut", "channel", pkt.channel)
if pkt.channel >= c.channelsAllocatedOut {
c.log(DebugLevel, pkg+"growing channelsOut", "channel", pkt.channel)
n := int(pkt.channel + 10)
var pkts []*packet
if s.channelsOut == nil {
if c.channelsOut == nil {
pkts = make([]*packet, n)
} else {
pkts = append(s.channelsOut[:pkt.channel:pkt.channel], make([]*packet, 10)...)
pkts = append(c.channelsOut[:pkt.channel:pkt.channel], make([]*packet, 10)...)
}
s.channelsOut = pkts
c.channelsOut = pkts
for i := int(s.channelsAllocatedOut); i < n; i++ {
s.channelsOut[i] = nil
for i := int(c.channelsAllocatedOut); i < n; i++ {
c.channelsOut[i] = nil
}
s.channelsAllocatedOut = int32(n)
c.channelsAllocatedOut = int32(n)
}
prevPkt := s.channelsOut[pkt.channel]
prevPkt := c.channelsOut[pkt.channel]
var last int
if prevPkt != nil && pkt.headerType != headerSizeLarge {
// Compress header by using the previous packet's attributes.
@ -311,13 +310,13 @@ func (pkt *packet) writeTo(s *Session, queue bool) error {
}
if pkt.headerType > 3 {
s.log(WarnLevel, pkg+"unexpected header type", "type", pkt.headerType)
c.log(WarnLevel, pkg+"unexpected header type", "type", pkt.headerType)
return errInvalidHeader
}
// The complete packet starts from headerSize _before_ the start the body.
// origIdx is the original offset, which will be 0 for a full (12-byte) header or 11 for a minimum (1-byte) header.
headBytes := pkt.header
buf := pkt.buf
hSize := headerSizes[pkt.headerType]
origIdx := fullHeaderSize - hSize
@ -343,30 +342,30 @@ func (pkt *packet) writeTo(s *Session, queue bool) error {
if ts >= 0xffffff {
origIdx -= 4
hSize += 4
s.log(DebugLevel, pkg+"larger timestamp than 24 bits", "timestamp", ts)
c.log(DebugLevel, pkg+"larger timestamp than 24 bits", "timestamp", ts)
}
headerIdx := origIdx
c := pkt.headerType << 6
ch := pkt.headerType << 6
switch cSize {
case 0:
c |= byte(pkt.channel)
ch |= byte(pkt.channel)
case 1:
// Do nothing.
case 2:
c |= 1
ch |= 1
}
headBytes[headerIdx] = c
buf[headerIdx] = ch
headerIdx++
if cSize != 0 {
tmp := pkt.channel - 64
headBytes[headerIdx] = byte(tmp & 0xff)
buf[headerIdx] = byte(tmp & 0xff)
headerIdx++
if cSize == 2 {
headBytes[headerIdx] = byte(tmp >> 8)
buf[headerIdx] = byte(tmp >> 8)
headerIdx++
}
}
@ -376,66 +375,66 @@ func (pkt *packet) writeTo(s *Session, queue bool) error {
if ts > 0xffffff {
tmp = 0xffffff
}
amf.EncodeInt24(headBytes[headerIdx:], tmp)
amf.EncodeInt24(buf[headerIdx:], tmp)
headerIdx += 3 // 24bits
}
if headerSizes[pkt.headerType] > 4 {
amf.EncodeInt24(headBytes[headerIdx:], pkt.bodySize)
amf.EncodeInt24(buf[headerIdx:], pkt.bodySize)
headerIdx += 3 // 24bits
headBytes[headerIdx] = pkt.packetType
buf[headerIdx] = pkt.packetType
headerIdx++
}
if headerSizes[pkt.headerType] > 8 {
binary.LittleEndian.PutUint32(headBytes[headerIdx:headerIdx+4], uint32(pkt.info))
binary.LittleEndian.PutUint32(buf[headerIdx:headerIdx+4], pkt.streamID)
headerIdx += 4 // 32bits
}
if ts >= 0xffffff {
amf.EncodeInt32(headBytes[headerIdx:], ts)
amf.EncodeInt32(buf[headerIdx:], ts)
headerIdx += 4 // 32bits
}
size := int(pkt.bodySize)
chunkSize := int(s.outChunkSize)
chunkSize := int(c.outChunkSize)
if s.deferred == nil {
if c.deferred == nil {
// Defer sending small audio packets (at most once).
if pkt.packetType == packetTypeAudio && size < chunkSize {
s.deferred = headBytes[origIdx:][:size+hSize]
s.log(DebugLevel, pkg+"deferred sending packet", "size", size, "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr())
c.deferred = buf[origIdx:][:size+hSize]
c.log(DebugLevel, pkg+"deferred sending packet", "size", size, "la", c.link.conn.LocalAddr(), "ra", c.link.conn.RemoteAddr())
return nil
}
} else {
// Send previously deferrd packet if combining it with the next one would exceed the chunk size.
if len(s.deferred)+size+hSize > chunkSize {
s.log(DebugLevel, pkg+"sending deferred packet separately", "size", len(s.deferred))
_, err := s.write(s.deferred)
if len(c.deferred)+size+hSize > chunkSize {
c.log(DebugLevel, pkg+"sending deferred packet separately", "size", len(c.deferred))
_, err := c.write(c.deferred)
if err != nil {
return err
}
s.deferred = nil
c.deferred = nil
}
}
// TODO(kortschak): Rewrite this horrific peice of premature optimisation.
s.log(DebugLevel, pkg+"sending packet", "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr(), "size", size)
c.log(DebugLevel, pkg+"sending packet", "la", c.link.conn.LocalAddr(), "ra", c.link.conn.RemoteAddr(), "size", size)
for size+hSize != 0 {
if chunkSize > size {
chunkSize = size
}
bytes := headBytes[origIdx:][:chunkSize+hSize]
if s.deferred != nil {
bytes := buf[origIdx:][:chunkSize+hSize]
if c.deferred != nil {
// Prepend the previously deferred packet and write it with the current one.
s.log(DebugLevel, pkg+"combining deferred packet", "size", len(s.deferred))
bytes = append(s.deferred, bytes...)
c.log(DebugLevel, pkg+"combining deferred packet", "size", len(c.deferred))
bytes = append(c.deferred, bytes...)
}
_, err := s.write(bytes)
_, err := c.write(bytes)
if err != nil {
return err
}
s.deferred = nil
c.deferred = nil
size -= chunkSize
origIdx += chunkSize + hSize
@ -451,18 +450,18 @@ func (pkt *packet) writeTo(s *Session, queue bool) error {
hSize += 4
}
headBytes[origIdx] = 0xc0 | c
buf[origIdx] = 0xc0 | ch
if cSize != 0 {
tmp := int(pkt.channel) - 64
headBytes[origIdx+1] = byte(tmp)
buf[origIdx+1] = byte(tmp)
if cSize == 2 {
headBytes[origIdx+2] = byte(tmp >> 8)
buf[origIdx+2] = byte(tmp >> 8)
}
}
if ts >= 0xffffff {
extendedTimestamp := headBytes[origIdx+1+cSize:]
extendedTimestamp := buf[origIdx+1+cSize:]
amf.EncodeInt32(extendedTimestamp[:4], ts)
}
}
@ -472,16 +471,16 @@ func (pkt *packet) writeTo(s *Session, queue bool) error {
if pkt.packetType == packetTypeInvoke && queue {
buf := pkt.body[1:]
meth := amf.DecodeString(buf)
s.log(DebugLevel, pkg+"queuing method "+meth)
c.log(DebugLevel, pkg+"queuing method "+meth)
buf = buf[3+len(meth):]
txn := int32(amf.DecodeNumber(buf[:8]))
s.methodCalls = append(s.methodCalls, method{name: meth, num: txn})
c.methodCalls = append(c.methodCalls, method{name: meth, num: txn})
}
if s.channelsOut[pkt.channel] == nil {
s.channelsOut[pkt.channel] = &packet{}
if c.channelsOut[pkt.channel] == nil {
c.channelsOut[pkt.channel] = &packet{}
}
*(s.channelsOut[pkt.channel]) = *pkt
*(c.channelsOut[pkt.channel]) = *pkt
return nil
}

264
rtmp/rtmp.go
View File

@ -162,79 +162,49 @@ var (
errUnimplemented = errors.New("rtmp: unimplemented feature")
)
// init initialises the Session link
func (s *Session) init() (err error) {
s.link.protocol, s.link.host, s.link.port, s.link.app, s.link.playpath, err = parseURL(s.url)
if err != nil {
return err
}
if s.link.app == "" {
return errInvalidURL
}
if s.link.port == 0 {
switch {
case (s.link.protocol & featureSSL) != 0:
s.link.port = 433
s.log(FatalLevel, pkg+"SSL not supported")
case (s.link.protocol & featureHTTP) != 0:
s.link.port = 80
default:
s.link.port = 1935
}
}
s.link.url = rtmpProtocolStrings[s.link.protocol] + "://" + s.link.host + ":" + strconv.Itoa(int(s.link.port)) + "/" + s.link.app
s.link.protocol |= featureWrite
return nil
}
// connect establishes an RTMP connection.
func connect(s *Session) error {
addr, err := net.ResolveTCPAddr("tcp4", s.link.host+":"+strconv.Itoa(int(s.link.port)))
func connect(c *Conn) error {
addr, err := net.ResolveTCPAddr("tcp4", c.link.host+":"+strconv.Itoa(int(c.link.port)))
if err != nil {
return err
}
s.link.conn, err = net.DialTCP("tcp4", nil, addr)
c.link.conn, err = net.DialTCP("tcp4", nil, addr)
if err != nil {
s.log(WarnLevel, pkg+"dial failed", "error", err.Error())
c.log(WarnLevel, pkg+"dial failed", "error", err.Error())
return err
}
s.log(DebugLevel, pkg+"connected")
err = handshake(s)
c.log(DebugLevel, pkg+"connected")
err = handshake(c)
if err != nil {
s.log(WarnLevel, pkg+"handshake failed", "error", err.Error())
c.log(WarnLevel, pkg+"handshake failed", "error", err.Error())
return err
}
s.log(DebugLevel, pkg+"handshaked")
err = sendConnectPacket(s)
c.log(DebugLevel, pkg+"handshaked")
err = sendConnectPacket(c)
if err != nil {
s.log(WarnLevel, pkg+"sendConnect failed", "error", err.Error())
c.log(WarnLevel, pkg+"sendConnect failed", "error", err.Error())
return err
}
return nil
}
// connectStream reads a packet and handles it
func connectStream(s *Session) error {
var err error
for !s.isPlaying {
c.log(DebugLevel, pkg+"negotiating")
for !c.isPlaying {
pkt := packet{}
err = pkt.readFrom(s)
err = pkt.readFrom(c)
if err != nil {
break
}
switch pkt.packetType {
case packetTypeAudio, packetTypeVideo, packetTypeInfo:
s.log(WarnLevel, pkg+"got packet before play; ignoring", "type", pkt.packetType)
c.log(WarnLevel, pkg+"got packet before play; ignoring", "type", pkt.packetType)
default:
err = handlePacket(s, &pkt)
err = handlePacket(c, &pkt)
if err != nil {
break
}
}
}
if !s.isPlaying {
if !c.isPlaying {
return err
}
return nil
@ -242,52 +212,56 @@ func connectStream(s *Session) error {
// handlePacket handles a packet that the client has received.
// NB: Unsupported packet types are logged fatally.
func handlePacket(s *Session, pkt *packet) error {
func handlePacket(c *Conn, pkt *packet) error {
if pkt.bodySize < 4 {
return errInvalidBody
}
switch pkt.packetType {
case packetTypeChunkSize:
s.inChunkSize = amf.DecodeInt32(pkt.body[:4])
c.inChunkSize = amf.DecodeInt32(pkt.body[:4])
c.log(DebugLevel, pkg+"set inChunkSize", "size", int(c.inChunkSize))
case packetTypeBytesReadReport:
s.serverBW = amf.DecodeInt32(pkt.body[:4])
c.log(DebugLevel, pkg+"received packetTypeBytesReadReport")
case packetTypeServerBW:
s.serverBW = amf.DecodeInt32(pkt.body[:4])
c.serverBW = amf.DecodeInt32(pkt.body[:4])
c.log(DebugLevel, pkg+"set serverBW", "size", int(c.serverBW))
case packetTypeClientBW:
s.clientBW = amf.DecodeInt32(pkt.body[:4])
c.clientBW = amf.DecodeInt32(pkt.body[:4])
c.log(DebugLevel, pkg+"set clientBW", "size", int(c.clientBW))
if pkt.bodySize > 4 {
s.clientBW2 = pkt.body[4]
c.clientBW2 = pkt.body[4]
c.log(DebugLevel, pkg+"set clientBW2", "size", int(c.clientBW2))
} else {
s.clientBW2 = 0xff
c.clientBW2 = 0xff
}
case packetTypeInvoke:
err := handleInvoke(s, pkt.body[:pkt.bodySize])
err := handleInvoke(c, pkt.body[:pkt.bodySize])
if err != nil {
s.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error())
c.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error())
return err
}
case packetTypeControl, packetTypeAudio, packetTypeVideo, packetTypeFlashVideo, packetTypeFlexMessage, packetTypeInfo:
s.log(FatalLevel, pkg+"unsupported packet type "+strconv.Itoa(int(pkt.packetType)))
c.log(FatalLevel, pkg+"unsupported packet type "+strconv.Itoa(int(pkt.packetType)))
default:
s.log(WarnLevel, pkg+"unknown packet type", "type", pkt.packetType)
c.log(WarnLevel, pkg+"unknown packet type", "type", pkt.packetType)
}
return nil
}
func sendConnectPacket(s *Session) error {
func sendConnectPacket(c *Conn) error {
var pbuf [4096]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeLarge,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -296,15 +270,15 @@ func sendConnectPacket(s *Session) error {
if err != nil {
return err
}
s.numInvokes += 1
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes += 1
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
enc[0] = amf.TypeObject
enc = enc[1:]
enc, err = amf.EncodeNamedString(enc, avApp, s.link.app)
enc, err = amf.EncodeNamedString(enc, avApp, c.link.app)
if err != nil {
return err
}
@ -312,7 +286,7 @@ func sendConnectPacket(s *Session) error {
if err != nil {
return err
}
enc, err = amf.EncodeNamedString(enc, avTcUrl, s.link.url)
enc, err = amf.EncodeNamedString(enc, avTcUrl, c.link.url)
if err != nil {
return err
}
@ -322,12 +296,12 @@ func sendConnectPacket(s *Session) error {
}
// add auth string, if any
if s.link.auth != "" {
enc, err = amf.EncodeBoolean(enc, s.link.flags&linkAuth != 0)
if c.link.auth != "" {
enc, err = amf.EncodeBoolean(enc, c.link.flags&linkAuth != 0)
if err != nil {
return err
}
enc, err = amf.EncodeString(enc, s.link.auth)
enc, err = amf.EncodeString(enc, c.link.auth)
if err != nil {
return err
}
@ -335,16 +309,16 @@ func sendConnectPacket(s *Session) error {
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, true) // response expected
return pkt.writeTo(c, true) // response expected
}
func sendCreateStream(s *Session) error {
func sendCreateStream(c *Conn) error {
var pbuf [256]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -353,8 +327,8 @@ func sendCreateStream(s *Session) error {
if err != nil {
return err
}
s.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
@ -363,16 +337,16 @@ func sendCreateStream(s *Session) error {
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, true) // response expected
return pkt.writeTo(c, true) // response expected
}
func sendReleaseStream(s *Session) error {
func sendReleaseStream(c *Conn) error {
var pbuf [1024]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -381,29 +355,29 @@ func sendReleaseStream(s *Session) error {
if err != nil {
return err
}
s.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
enc[0] = amf.TypeNull
enc = enc[1:]
enc, err = amf.EncodeString(enc, s.link.playpath)
enc, err = amf.EncodeString(enc, c.link.playpath)
if err != nil {
return err
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, false)
return pkt.writeTo(c, false)
}
func sendFCPublish(s *Session) error {
func sendFCPublish(c *Conn) error {
var pbuf [1024]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -412,30 +386,30 @@ func sendFCPublish(s *Session) error {
if err != nil {
return err
}
s.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
enc[0] = amf.TypeNull
enc = enc[1:]
enc, err = amf.EncodeString(enc, s.link.playpath)
enc, err = amf.EncodeString(enc, c.link.playpath)
if err != nil {
return err
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, false)
return pkt.writeTo(c, false)
}
func sendFCUnpublish(s *Session) error {
func sendFCUnpublish(c *Conn) error {
var pbuf [1024]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -444,30 +418,30 @@ func sendFCUnpublish(s *Session) error {
if err != nil {
return err
}
s.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
enc[0] = amf.TypeNull
enc = enc[1:]
enc, err = amf.EncodeString(enc, s.link.playpath)
enc, err = amf.EncodeString(enc, c.link.playpath)
if err != nil {
return err
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, false)
return pkt.writeTo(c, false)
}
func sendPublish(s *Session) error {
func sendPublish(c *Conn) error {
var pbuf [1024]byte
pkt := packet{
channel: chanSource,
headerType: headerSizeLarge,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -476,14 +450,14 @@ func sendPublish(s *Session) error {
if err != nil {
return err
}
s.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
enc[0] = amf.TypeNull
enc = enc[1:]
enc, err = amf.EncodeString(enc, s.link.playpath)
enc, err = amf.EncodeString(enc, c.link.playpath)
if err != nil {
return err
}
@ -494,16 +468,16 @@ func sendPublish(s *Session) error {
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, true) // response expected
return pkt.writeTo(c, true) // response expected
}
func sendDeleteStream(s *Session, dStreamId float64) error {
func sendDeleteStream(c *Conn, streamID float64) error {
var pbuf [256]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -512,52 +486,52 @@ func sendDeleteStream(s *Session, dStreamId float64) error {
if err != nil {
return err
}
s.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
enc[0] = amf.TypeNull
enc = enc[1:]
enc, err = amf.EncodeNumber(enc, dStreamId)
enc, err = amf.EncodeNumber(enc, streamID)
if err != nil {
return err
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, false)
return pkt.writeTo(c, false)
}
// sendBytesReceived tells the server how many bytes the client has received.
func sendBytesReceived(s *Session) error {
func sendBytesReceived(c *Conn) error {
var pbuf [256]byte
pkt := packet{
channel: chanBytesRead,
headerType: headerSizeMedium,
packetType: packetTypeBytesReadReport,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
s.nBytesInSent = s.nBytesIn
c.nBytesInSent = c.nBytesIn
enc, err := amf.EncodeInt32(enc, s.nBytesIn)
enc, err := amf.EncodeInt32(enc, c.nBytesIn)
if err != nil {
return err
}
pkt.bodySize = 4
return pkt.writeTo(s, false)
return pkt.writeTo(c, false)
}
func sendCheckBW(s *Session) error {
func sendCheckBW(c *Conn) error {
var pbuf [256]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeLarge,
packetType: packetTypeInvoke,
header: pbuf[:],
buf: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
@ -566,8 +540,8 @@ func sendCheckBW(s *Session) error {
if err != nil {
return err
}
s.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
c.numInvokes++
enc, err = amf.EncodeNumber(enc, float64(c.numInvokes))
if err != nil {
return err
}
@ -576,7 +550,7 @@ func sendCheckBW(s *Session) error {
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.writeTo(s, false)
return pkt.writeTo(c, false)
}
func eraseMethod(m []method, i int) []method {
@ -586,8 +560,8 @@ func eraseMethod(m []method, i int) []method {
}
// int handleInvoke handles a packet invoke request
// Side effects: s.isPlaying set to true upon avNetStreamPublish_Start
func handleInvoke(s *Session, body []byte) error {
// Side effects: c.isPlaying set to true upon avNetStreamPublish_Start
func handleInvoke(c *Conn, body []byte) error {
if body[0] != 0x02 {
return errInvalidBody
}
@ -606,37 +580,37 @@ func handleInvoke(s *Session, body []byte) error {
return err
}
s.log(DebugLevel, pkg+"invoking method "+meth)
c.log(DebugLevel, pkg+"invoking method "+meth)
switch meth {
case av_result:
if (s.link.protocol & featureWrite) == 0 {
if (c.link.protocol & featureWrite) == 0 {
return errNotWritable
}
var methodInvoked string
for i, m := range s.methodCalls {
for i, m := range c.methodCalls {
if float64(m.num) == txn {
methodInvoked = m.name
s.methodCalls = eraseMethod(s.methodCalls, i)
c.methodCalls = eraseMethod(c.methodCalls, i)
break
}
}
if methodInvoked == "" {
s.log(WarnLevel, pkg+"received result without matching request", "id", txn)
c.log(WarnLevel, pkg+"received result without matching request", "id", txn)
return nil
}
s.log(DebugLevel, pkg+"received result for "+methodInvoked)
c.log(DebugLevel, pkg+"received result for "+methodInvoked)
switch methodInvoked {
case avConnect:
err := sendReleaseStream(s)
err := sendReleaseStream(c)
if err != nil {
return err
}
err = sendFCPublish(s)
err = sendFCPublish(c)
if err != nil {
return err
}
err = sendCreateStream(s)
err = sendCreateStream(c)
if err != nil {
return err
}
@ -646,18 +620,18 @@ func handleInvoke(s *Session, body []byte) error {
if err != nil {
return err
}
s.streamID = int32(n)
err = sendPublish(s)
c.streamID = uint32(n)
err = sendPublish(c)
if err != nil {
return err
}
default:
s.log(FatalLevel, pkg+"unexpected method invoked"+methodInvoked)
c.log(FatalLevel, pkg+"unexpected method invoked"+methodInvoked)
}
case avOnBWDone:
err := sendCheckBW(s)
err := sendCheckBW(c)
if err != nil {
return err
}
@ -675,27 +649,27 @@ func handleInvoke(s *Session, body []byte) error {
if err != nil {
return err
}
s.log(DebugLevel, pkg+"onStatus", "code", code, "level", level)
c.log(DebugLevel, pkg+"onStatus", "code", code, "level", level)
if code != avNetStreamPublish_Start {
s.log(ErrorLevel, pkg+"unexpected response "+code)
c.log(ErrorLevel, pkg+"unexpected response "+code)
return errUnimplemented
}
s.log(DebugLevel, pkg+"playing")
s.isPlaying = true
for i, m := range s.methodCalls {
c.log(DebugLevel, pkg+"playing")
c.isPlaying = true
for i, m := range c.methodCalls {
if m.name == avPublish {
s.methodCalls = eraseMethod(s.methodCalls, i)
c.methodCalls = eraseMethod(c.methodCalls, i)
}
}
default:
s.log(FatalLevel, pkg+"unsuppoted method "+meth)
c.log(FatalLevel, pkg+"unsuppoted method "+meth)
}
return nil
}
func handshake(s *Session) error {
func handshake(c *Conn) error {
var clientbuf [signatureSize + 1]byte
clientsig := clientbuf[1:]
@ -708,44 +682,44 @@ func handshake(s *Session) error {
clientsig[i] = byte(rand.Intn(256))
}
_, err := s.write(clientbuf[:])
_, err := c.write(clientbuf[:])
if err != nil {
return err
}
s.log(DebugLevel, pkg+"handshake sent")
c.log(DebugLevel, pkg+"handshake sent")
var typ [1]byte
_, err = s.read(typ[:])
_, err = c.read(typ[:])
if err != nil {
return err
}
s.log(DebugLevel, pkg+"handshake received")
c.log(DebugLevel, pkg+"handshake received")
if typ[0] != clientbuf[0] {
s.log(WarnLevel, pkg+"handshake type mismatch", "sent", clientbuf[0], "received", typ)
c.log(WarnLevel, pkg+"handshake type mismatch", "sent", clientbuf[0], "received", typ)
}
_, err = s.read(serversig[:])
_, err = c.read(serversig[:])
if err != nil {
return err
}
// decode server response
suptime := binary.BigEndian.Uint32(serversig[:4])
s.log(DebugLevel, pkg+"server uptime", "uptime", suptime)
c.log(DebugLevel, pkg+"server uptime", "uptime", suptime)
// 2nd part of handshake
_, err = s.write(serversig[:])
_, err = c.write(serversig[:])
if err != nil {
return err
}
_, err = s.read(serversig[:])
_, err = c.read(serversig[:])
if err != nil {
return err
}
if !bytes.Equal(serversig[:signatureSize], clientbuf[1:signatureSize+1]) {
s.log(WarnLevel, pkg+"signature mismatch", "serversig", serversig[:signatureSize], "clientsig", clientbuf[1:signatureSize+1])
c.log(WarnLevel, pkg+"signature mismatch", "serversig", serversig[:signatureSize], "clientsig", clientbuf[1:signatureSize+1])
}
return nil
}

86
rtmp/rtmp_test.go
View File

@ -116,75 +116,54 @@ func TestKey(t *testing.T) {
testLog(0, "Testing against URL "+testBaseURL+testKey)
}
// TestInit tests session construction and link initialization.
func TestInit(t *testing.T) {
testLog(0, "TestInit")
// test with just the base URL
s := NewSession(testBaseURL, testTimeout, testLog)
if s.url != testBaseURL && s.link.timeout != testTimeout {
t.Errorf("NewSession failed")
}
err := s.init()
if err != nil {
t.Errorf("setupURL: failed with error: %v", err)
}
// test the parts are as expected
if s.link.protocol&featureWrite == 0 {
t.Errorf("setupURL: link not writable")
}
if rtmpProtocolStrings[s.link.protocol&^featureWrite] != rtmpProtocol {
t.Errorf("setupURL: wrong protocol: %v", s.link.protocol)
}
if s.link.host != testHost {
t.Errorf("setupURL: wrong host: %v", s.link.host)
}
if s.link.app != testApp {
t.Errorf("setupURL: wrong app: %v", s.link.app)
}
}
// TestErrorHandling tests error handling
func TestErorHandling(t *testing.T) {
testLog(0, "TestErrorHandling")
if testKey == "" {
t.Skip("Skipping TestErrorHandling since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
// test errNotConnected
var buf [1024]byte
tag := buf[:0]
_, err := s.Write(tag)
if err == nil {
t.Errorf("Write did not return errNotConnected")
c, err := Dial(testBaseURL+testKey, testTimeout, testLog)
if err != nil {
t.Errorf("Dial failed with error: %v", err)
}
err = s.Open()
if err != nil {
t.Errorf("Open failed with error: %v", err)
// test the link parts are as expected
if c.link.protocol&featureWrite == 0 {
t.Errorf("link not writable")
}
if rtmpProtocolStrings[c.link.protocol&^featureWrite] != rtmpProtocol {
t.Errorf("wrong protocol: %v", c.link.protocol)
}
if c.link.host != testHost {
t.Errorf("wrong host: %v", c.link.host)
}
if c.link.app != testApp {
t.Errorf("wrong app: %v", c.link.app)
}
// test errInvalidFlvTag
_, err = s.Write(tag)
var buf [1024]byte
tag := buf[:0]
_, err = c.Write(tag)
if err == nil {
t.Errorf("Write did not return errInvalidFlvTag")
}
// test errUnimplemented
copy(tag, []byte("FLV"))
_, err = s.Write(tag)
_, err = c.Write(tag)
if err == nil {
t.Errorf("Write did not return errUnimplemented")
}
// test errInvalidBody
tag = buf[:11]
_, err = s.Write(tag)
_, err = c.Write(tag)
if err == nil {
t.Errorf("Write did not return errInvalidBody")
}
err = s.Close()
err = c.Close()
if err != nil {
t.Errorf("Close failed with error: %v", err)
return
@ -197,10 +176,9 @@ func TestFromFrame(t *testing.T) {
if testKey == "" {
t.Skip("Skipping TestFromFrame since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
c, err := Dial(testBaseURL+testKey, testTimeout, testLog)
if err != nil {
t.Errorf("Session.Open failed with error: %v", err)
t.Errorf("Dial failed with error: %v", err)
}
testFrame := os.Getenv("RTMP_TEST_FRAME")
@ -216,7 +194,7 @@ func TestFromFrame(t *testing.T) {
}
const frameRate = 25
rs := &rtmpSender{s: s}
rs := &rtmpSender{conn: c}
flvEncoder, err := flv.NewEncoder(rs, true, true, frameRate)
if err != nil {
t.Errorf("Failed to create flv encoder with error: %v", err)
@ -226,18 +204,18 @@ func TestFromFrame(t *testing.T) {
t.Errorf("Lexing failed with error: %v", err)
}
err = s.Close()
err = c.Close()
if err != nil {
t.Errorf("Session.Close failed with error: %v", err)
}
}
type rtmpSender struct {
s *Session
conn *Conn
}
func (rs *rtmpSender) Write(p []byte) (int, error) {
n, err := rs.s.Write(p)
n, err := rs.conn.Write(p)
if err != errInvalidFlvTag && err != nil {
return 0, err
}
@ -255,12 +233,10 @@ func TestFromFile(t *testing.T) {
if testKey == "" {
t.Skip("Skipping TestFromFile since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
c, err := Dial(testBaseURL+testKey, testTimeout, testLog)
if err != nil {
t.Errorf("Session.Open failed with error: %v", err)
t.Errorf("Dial failed with error: %v", err)
}
f, err := os.Open(testFile)
if err != nil {
t.Errorf("Open failed with error: %v", err)
@ -268,7 +244,7 @@ func TestFromFile(t *testing.T) {
defer f.Close()
// Pass RTMP session, true for audio, true for video, and 25 FPS
flvEncoder, err := flv.NewEncoder(s, true, true, 25)
flvEncoder, err := flv.NewEncoder(c, true, true, 25)
if err != nil {
t.Fatalf("failed to create encoder: %v", err)
}
@ -277,7 +253,7 @@ func TestFromFile(t *testing.T) {
t.Errorf("Lexing and encoding failed with error: %v", err)
}
err = s.Close()
err = c.Close()
if err != nil {
t.Errorf("Session.Close failed with error: %v", err)
}