实现rtsp发送rtcp消息

rtsp/rtsp_server.go

@@ -1,7 +1,6 @@
 package rtsp
 
 import (
-	"fmt"
 	"github.com/yangjiechina/avformat/transport"
 	"github.com/yangjiechina/avformat/utils"
 	"github.com/yangjiechina/live-server/log"
@@ -72,7 +71,7 @@ func (s *serverImpl) OnPacket(conn net.Conn, data []byte) {
 
 	message, url, header, err := parseMessage(data)
 	if err != nil {
-		println(fmt.Sprintf("failed to prase message:%s. err:%s peer:%s", string(data), err.Error(), conn.RemoteAddr().String()))
+		log.Sugar.Errorf("failed to prase message:%s. err:%s peer:%s", string(data), err.Error(), conn.RemoteAddr().String())
 		_ = conn.Close()
 		s.closeSession(conn)
 		return
@@ -80,7 +79,8 @@ func (s *serverImpl) OnPacket(conn net.Conn, data []byte) {
 
 	err = t.Data.(*session).Input(message, url, header)
 	if err != nil {
-		println(fmt.Sprintf("failed to process message of RTSP. err:%s peer:%s msg:%s", err.Error(), conn.RemoteAddr().String(), string(data)))
+		log.Sugar.Errorf("failed to process message of RTSP. err:%s peer:%s msg:%s", err.Error(), conn.RemoteAddr().String(), string(data))
+
 		_ = conn.Close()
 	}
 }

rtsp/rtsp_session.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"fmt"
 	"github.com/yangjiechina/avformat/utils"
+	"github.com/yangjiechina/live-server/log"
 	"github.com/yangjiechina/live-server/stream"
 	"net"
 	"net/http"
@@ -60,7 +61,7 @@ type requestHandler interface {
 type session struct {
 	conn net.Conn
 
-	sink_       *Sink
+	sink_       *sink
 	sessionId   string
 	writeBuffer *bytes.Buffer
 }
@@ -165,7 +166,7 @@ func (s *session) onDescribe(source string, headers textproto.MIMEHeader) error
 	})
 
 	code := utils.HookStateOK
-	s.sink_ = sink_
+	s.sink_ = sink_.(*sink)
 
 	stream.HookPlaying(sink_, func() {
 
@@ -209,15 +210,18 @@ func (s *session) onSetup(sourceId string, index int, headers textproto.MIMEHead
 			}
 			_ = port
 
-			port, err = strconv.Atoi(pairPort[1])
+			port2, err := strconv.Atoi(pairPort[1])
 			if err != nil {
 				return err
 			}
-			_ = port
+			_ = port2
+
+			log.Sugar.Debugf("client port:%d-%d", port, port2)
 		}
 	}
 
-	rtpPort, rtcpPort, err := s.sink_.addTrack(index, tcp)
+	ssrc := 0xFFFFFFFF
+	rtpPort, rtcpPort, err := s.sink_.addTrack(index, tcp, uint32(ssrc))
 	if err != nil {
 		return err
 	}
@@ -229,7 +233,8 @@ func (s *session) onSetup(sourceId string, index int, headers textproto.MIMEHead
 	} else {
 		responseHeader += ";server_port=" + fmt.Sprintf("%d-%d", rtpPort, rtcpPort)
 	}
-	responseHeader += ";ssrc=FFFFFFFF"
+
+	responseHeader += ";ssrc=" + strconv.FormatInt(int64(ssrc), 16)
 
 	response := NewOKResponse(headers.Get("Cseq"))
 	response.Header.Set("Transport", responseHeader)
@@ -297,5 +302,8 @@ func (s *session) Input(method string, url_ *url.URL, headers textproto.MIMEHead
 }
 
 func (s *session) close() {
-
+	if s.sink_ != nil {
+		s.sink_.Close()
+		s.sink_ = nil
+	}
 }

rtsp/rtsp_sink.go

@@ -2,10 +2,13 @@ package rtsp
 
 import (
 	"fmt"
+	"github.com/pion/rtcp"
 	"github.com/yangjiechina/avformat/transport"
 	"github.com/yangjiechina/avformat/utils"
+	"github.com/yangjiechina/live-server/log"
 	"github.com/yangjiechina/live-server/stream"
 	"net"
+	"time"
 )
 
 // 对于UDP而言, 每个sink维护一对UDPTransport
@@ -17,6 +20,7 @@ type sink struct {
 	tracks []*rtspTrack
 	sdpCb  func(sdp string)
 
+	//是否是TCP拉流
 	tcp     bool
 	playing bool
 }
@@ -35,7 +39,7 @@ func (s *sink) setTrackCount(count int) {
 	s.tracks = make([]*rtspTrack, count)
 }
 
-func (s *sink) addTrack(index int, tcp bool) (int, int, error) {
+func (s *sink) addTrack(index int, tcp bool, ssrc uint32) (int, int, error) {
 	utils.Assert(index < cap(s.tracks))
 	utils.Assert(s.tracks[index] == nil)
 
@@ -43,7 +47,9 @@ func (s *sink) addTrack(index int, tcp bool) (int, int, error) {
 	var rtpPort int
 	var rtcpPort int
 
-	track := rtspTrack{}
+	track := rtspTrack{
+		ssrc: ssrc,
+	}
 	if tcp {
 		s.tcp = true
 	} else {
@@ -83,14 +89,38 @@ func (s *sink) addTrack(index int, tcp bool) (int, int, error) {
 	return rtpPort, rtcpPort, err
 }
 
-func (s *sink) input(index int, data []byte) error {
+func (s *sink) input(index int, data []byte, rtpTime uint32) error {
-	utils.Assert(index < cap(s.tracks))
 	//拉流方还没有连上来
-	s.tracks[index].pktCount++
+	utils.Assert(index < cap(s.tracks))
+
+	track := s.tracks[index]
+	track.pktCount++
+	track.octetCount += len(data)
 	if s.tcp {
 		s.Conn.Write(data)
 	} else {
-		s.tracks[index].rtpConn.Write(data)
+		track.rtpConn.Write(data)
+
+		if track.rtcpConn == nil || track.pktCount%100 != 0 {
+			return nil
+		}
+
+		nano := uint64(time.Now().UnixNano())
+		ntp := (nano/1000000000 + 2208988800<<32) | (nano % 1000000000)
+		sr := rtcp.SenderReport{
+			SSRC:        track.ssrc,
+			NTPTime:     ntp,
+			RTPTime:     rtpTime,
+			PacketCount: uint32(track.pktCount),
+			OctetCount:  uint32(track.octetCount),
+		}
+
+		marshal, err := sr.Marshal()
+		if err != nil {
+			log.Sugar.Errorf("创建rtcp sr消息失败 err:%s msg:%v", err.Error(), sr)
+		}
+
+		track.rtcpConn.Write(marshal)
 	}
 	return nil
 }
@@ -117,6 +147,8 @@ func (s *sink) TrackConnected(index int) bool {
 }
 
 func (s *sink) Close() {
+	s.SinkImpl.Close()
+
 	for _, track := range s.tracks {
 		if track.rtp != nil {
 			track.rtp.Close()

rtsp/rtsp_stream.go

@@ -73,9 +73,9 @@ func (t *tranStream) onRtpPacket(data []byte, timestamp uint32, params interface
 			for i, rtp := range track.header {
 				librtp.RollbackSeq(rtp[OverTcpHeaderSize:], int(seq)-(count-i-1))
 				if sink_.tcp {
-					sink_.input(index, rtp)
+					sink_.input(index, rtp, 0)
 				} else {
-					sink_.input(index, rtp[OverTcpHeaderSize:])
+					sink_.input(index, rtp[OverTcpHeaderSize:], timestamp)
 				}
 			}
 		}
@@ -84,9 +84,9 @@ func (t *tranStream) onRtpPacket(data []byte, timestamp uint32, params interface
 		t.overTCP(track.buffer[:end], index)
 
 		if sink_.tcp {
-			sink_.input(index, track.buffer[:end])
+			sink_.input(index, track.buffer[:end], 0)
 		} else {
-			sink_.input(index, data)
+			sink_.input(index, data, timestamp)
 		}
 	}
 }
@@ -133,8 +133,8 @@ func (t *tranStream) Input(packet utils.AVPacket) error {
 }
 
 func (t *tranStream) AddSink(sink_ stream.ISink) error {
-	sink_.(*Sink).setTrackCount(len(t.TransStreamImpl.Tracks))
+	sink_.(*sink).setTrackCount(len(t.TransStreamImpl.Tracks))
-	if err := sink_.(*Sink).SendHeader([]byte(t.sdp)); err != nil {
+	if err := sink_.(*sink).SendHeader([]byte(t.sdp)); err != nil {
 		return err
 	}
 

rtsp/rtsp_track.go

@@ -13,7 +13,9 @@ type rtspTrack struct {
 	rtcpConn net.Conn
 
 	//rtcp
-	pktCount int
+	pktCount   int
+	ssrc       uint32
+	octetCount int
 }
 
 func (s *rtspTrack) onRTPPacket(conn net.Conn, data []byte) {
@@ -26,6 +28,19 @@ func (s *rtspTrack) onRTCPPacket(conn net.Conn, data []byte) {
 	if s.rtcpConn == nil {
 		s.rtcpConn = conn
 	}
+
+	//packs, err := rtcp.Unmarshal(data)
+	//if err != nil {
+	//	log.Sugar.Warnf("解析rtcp包失败 err:%s conn:%s pkt:%s", err.Error(), conn.RemoteAddr().String(), hex.EncodeToString(data))
+	//	return
+	//}
+	//
+	//for _, pkt := range packs {
+	//	if _, ok := pkt.(*rtcp.ReceiverReport); ok {
+	//	} else if _, ok := pkt.(*rtcp.SourceDescription); ok {
+	//	} else if _, ok := pkt.(*rtcp.Goodbye); ok {
+	//	}
+	//}
 }
 
 // tcp链接成功回调