1154 lines
26 KiB
Go
1154 lines
26 KiB
Go
package rtsp
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"crypto/md5"
|
||
"encoding/base64"
|
||
"encoding/binary"
|
||
"encoding/hex"
|
||
"fmt"
|
||
"github.com/nareix/av"
|
||
"github.com/nareix/codec"
|
||
"github.com/nareix/codec/aacparser"
|
||
"github.com/nareix/codec/h264parser"
|
||
"github.com/nareix/rtsp/sdp"
|
||
"html"
|
||
"io"
|
||
"net"
|
||
"net/textproto"
|
||
"net/url"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
var ErrCodecDataChange = fmt.Errorf("rtsp: codec data change, please call HandleCodecDataChange()")
|
||
|
||
type Client struct {
|
||
DebugRtsp bool
|
||
DebugRtp bool
|
||
Headers []string
|
||
|
||
RtspTimeout time.Duration
|
||
RtpTimeout time.Duration
|
||
RtpKeepAliveTimeout time.Duration
|
||
rtpKeepaliveTimer time.Time
|
||
rtpKeepaliveEnterCnt int
|
||
|
||
setupCalled bool
|
||
setupIdx []int
|
||
setupMap []int
|
||
playCalled bool
|
||
|
||
authHeaders func(method string) []string
|
||
|
||
url *url.URL
|
||
conn *connWithTimeout
|
||
brconn *bufio.Reader
|
||
requestUri string
|
||
cseq uint
|
||
streams []*Stream
|
||
streamsintf []av.CodecData
|
||
session string
|
||
body io.Reader
|
||
}
|
||
|
||
type Request struct {
|
||
Header []string
|
||
Uri string
|
||
Method string
|
||
}
|
||
|
||
type Response struct {
|
||
StatusCode int
|
||
Headers textproto.MIMEHeader
|
||
ContentLength int
|
||
Body []byte
|
||
|
||
Block []byte
|
||
}
|
||
|
||
func DialTimeout(uri string, timeout time.Duration) (self *Client, err error) {
|
||
var URL *url.URL
|
||
if URL, err = url.Parse(html.UnescapeString(uri)); err != nil {
|
||
return
|
||
}
|
||
|
||
if _, _, err := net.SplitHostPort(URL.Host); err != nil {
|
||
URL.Host = URL.Host + ":554"
|
||
}
|
||
|
||
dailer := net.Dialer{Timeout: timeout}
|
||
var conn net.Conn
|
||
if conn, err = dailer.Dial("tcp", URL.Host); err != nil {
|
||
return
|
||
}
|
||
|
||
u2 := *URL
|
||
u2.User = nil
|
||
|
||
connt := &connWithTimeout{Conn: conn}
|
||
|
||
self = &Client{
|
||
conn: connt,
|
||
brconn: bufio.NewReaderSize(connt, 256),
|
||
url: URL,
|
||
requestUri: u2.String(),
|
||
}
|
||
return
|
||
}
|
||
|
||
func Dial(uri string) (self *Client, err error) {
|
||
return DialTimeout(uri, 0)
|
||
}
|
||
|
||
func (self *Client) Streams() (streams []av.CodecData, err error) {
|
||
if self.setupCalled {
|
||
streams = self.streamsintf
|
||
} else {
|
||
err = fmt.Errorf("rtsp: no streams")
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Client) SendRtpKeepalive() (err error) {
|
||
if self.RtpKeepAliveTimeout > 0 {
|
||
if self.rtpKeepaliveTimer.IsZero() {
|
||
self.rtpKeepaliveTimer = time.Now()
|
||
} else if time.Now().Sub(self.rtpKeepaliveTimer) > self.RtpKeepAliveTimeout {
|
||
self.rtpKeepaliveTimer = time.Now()
|
||
if self.DebugRtsp {
|
||
fmt.Println("rtp: keep alive")
|
||
}
|
||
req := Request{
|
||
Method: "OPTIONS",
|
||
Uri: self.requestUri,
|
||
}
|
||
if err = self.WriteRequest(req); err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Client) WriteRequest(req Request) (err error) {
|
||
self.conn.Timeout = self.RtspTimeout
|
||
self.cseq++
|
||
|
||
buf := &bytes.Buffer{}
|
||
|
||
fmt.Fprintf(buf, "%s %s RTSP/1.0\r\n", req.Method, req.Uri)
|
||
fmt.Fprintf(buf, "CSeq: %d\r\n", self.cseq)
|
||
|
||
if self.authHeaders != nil {
|
||
headers := self.authHeaders(req.Method)
|
||
for _, s := range headers {
|
||
io.WriteString(buf, s)
|
||
io.WriteString(buf, "\r\n")
|
||
}
|
||
}
|
||
for _, s := range req.Header {
|
||
io.WriteString(buf, s)
|
||
io.WriteString(buf, "\r\n")
|
||
}
|
||
for _, s := range self.Headers {
|
||
io.WriteString(buf, s)
|
||
io.WriteString(buf, "\r\n")
|
||
}
|
||
io.WriteString(buf, "\r\n")
|
||
|
||
bufout := buf.Bytes()
|
||
|
||
if self.DebugRtsp {
|
||
fmt.Print("> ", string(bufout))
|
||
}
|
||
|
||
if _, err = self.conn.Write(bufout); err != nil {
|
||
return
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) parseBlockHeader(h []byte) (length int, no int, valid bool) {
|
||
length = int(h[2])<<8 + int(h[3])
|
||
no = int(h[1])
|
||
if no/2 >= len(self.streams) {
|
||
return
|
||
}
|
||
if length < 8 {
|
||
return
|
||
}
|
||
stream := self.streams[no/2]
|
||
if int(h[5]&0x7f) != stream.Sdp.PayloadType {
|
||
return
|
||
}
|
||
valid = true
|
||
return
|
||
}
|
||
|
||
func (self *Client) parseHeaders(b []byte) (statusCode int, headers textproto.MIMEHeader, err error) {
|
||
var line string
|
||
r := textproto.NewReader(bufio.NewReader(bytes.NewReader(b)))
|
||
if line, err = r.ReadLine(); err != nil {
|
||
err = fmt.Errorf("rtsp: header invalid")
|
||
return
|
||
}
|
||
|
||
if codes := strings.Split(line, " "); len(codes) >= 2 {
|
||
if statusCode, err = strconv.Atoi(codes[1]); err != nil {
|
||
err = fmt.Errorf("rtsp: header invalid: %s", err)
|
||
return
|
||
}
|
||
}
|
||
|
||
if headers, _ = r.ReadMIMEHeader(); err != nil {
|
||
return
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) handleResp(res *Response) (err error) {
|
||
if sess := res.Headers.Get("Session"); sess != "" && self.session == "" {
|
||
if fields := strings.Split(sess, ";"); len(fields) > 0 {
|
||
self.session = fields[0]
|
||
}
|
||
}
|
||
if res.StatusCode == 401 {
|
||
if err = self.handle401(res); err != nil {
|
||
return
|
||
}
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Client) handle401(res *Response) (err error) {
|
||
/*
|
||
RTSP/1.0 401 Unauthorized
|
||
CSeq: 2
|
||
Date: Wed, May 04 2016 10:10:51 GMT
|
||
WWW-Authenticate: Digest realm="LIVE555 Streaming Media", nonce="c633aaf8b83127633cbe98fac1d20d87"
|
||
*/
|
||
authval := res.Headers.Get("WWW-Authenticate")
|
||
hdrval := strings.SplitN(authval, " ", 2)
|
||
var realm, nonce string
|
||
|
||
if len(hdrval) == 2 {
|
||
for _, field := range strings.Split(hdrval[1], ",") {
|
||
field = strings.Trim(field, ", ")
|
||
if keyval := strings.Split(field, "="); len(keyval) == 2 {
|
||
key := keyval[0]
|
||
val := strings.Trim(keyval[1], `"`)
|
||
switch key {
|
||
case "realm":
|
||
realm = val
|
||
case "nonce":
|
||
nonce = val
|
||
}
|
||
}
|
||
}
|
||
|
||
if realm != "" {
|
||
var username string
|
||
var password string
|
||
|
||
if self.url.User == nil {
|
||
err = fmt.Errorf("rtsp: no username")
|
||
return
|
||
}
|
||
username = self.url.User.Username()
|
||
password, _ = self.url.User.Password()
|
||
|
||
self.authHeaders = func(method string) []string {
|
||
headers := []string{
|
||
fmt.Sprintf(`Authorization: Basic %s`, base64.StdEncoding.EncodeToString([]byte(username+":"+password))),
|
||
}
|
||
if nonce != "" {
|
||
hs1 := md5hash(username + ":" + realm + ":" + password)
|
||
hs2 := md5hash(method + ":" + self.requestUri)
|
||
response := md5hash(hs1 + ":" + nonce + ":" + hs2)
|
||
headers = append(headers, fmt.Sprintf(
|
||
`Authorization: Digest username="%s", realm="%s", nonce="%s", uri="%s", response="%s"`,
|
||
username, realm, nonce, self.requestUri, response))
|
||
}
|
||
return headers
|
||
}
|
||
}
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) findRTSP() (block []byte, data []byte, err error) {
|
||
const (
|
||
R = iota+1
|
||
T
|
||
S
|
||
Header
|
||
Dollar
|
||
)
|
||
var _peek [8]byte
|
||
peek := _peek[0:0]
|
||
stat := 0
|
||
|
||
for {
|
||
var b byte
|
||
if b, err = self.brconn.ReadByte(); err != nil {
|
||
return
|
||
}
|
||
switch b {
|
||
case 'R':
|
||
if stat == 0 {
|
||
stat = R
|
||
}
|
||
case 'T':
|
||
if stat == R {
|
||
stat = T
|
||
}
|
||
case 'S':
|
||
if stat == T {
|
||
stat = S
|
||
}
|
||
case 'P':
|
||
if stat == S {
|
||
stat = Header
|
||
}
|
||
case '$':
|
||
stat = Dollar
|
||
peek = _peek[0:0]
|
||
default:
|
||
if stat != Dollar {
|
||
stat = 0
|
||
peek = _peek[0:0]
|
||
}
|
||
}
|
||
|
||
if stat != 0 {
|
||
peek = append(peek, b)
|
||
}
|
||
if stat == Header {
|
||
data = peek
|
||
return
|
||
}
|
||
|
||
if stat == Dollar && len(peek) >= 8 {
|
||
if blocklen, _, ok := self.parseBlockHeader(peek); ok {
|
||
left := blocklen+4-len(peek)
|
||
block = append(peek, make([]byte, left)...)
|
||
if _, err = io.ReadFull(self.brconn, block[len(peek):]); err != nil {
|
||
return
|
||
}
|
||
return
|
||
}
|
||
stat = 0
|
||
peek = _peek[0:0]
|
||
}
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) readLFLF() (block []byte, data []byte, err error) {
|
||
const (
|
||
LF = iota+1
|
||
LFLF
|
||
)
|
||
peek := []byte{}
|
||
stat := 0
|
||
dollarpos := -1
|
||
lpos := 0
|
||
pos := 0
|
||
|
||
for {
|
||
var b byte
|
||
if b, err = self.brconn.ReadByte(); err != nil {
|
||
return
|
||
}
|
||
switch b {
|
||
case '\n':
|
||
if stat == 0 {
|
||
stat = LF
|
||
lpos = pos
|
||
} else if stat == LF {
|
||
if pos - lpos <= 2 {
|
||
stat = LFLF
|
||
} else {
|
||
lpos = pos
|
||
}
|
||
}
|
||
case '$':
|
||
dollarpos = pos
|
||
}
|
||
peek = append(peek, b)
|
||
|
||
if stat == LFLF {
|
||
data = peek
|
||
return
|
||
} else if dollarpos != -1 && dollarpos - pos >= 8 {
|
||
hdrlen := dollarpos-pos
|
||
start := len(peek)-hdrlen
|
||
if blocklen, _, ok := self.parseBlockHeader(peek[start:]); ok {
|
||
block = append(peek[start:], make([]byte, blocklen+4-hdrlen)...)
|
||
if _, err = io.ReadFull(self.brconn, block[hdrlen:]); err != nil {
|
||
return
|
||
}
|
||
return
|
||
}
|
||
dollarpos = -1
|
||
}
|
||
|
||
pos++
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) readResp(b []byte) (res Response, err error) {
|
||
if res.StatusCode, res.Headers, err = self.parseHeaders(b); err != nil {
|
||
return
|
||
}
|
||
res.ContentLength, _ = strconv.Atoi(res.Headers.Get("Content-Length"))
|
||
if res.ContentLength > 0 {
|
||
res.Body = make([]byte, res.ContentLength)
|
||
if _, err = io.ReadFull(self.brconn, res.Body); err != nil {
|
||
return
|
||
}
|
||
}
|
||
if err = self.handleResp(&res); err != nil {
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Client) poll() (res Response, err error) {
|
||
var block []byte
|
||
var rtsp []byte
|
||
var headers []byte
|
||
|
||
self.conn.Timeout = self.RtspTimeout
|
||
for {
|
||
if block, rtsp, err = self.findRTSP(); err != nil {
|
||
return
|
||
}
|
||
if len(block) > 0 {
|
||
res.Block = block
|
||
return
|
||
} else {
|
||
if block, headers, err = self.readLFLF(); err != nil {
|
||
return
|
||
}
|
||
if len(block) > 0 {
|
||
res.Block = block
|
||
return
|
||
}
|
||
if res, err = self.readResp(append(rtsp, headers...)); err != nil {
|
||
return
|
||
}
|
||
}
|
||
return
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) ReadResponse() (res Response, err error) {
|
||
for {
|
||
if res, err = self.poll(); err != nil {
|
||
return
|
||
}
|
||
if res.StatusCode > 0 {
|
||
return
|
||
}
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Client) setupAll() (err error) {
|
||
idx := []int{}
|
||
for i := range self.streams {
|
||
idx = append(idx, i)
|
||
}
|
||
return self.Setup(idx)
|
||
}
|
||
|
||
func (self *Client) Setup(idx []int) (err error) {
|
||
if self.setupCalled {
|
||
err = fmt.Errorf("rtsp: Setup() called twice")
|
||
return
|
||
}
|
||
|
||
if len(self.streams) == 0 {
|
||
err = fmt.Errorf("rtsp: no streams, please call Describe() first")
|
||
return
|
||
}
|
||
|
||
self.setupMap = make([]int, len(self.streams))
|
||
for i := range self.setupMap {
|
||
self.setupMap[i] = -1
|
||
}
|
||
self.setupIdx = idx
|
||
|
||
for i, si := range idx {
|
||
self.setupMap[si] = i
|
||
|
||
uri := ""
|
||
control := self.streams[si].Sdp.Control
|
||
if strings.HasPrefix(control, "rtsp://") {
|
||
uri = control
|
||
} else {
|
||
uri = self.requestUri + "/" + control
|
||
}
|
||
req := Request{Method: "SETUP", Uri: uri}
|
||
req.Header = append(req.Header, fmt.Sprintf("Transport: RTP/AVP/TCP;unicast;interleaved=%d-%d", si*2, si*2+1))
|
||
if self.session != "" {
|
||
req.Header = append(req.Header, "Session: "+self.session)
|
||
}
|
||
if err = self.WriteRequest(req); err != nil {
|
||
return
|
||
}
|
||
if _, err = self.ReadResponse(); err != nil {
|
||
return
|
||
}
|
||
}
|
||
self.initstructs()
|
||
|
||
self.setupCalled = true
|
||
return
|
||
}
|
||
|
||
func md5hash(s string) string {
|
||
h := md5.Sum([]byte(s))
|
||
return hex.EncodeToString(h[:])
|
||
}
|
||
|
||
func (self *Client) initstructs() {
|
||
self.streamsintf = make([]av.CodecData, len(self.setupIdx))
|
||
for i := range self.setupIdx {
|
||
self.streamsintf[i] = self.streams[self.setupIdx[i]].CodecData
|
||
}
|
||
}
|
||
|
||
func (self *Client) Describe() (streams []av.CodecData, err error) {
|
||
var res Response
|
||
|
||
for i := 0; i < 2; i++ {
|
||
req := Request{
|
||
Method: "DESCRIBE",
|
||
Uri: self.requestUri,
|
||
Header: []string{"Accept: application/sdp"},
|
||
}
|
||
if err = self.WriteRequest(req); err != nil {
|
||
return
|
||
}
|
||
if res, err = self.ReadResponse(); err != nil {
|
||
return
|
||
}
|
||
if res.StatusCode == 200 {
|
||
break
|
||
}
|
||
}
|
||
if res.ContentLength == 0 {
|
||
err = fmt.Errorf("rtsp: Describe failed, StatusCode=%d", res.StatusCode)
|
||
return
|
||
}
|
||
|
||
body := string(res.Body)
|
||
|
||
if self.DebugRtsp {
|
||
fmt.Println("<", body)
|
||
}
|
||
|
||
_, medias := sdp.Parse(body)
|
||
|
||
self.streams = []*Stream{}
|
||
for _, media := range medias {
|
||
stream := &Stream{Sdp: media, client: self}
|
||
if err = stream.makeCodecData(); err != nil {
|
||
return
|
||
}
|
||
self.streams = append(self.streams, stream)
|
||
}
|
||
for _, stream := range self.streams {
|
||
streams = append(streams, stream)
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) Options() (err error) {
|
||
req := Request{
|
||
Method: "OPTIONS",
|
||
Uri: self.requestUri,
|
||
}
|
||
if self.session != "" {
|
||
req.Header = append(req.Header, "Session: "+self.session)
|
||
}
|
||
if err = self.WriteRequest(req); err != nil {
|
||
return
|
||
}
|
||
if _, err = self.ReadResponse(); err != nil {
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Client) HandleCodecDataChange() (_newcli *Client, err error) {
|
||
newcli := &Client{}
|
||
*newcli = *self
|
||
|
||
newcli.streams = []*Stream{}
|
||
for _, stream := range self.streams {
|
||
newstream := &Stream{}
|
||
*newstream = *stream
|
||
newstream.client = newcli
|
||
|
||
if newstream.isCodecDataChange() {
|
||
if err = newstream.makeCodecData(); err != nil {
|
||
return
|
||
}
|
||
newstream.clearCodecDataChange()
|
||
}
|
||
newcli.streams = append(newcli.streams, newstream)
|
||
}
|
||
newcli.initstructs()
|
||
|
||
_newcli = newcli
|
||
return
|
||
}
|
||
|
||
func (self *Stream) clearCodecDataChange() {
|
||
self.spsChanged = false
|
||
self.ppsChanged = false
|
||
}
|
||
|
||
func (self *Stream) isCodecDataChange() bool {
|
||
if self.spsChanged && self.ppsChanged {
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
func (self *Stream) makeCodecData() (err error) {
|
||
media := self.Sdp
|
||
|
||
if media.PayloadType >= 96 && media.PayloadType <= 127 {
|
||
switch media.Type {
|
||
case av.H264:
|
||
for _, nalu := range media.SpropParameterSets {
|
||
if len(nalu) > 0 {
|
||
self.handleH264Payload(0, nalu)
|
||
}
|
||
}
|
||
|
||
if len(self.sps) == 0 || len(self.pps) == 0 {
|
||
if nalus, ok := h264parser.SplitNALUs(media.Config); ok {
|
||
for _, nalu := range nalus {
|
||
if len(nalu) > 0 {
|
||
self.handleH264Payload(0, nalu)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if len(self.sps) > 0 && len(self.pps) > 0 {
|
||
if self.CodecData, err = h264parser.NewCodecDataFromSPSAndPPS(self.sps, self.pps); err != nil {
|
||
err = fmt.Errorf("rtsp: h264 sps/pps invalid: %s", err)
|
||
return
|
||
}
|
||
} else {
|
||
err = fmt.Errorf("rtsp: missing h264 sps or pps")
|
||
return
|
||
}
|
||
|
||
case av.AAC:
|
||
if len(media.Config) == 0 {
|
||
err = fmt.Errorf("rtsp: aac sdp config missing")
|
||
return
|
||
}
|
||
if self.CodecData, err = aacparser.NewCodecDataFromMPEG4AudioConfigBytes(media.Config); err != nil {
|
||
err = fmt.Errorf("rtsp: aac sdp config invalid: %s", err)
|
||
return
|
||
}
|
||
}
|
||
} else {
|
||
switch media.PayloadType {
|
||
case 0:
|
||
self.CodecData = codec.NewPCMMulawCodecData()
|
||
|
||
case 8:
|
||
self.CodecData = codec.NewPCMAlawCodecData()
|
||
|
||
default:
|
||
err = fmt.Errorf("rtsp: PayloadType=%d unsupported", media.PayloadType)
|
||
return
|
||
}
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Stream) handleBuggyAnnexbH264Packet(timestamp uint32, packet []byte) (isBuggy bool, err error) {
|
||
if len(packet) >= 4 && packet[0] == 0 && packet[1] == 0 && packet[2] == 0 && packet[3] == 1 {
|
||
isBuggy = true
|
||
if nalus, ok := h264parser.SplitNALUs(packet); ok {
|
||
for _, nalu := range nalus {
|
||
if len(nalu) > 0 {
|
||
if err = self.handleH264Payload(timestamp, nalu); err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Stream) handleH264Payload(timestamp uint32, packet []byte) (err error) {
|
||
if len(packet) < 2 {
|
||
err = fmt.Errorf("rtp: h264 packet too short")
|
||
return
|
||
}
|
||
|
||
var isBuggy bool
|
||
if isBuggy, err = self.handleBuggyAnnexbH264Packet(timestamp, packet); isBuggy {
|
||
return
|
||
}
|
||
|
||
naluType := packet[0]&0x1f
|
||
|
||
/*
|
||
Table 7-1 – NAL unit type codes
|
||
1 Coded slice of a non-IDR picture
|
||
5 Coded slice of an IDR picture
|
||
6 Supplemental enhancement information (SEI)
|
||
7 Sequence parameter set
|
||
8 Picture parameter set
|
||
1-23 NAL unit Single NAL unit packet 5.6
|
||
24 STAP-A Single-time aggregation packet 5.7.1
|
||
25 STAP-B Single-time aggregation packet 5.7.1
|
||
26 MTAP16 Multi-time aggregation packet 5.7.2
|
||
27 MTAP24 Multi-time aggregation packet 5.7.2
|
||
28 FU-A Fragmentation unit 5.8
|
||
29 FU-B Fragmentation unit 5.8
|
||
30-31 reserved -
|
||
*/
|
||
switch {
|
||
default:
|
||
if naluType >= 1 && naluType <= 23 {
|
||
if naluType == 5 {
|
||
self.pkt.IsKeyFrame = true
|
||
}
|
||
self.gotpkt = true
|
||
self.pkt.Data = packet
|
||
self.timestamp = timestamp
|
||
} else {
|
||
err = fmt.Errorf("rtsp: unsupported H264 naluType=%d", naluType)
|
||
return
|
||
}
|
||
|
||
case naluType == 6: // SEI ignored
|
||
|
||
case naluType == 7: // sps
|
||
if self.client != nil && self.client.DebugRtp {
|
||
fmt.Println("rtsp: got sps")
|
||
}
|
||
if len(self.sps) == 0 {
|
||
self.sps = packet
|
||
} else if bytes.Compare(self.sps, packet) != 0 {
|
||
self.spsChanged = true
|
||
self.sps = packet
|
||
if self.client != nil && self.client.DebugRtp {
|
||
fmt.Println("rtsp: sps changed")
|
||
}
|
||
}
|
||
|
||
case naluType == 8: // pps
|
||
if self.client != nil && self.client.DebugRtp {
|
||
fmt.Println("rtsp: got pps")
|
||
}
|
||
if len(self.pps) == 0 {
|
||
self.pps = packet
|
||
} else if bytes.Compare(self.pps, packet) != 0 {
|
||
self.ppsChanged = true
|
||
self.pps = packet
|
||
if self.client != nil && self.client.DebugRtp {
|
||
fmt.Println("rtsp: pps changed")
|
||
}
|
||
}
|
||
|
||
case naluType == 28: // FU-A
|
||
/*
|
||
0 1 2 3
|
||
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| FU indicator | FU header | |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
|
||
| |
|
||
| FU payload |
|
||
| |
|
||
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| :...OPTIONAL RTP padding |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
Figure 14. RTP payload format for FU-A
|
||
|
||
The FU indicator octet has the following format:
|
||
+---------------+
|
||
|0|1|2|3|4|5|6|7|
|
||
+-+-+-+-+-+-+-+-+
|
||
|F|NRI| Type |
|
||
+---------------+
|
||
|
||
|
||
The FU header has the following format:
|
||
+---------------+
|
||
|0|1|2|3|4|5|6|7|
|
||
+-+-+-+-+-+-+-+-+
|
||
|S|E|R| Type |
|
||
+---------------+
|
||
|
||
S: 1 bit
|
||
When set to one, the Start bit indicates the start of a fragmented
|
||
NAL unit. When the following FU payload is not the start of a
|
||
fragmented NAL unit payload, the Start bit is set to zero.
|
||
|
||
E: 1 bit
|
||
When set to one, the End bit indicates the end of a fragmented NAL
|
||
unit, i.e., the last byte of the payload is also the last byte of
|
||
the fragmented NAL unit. When the following FU payload is not the
|
||
last fragment of a fragmented NAL unit, the End bit is set to
|
||
zero.
|
||
|
||
R: 1 bit
|
||
The Reserved bit MUST be equal to 0 and MUST be ignored by the
|
||
receiver.
|
||
|
||
Type: 5 bits
|
||
The NAL unit payload type as defined in table 7-1 of [1].
|
||
*/
|
||
fuIndicator := packet[0]
|
||
fuHeader := packet[1]
|
||
isStart := fuHeader&0x80 != 0
|
||
isEnd := fuHeader&0x40 != 0
|
||
if isStart {
|
||
self.fuStarted = true
|
||
self.fuBuffer = []byte{fuIndicator&0xe0 | fuHeader&0x1f}
|
||
}
|
||
if self.fuStarted {
|
||
self.fuBuffer = append(self.fuBuffer, packet[2:]...)
|
||
if isEnd {
|
||
self.fuStarted = false
|
||
if err = self.handleH264Payload(timestamp, self.fuBuffer); err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
case naluType == 24: // STAP-A
|
||
/*
|
||
0 1 2 3
|
||
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| RTP Header |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
|STAP-A NAL HDR | NALU 1 Size | NALU 1 HDR |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| NALU 1 Data |
|
||
: :
|
||
+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| | NALU 2 Size | NALU 2 HDR |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| NALU 2 Data |
|
||
: :
|
||
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| :...OPTIONAL RTP padding |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
|
||
Figure 7. An example of an RTP packet including an STAP-A
|
||
containing two single-time aggregation units
|
||
*/
|
||
packet = packet[1:]
|
||
for len(packet) >= 2 {
|
||
size := int(packet[0])<<8|int(packet[1])
|
||
if size+2 > len(packet) {
|
||
break
|
||
}
|
||
if err = self.handleH264Payload(timestamp, packet[2:size+2]); err != nil {
|
||
return
|
||
}
|
||
packet = packet[size+2:]
|
||
}
|
||
return
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Stream) handleRtpPacket(packet []byte) (err error) {
|
||
if self.isCodecDataChange() {
|
||
err = ErrCodecDataChange
|
||
return
|
||
}
|
||
|
||
if self.client != nil && self.client.DebugRtp {
|
||
fmt.Println("rtp: packet len", len(packet))
|
||
dumpsize := len(packet)
|
||
if dumpsize > 32 {
|
||
dumpsize = 32
|
||
}
|
||
fmt.Print(hex.Dump(packet[:dumpsize]))
|
||
}
|
||
|
||
/*
|
||
0 1 2 3
|
||
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
|V=2|P|X| CC |M| PT | sequence number |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| timestamp |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
| synchronization source (SSRC) identifier |
|
||
+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
|
||
| contributing source (CSRC) identifiers |
|
||
| .... |
|
||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||
*/
|
||
if len(packet) < 8 {
|
||
err = fmt.Errorf("rtp: packet too short")
|
||
return
|
||
}
|
||
payloadOffset := 12 + int(packet[0]&0xf)*4
|
||
if payloadOffset > len(packet) {
|
||
err = fmt.Errorf("rtp: packet too short")
|
||
return
|
||
}
|
||
timestamp := binary.BigEndian.Uint32(packet[4:8])
|
||
payload := packet[payloadOffset:]
|
||
|
||
/*
|
||
PT Encoding Name Audio/Video (A/V) Clock Rate (Hz) Channels Reference
|
||
0 PCMU A 8000 1 [RFC3551]
|
||
1 Reserved
|
||
2 Reserved
|
||
3 GSM A 8000 1 [RFC3551]
|
||
4 G723 A 8000 1 [Vineet_Kumar][RFC3551]
|
||
5 DVI4 A 8000 1 [RFC3551]
|
||
6 DVI4 A 16000 1 [RFC3551]
|
||
7 LPC A 8000 1 [RFC3551]
|
||
8 PCMA A 8000 1 [RFC3551]
|
||
9 G722 A 8000 1 [RFC3551]
|
||
10 L16 A 44100 2 [RFC3551]
|
||
11 L16 A 44100 1 [RFC3551]
|
||
12 QCELP A 8000 1 [RFC3551]
|
||
13 CN A 8000 1 [RFC3389]
|
||
14 MPA A 90000 [RFC3551][RFC2250]
|
||
15 G728 A 8000 1 [RFC3551]
|
||
16 DVI4 A 11025 1 [Joseph_Di_Pol]
|
||
17 DVI4 A 22050 1 [Joseph_Di_Pol]
|
||
18 G729 A 8000 1 [RFC3551]
|
||
19 Reserved A
|
||
20 Unassigned A
|
||
21 Unassigned A
|
||
22 Unassigned A
|
||
23 Unassigned A
|
||
24 Unassigned V
|
||
25 CelB V 90000 [RFC2029]
|
||
26 JPEG V 90000 [RFC2435]
|
||
27 Unassigned V
|
||
28 nv V 90000 [RFC3551]
|
||
29 Unassigned V
|
||
30 Unassigned V
|
||
31 H261 V 90000 [RFC4587]
|
||
32 MPV V 90000 [RFC2250]
|
||
33 MP2T AV 90000 [RFC2250]
|
||
34 H263 V 90000 [Chunrong_Zhu]
|
||
35-71 Unassigned ?
|
||
72-76 Reserved for RTCP conflict avoidance [RFC3551]
|
||
77-95 Unassigned ?
|
||
96-127 dynamic ? [RFC3551]
|
||
*/
|
||
//payloadType := packet[1]&0x7f
|
||
|
||
switch self.Type() {
|
||
case av.H264:
|
||
if err = self.handleH264Payload(timestamp, payload); err != nil {
|
||
return
|
||
}
|
||
|
||
case av.AAC:
|
||
if len(payload) < 4 {
|
||
err = fmt.Errorf("rtp: aac packet too short")
|
||
return
|
||
}
|
||
self.gotpkt = true
|
||
self.pkt.Data = payload[4:] // TODO: remove this hack
|
||
self.timestamp = timestamp
|
||
|
||
default:
|
||
self.gotpkt = true
|
||
self.pkt.Data = packet
|
||
self.timestamp = timestamp
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) Play() (err error) {
|
||
req := Request{
|
||
Method: "PLAY",
|
||
Uri: self.requestUri,
|
||
}
|
||
req.Header = append(req.Header, "Session: "+self.session)
|
||
if err = self.WriteRequest(req); err != nil {
|
||
return
|
||
}
|
||
self.playCalled = true
|
||
return
|
||
}
|
||
|
||
func (self *Client) Teardown() (err error) {
|
||
req := Request{
|
||
Method: "TEARDOWN",
|
||
Uri: self.requestUri,
|
||
}
|
||
req.Header = append(req.Header, "Session: "+self.session)
|
||
if err = self.WriteRequest(req); err != nil {
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func (self *Client) Close() (err error) {
|
||
return self.conn.Conn.Close()
|
||
}
|
||
|
||
func (self *Client) handleBlock(block []byte) (pkt av.Packet, ok bool, err error) {
|
||
_, blockno, _ := self.parseBlockHeader(block)
|
||
if blockno%2 != 0 {
|
||
return
|
||
}
|
||
|
||
i := blockno/2
|
||
if i >= len(self.streams) {
|
||
err = fmt.Errorf("rtsp: block no=%d invalid", blockno)
|
||
return
|
||
}
|
||
stream := self.streams[i]
|
||
|
||
if err = stream.handleRtpPacket(block[4:]); err != nil {
|
||
return
|
||
}
|
||
|
||
if stream.gotpkt {
|
||
timeScale := stream.Sdp.TimeScale
|
||
|
||
/*
|
||
TODO: sync AV by rtcp NTP timestamp
|
||
https://tools.ietf.org/html/rfc3550
|
||
A receiver can then synchronize presentation of the audio and video packets by relating
|
||
their RTP timestamps using the timestamp pairs in RTCP SR packets.
|
||
*/
|
||
if stream.firsttimestamp == 0 {
|
||
stream.firsttimestamp = stream.timestamp
|
||
}
|
||
stream.timestamp -= stream.firsttimestamp
|
||
|
||
if timeScale == 0 {
|
||
/*
|
||
https://tools.ietf.org/html/rfc5391
|
||
The RTP timestamp clock frequency is the same as the default sampling frequency: 16 kHz.
|
||
*/
|
||
timeScale = 16000
|
||
}
|
||
|
||
ok = true
|
||
pkt = stream.pkt
|
||
pkt.Time = time.Duration(stream.timestamp)*time.Second / time.Duration(timeScale)
|
||
pkt.Idx = int8(self.setupMap[i])
|
||
|
||
if pkt.Time < stream.lasttime {
|
||
err = fmt.Errorf("rtp: stream#%d time=%v < lasttime=%v", pkt.Time, stream.lasttime)
|
||
}
|
||
stream.lasttime = pkt.Time
|
||
|
||
if self.DebugRtp {
|
||
fmt.Println("rtp: pktout", pkt.Idx, pkt.Time, len(pkt.Data))
|
||
}
|
||
|
||
stream.pkt = av.Packet{}
|
||
stream.gotpkt = false
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) ReadPacket() (pkt av.Packet, err error) {
|
||
if !self.setupCalled {
|
||
if err = self.setupAll(); err != nil {
|
||
return
|
||
}
|
||
}
|
||
if !self.playCalled {
|
||
if err = self.Play(); err != nil {
|
||
return
|
||
}
|
||
}
|
||
|
||
if err = self.SendRtpKeepalive(); err != nil {
|
||
return
|
||
}
|
||
|
||
for {
|
||
var res Response
|
||
for {
|
||
if res, err = self.poll(); err != nil {
|
||
return
|
||
}
|
||
if len(res.Block) > 0 {
|
||
break
|
||
}
|
||
}
|
||
|
||
var ok bool
|
||
if pkt, ok, err = self.handleBlock(res.Block); err != nil {
|
||
return
|
||
}
|
||
if ok {
|
||
return
|
||
}
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func (self *Client) ReadHeader() (err error) {
|
||
if err = self.Options(); err != nil {
|
||
return
|
||
}
|
||
if _, err = self.Describe(); err != nil {
|
||
return
|
||
}
|
||
if err = self.setupAll(); err != nil {
|
||
return
|
||
}
|
||
if err = self.Play(); err != nil {
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func Open(uri string) (cli *Client, err error) {
|
||
var _cli *Client
|
||
if _cli, err = Dial(uri); err != nil {
|
||
return
|
||
}
|
||
if err = _cli.ReadHeader(); err != nil {
|
||
return
|
||
}
|
||
cli = _cli
|
||
return
|
||
}
|