Merge pull request #5 from LdDl/master

Handle 302, handle another stream index

format/rtsp/client.go

@@ -8,13 +8,6 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
-	"github.com/deepch/vdk/av"
-	"github.com/deepch/vdk/av/avutil"
-	"github.com/deepch/vdk/codec"
-	"github.com/deepch/vdk/codec/aacparser"
-	"github.com/deepch/vdk/codec/h264parser"
-	"github.com/deepch/vdk/format/rtsp/sdp"
-	"github.com/deepch/vdk/utils/bits/pio"
 	"io"
 	"log"
 	"net"
@@ -23,6 +16,14 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/deepch/vdk/av"
+	"github.com/deepch/vdk/av/avutil"
+	"github.com/deepch/vdk/codec"
+	"github.com/deepch/vdk/codec/aacparser"
+	"github.com/deepch/vdk/codec/h264parser"
+	"github.com/deepch/vdk/format/rtsp/sdp"
+	"github.com/deepch/vdk/utils/bits/pio"
 )
 
 var ErrCodecDataChange = fmt.Errorf("rtsp: codec data change, please call HandleCodecDataChange()")
@@ -309,6 +310,11 @@ func (self *Client) handleResp(res *Response) (err error) {
 			self.session = fields[0]
 		}
 	}
+	if res.StatusCode == 302 {
+		if err = self.handle302(res); err != nil {
+			return
+		}
+	}
 	if res.StatusCode == 401 {
 		if err = self.handle401(res); err != nil {
 			return
@@ -317,6 +323,31 @@ func (self *Client) handleResp(res *Response) (err error) {
 	return
 }
 
+func (self *Client) handle302(res *Response) (err error) {
+	/*
+		RTSP/1.0 200 OK
+		CSeq: 302
+	*/
+	newLocation := res.Headers.Get("Location")
+	fmt.Printf("\tRedirecting stream to other location: %s\n", newLocation)
+
+	err = self.Close()
+	if err != nil {
+		return err
+	}
+
+	newConnect, err := Dial(newLocation)
+	if err != nil {
+		return err
+	}
+
+	self.requestUri = newLocation
+	self.conn = newConnect.conn
+	self.brconn = newConnect.brconn
+
+	return err
+}
+
 func (self *Client) handle401(res *Response) (err error) {
 	/*
 		RTSP/1.0 401 Unauthorized
@@ -440,11 +471,15 @@ func (self *Client) findRTSP() (block []byte, data []byte, err error) {
 			}
 			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 {
+				if left >= 0 {
+					block = append(peek, make([]byte, left)...)
+					if _, err = io.ReadFull(self.brconn, block[len(peek):]); err != nil {
+						return
+					}
 					return
+				} else {
+					fmt.Println("Left < 0 ", blocklen, len(peek), left)
 				}
-				return
 			}
 			stat = 0
 			peek = _peek[0:0]

format/rtsp/sdp/parser.go

@@ -110,7 +110,10 @@ func Parse(content string) (sess Session, medias []Media) {
 					}
 				}
 
+			default:
+				media = nil
 			}
+
 		}
 	}
 	return

format/ts/muxer.go

@@ -14,8 +14,9 @@ import (
 var CodecTypes = []av.CodecType{av.H264, av.AAC}
 
 type Muxer struct {
-	w                        io.Writer
-	streams                  []*Stream
+	w       io.Writer
+	streams map[int]*Stream
+
 	PaddingToMakeCounterCont bool
 
 	psidata []byte
@@ -42,7 +43,7 @@ func NewMuxer(w io.Writer) *Muxer {
 	}
 }
 
-func (self *Muxer) newStream(codec av.CodecData) (err error) {
+func (self *Muxer) newStream(idx int, codec av.CodecData) (err error) {
 	ok := false
 	for _, c := range CodecTypes {
 		if codec.Type() == c {
@@ -55,14 +56,14 @@ func (self *Muxer) newStream(codec av.CodecData) (err error) {
 		return
 	}
 
-	pid := uint16(len(self.streams) + 0x100)
+	pid := uint16(idx + 0x100)
 	stream := &Stream{
 		muxer:     self,
 		CodecData: codec,
 		pid:       pid,
 		tsw:       tsio.NewTSWriter(pid),
 	}
-	self.streams = append(self.streams, stream)
+	self.streams[idx] = stream
 	return
 }
 
@@ -140,10 +141,11 @@ func (self *Muxer) WritePATPMT() (err error) {
 }
 
 func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) {
-	self.streams = []*Stream{}
-	for _, stream := range streams {
-		if err = self.newStream(stream); err != nil {
-			return
+	self.streams = map[int]*Stream{}
+
+	for idx, stream := range streams {
+		if err = self.newStream(idx, stream); err != nil {
+			fmt.Println(err)
 		}
 	}
 
@@ -154,7 +156,14 @@ func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) {
 }
 
 func (self *Muxer) WritePacket(pkt av.Packet) (err error) {
-	stream := self.streams[pkt.Idx]
+	var stream *Stream = nil
+
+	stream, ok := self.streams[int(pkt.Idx)]
+	if !ok {
+		fmt.Printf("Warning, unsupported stream index: %d\n", pkt.Idx)
+		return
+	}
+
 	pkt.Time += time.Second
 
 	switch stream.Type() {