Adds keep-alive to RTSP connection

cmd/streams/handlers.go

@@ -19,6 +19,9 @@ func HandleFunc(scheme string, handler Handler) {
 
 func HasProducer(url string) bool {
 	i := strings.IndexByte(url, ':')
+	if i <= 0 { // TODO: i < 4 ?
+		return false
+	}
 	return handlers[url[:i]] != nil
 }
 

pkg/rtmp/client.go

@@ -3,9 +3,12 @@ package rtmp
 import (
 	"encoding/base64"
 	"encoding/binary"
+	"encoding/hex"
+	"fmt"
 	"github.com/AlexxIT/go2rtc/pkg/h264"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/deepch/vdk/av"
+	"github.com/deepch/vdk/codec/aacparser"
 	"github.com/deepch/vdk/codec/h264parser"
 	"github.com/deepch/vdk/format/rtmp"
 	"github.com/pion/rtp"
@@ -70,9 +73,36 @@ func (c *Client) Dial() (err error) {
 			c.tracks = append(c.tracks, track)
 
 		case av.AAC:
-			panic("not implemented")
+			// TODO: fix support
+			cd := stream.(aacparser.CodecData)
+
+			// a=fmtp:97 streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1588
+			fmtp := fmt.Sprintf(
+				"config=%s",
+				hex.EncodeToString(cd.ConfigBytes),
+			)
+
+			codec := &streamer.Codec{
+				Name:      streamer.CodecAAC,
+				ClockRate: uint32(cd.Config.SampleRate),
+				Channels:  uint16(cd.Config.ChannelConfig),
+				FmtpLine:  fmtp,
+			}
+
+			media := &streamer.Media{
+				Kind:      streamer.KindAudio,
+				Direction: streamer.DirectionSendonly,
+				Codecs:    []*streamer.Codec{codec},
+			}
+			c.medias = append(c.medias, media)
+
+			track := &streamer.Track{
+				Codec: codec, Direction: media.Direction,
+			}
+			c.tracks = append(c.tracks, track)
+
 		default:
-			panic("unsupported codec")
+			fmt.Printf("[rtmp] unsupported codec %+v\n", stream)
 		}
 	}
 

pkg/rtsp/conn.go

@@ -43,6 +43,8 @@ const (
 	ModeServerConsumer
 )
 
+const KeepAlive = time.Second * 25
+
 type Conn struct {
 	streamer.Element
 
@@ -575,6 +577,7 @@ func (c *Conn) Handle() (err error) {
 	}()
 
 	//c.Fire(streamer.StatePlaying)
+	ts := time.Now().Add(KeepAlive)
 
 	for {
 		// we can read:
@@ -629,7 +632,7 @@ func (c *Conn) Handle() (err error) {
 		if channelID&1 == 0 {
 			packet := &rtp.Packet{}
 			if err = packet.Unmarshal(buf); err != nil {
-				return errors.New("wrong RTP data")
+				return
 			}
 
 			track := c.channels[channelID]
@@ -643,16 +646,27 @@ func (c *Conn) Handle() (err error) {
 			msg := &RTCP{Channel: channelID}
 
 			if err = msg.Header.Unmarshal(buf); err != nil {
-				return errors.New("wrong RTCP data")
+				return
 			}
 
 			msg.Packets, err = rtcp.Unmarshal(buf)
 			if err != nil {
-				return errors.New("wrong RTCP data")
+				return
 			}
 
 			c.Fire(msg)
 		}
+
+		// keep-alive
+		now := time.Now()
+		if now.After(ts) {
+			req := &tcp.Request{Method: MethodOptions, URL: c.URL}
+			// don't need to wait respose on this request
+			if err = c.Request(req); err != nil {
+				return err
+			}
+			ts = now.Add(KeepAlive)
+		}
 	}
 }