Support stream-specific multicast IP address and RTP/RTCP ports (#965)

Co-authored-by: aler9 <[email protected]>

Author: kelvinlawson

pkg/rtpreceiver/receiver.go

@@ -11,7 +11,8 @@ import (
 	"github.com/pion/rtp"
 )
 
-// Receiver is a utility to receive RTP packets. It is in charge of:
+// Receiver is a utility to receive RTP packets.
+// It is in charge of:
 // - removing duplicate packets (when transport is unreliable)
 // - reordering packets (when transport is unrealiable)
 // - counting received and lost packets

pkg/rtpsender/sender.go

@@ -11,7 +11,7 @@ import (
 )
 
 // Sender is a utility to send RTP packets.
-// It is in charge of
+// It is in charge of:
 // - counting sent packets
 // - generating RTCP sender reports.
 type Sender struct {

server_multicast_writer.go

@@ -3,13 +3,20 @@ package gortsplib
 import (
 	"context"
 	"net"
+	"time"
 
 	"github.com/bluenviron/gortsplib/v5/internal/asyncprocessor"
 	"github.com/bluenviron/gortsplib/v5/pkg/liberrors"
 )
 
 type serverMulticastWriter struct {
-	s *Server
+	udpReadBufferSize int
+	listenPacket      func(network string, address string) (net.PacketConn, error)
+	writeQueueSize    int
+	writeTimeout      time.Duration
+	ip                net.IP
+	rtpPort           int
+	rtcpPort          int
 
 	rtpl     *serverUDPListener
 	rtcpl    *serverUDPListener
@@ -19,31 +26,26 @@ type serverMulticastWriter struct {
 }
 
 func (h *serverMulticastWriter) initialize() error {
-	ip, err := h.s.getMulticastIP()
-	if err != nil {
-		return err
-	}
-
 	rtpl, rtcpl, err := createUDPListenerMulticastPair(
-		h.s.UDPReadBufferSize,
-		h.s.ListenPacket,
-		h.s.WriteTimeout,
-		h.s.MulticastRTPPort,
-		h.s.MulticastRTCPPort,
-		ip,
+		h.udpReadBufferSize,
+		h.listenPacket,
+		h.writeTimeout,
+		h.rtpPort,
+		h.rtcpPort,
+		h.ip,
 	)
 	if err != nil {
 		return err
 	}
 
 	rtpAddr := &net.UDPAddr{
-		IP:   rtpl.ip(),
-		Port: rtpl.port(),
+		IP:   h.ip,
+		Port: h.rtpPort,
 	}
 
 	rtcpAddr := &net.UDPAddr{
-		IP:   rtcpl.ip(),
-		Port: rtcpl.port(),
+		IP:   h.ip,
+		Port: h.rtcpPort,
 	}
 
 	h.rtpl = rtpl
@@ -52,7 +54,7 @@ func (h *serverMulticastWriter) initialize() error {
 	h.rtcpAddr = rtcpAddr
 
 	h.writer = &asyncprocessor.Processor{
-		BufferSize: h.s.WriteQueueSize,
+		BufferSize: h.writeQueueSize,
 		OnError:    func(_ context.Context, _ error) {},
 	}
 	h.writer.Initialize()
@@ -67,10 +69,6 @@ func (h *serverMulticastWriter) close() {
 	h.writer.Close()
 }
 
-func (h *serverMulticastWriter) ip() net.IP {
-	return h.rtpl.ip()
-}
-
 func (h *serverMulticastWriter) writePacketRTP(byts []byte) error {
 	ok := h.writer.Push(func() error {
 		return h.rtpl.write(byts, h.rtpAddr)

server_play_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/bluenviron/gortsplib/v5/pkg/mikey"
 	"github.com/bluenviron/gortsplib/v5/pkg/ntp"
 	"github.com/bluenviron/gortsplib/v5/pkg/sdp"
+	"github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio"
 )
 
 func multicastCapableIP(t *testing.T) string {
@@ -2788,3 +2789,127 @@ func TestServerPlayBackChannel(t *testing.T) {
 		})
 	}
 }
+
+func TestServerPlayMulticastParams(t *testing.T) {
+	listenIP := multicastCapableIP(t)
+
+	var stream *ServerStream
+
+	s := &Server{
+		Handler: &testServerHandler{
+			onDescribe: func(_ *ServerHandlerOnDescribeCtx) (*base.Response, *ServerStream, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, stream, nil
+			},
+			onSetup: func(_ *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, stream, nil
+			},
+			onPlay: func(_ *ServerHandlerOnPlayCtx) (*base.Response, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, nil
+			},
+		},
+		RTSPAddress:       listenIP + ":8554",
+		MulticastIPRange:  "224.1.0.0/16",
+		MulticastRTPPort:  8000,
+		MulticastRTCPPort: 8001,
+	}
+
+	err := s.Start()
+	require.NoError(t, err)
+	defer s.Close()
+
+	media1 := &description.Media{
+		Type: description.MediaTypeVideo,
+		Formats: []format.Format{&format.H264{
+			PayloadTyp: 96,
+			SPS: []byte{
+				0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
+				0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
+				0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
+				0x20,
+			},
+			PPS: []byte{
+				0x44, 0x01, 0xc0, 0x25, 0x2f, 0x05, 0x32, 0x40,
+			},
+			PacketizationMode: 1,
+		}},
+	}
+
+	media2 := &description.Media{
+		Type: description.MediaTypeAudio,
+		Formats: []format.Format{&format.MPEG4Audio{
+			PayloadTyp: 97,
+			Config: &mpeg4audio.AudioSpecificConfig{
+				Type:         mpeg4audio.ObjectTypeAACLC,
+				SampleRate:   48000,
+				ChannelCount: 2,
+			},
+			SizeLength:       13,
+			IndexLength:      3,
+			IndexDeltaLength: 3,
+		}},
+	}
+
+	multicastParams := map[*description.Media]StreamMediaMulticastParams{
+		media1: {
+			IP:       net.ParseIP("224.2.0.50"),
+			RTPPort:  9000,
+			RTCPPort: 9001,
+		},
+		media2: {
+			IP:       net.ParseIP("224.2.0.51"),
+			RTPPort:  9002,
+			RTCPPort: 9003,
+		},
+	}
+
+	stream = &ServerStream{
+		Server: s,
+		Desc: &description.Session{
+			Medias: []*description.Media{media1, media2},
+		},
+		MulticastParams: multicastParams,
+	}
+	err = stream.Initialize()
+	require.NoError(t, err)
+	defer stream.Close()
+
+	nconn, err := net.Dial("tcp", listenIP+":8554")
+	require.NoError(t, err)
+	defer nconn.Close()
+	conn := conn.NewConn(bufio.NewReader(nconn), nconn)
+
+	desc := doDescribe(t, conn, false)
+
+	inTH1 := &headers.Transport{
+		Delivery: ptrOf(headers.TransportDeliveryMulticast),
+		Protocol: headers.TransportProtocolUDP,
+		Mode:     ptrOf(headers.TransportModePlay),
+	}
+
+	res1, th1 := doSetup(t, conn, mediaURL(t, desc.BaseURL, desc.Medias[0]).String(), inTH1, "")
+	session := readSession(t, res1)
+
+	require.Equal(t, headers.TransportProtocolUDP, th1.Protocol)
+	require.Equal(t, headers.TransportDeliveryMulticast, *th1.Delivery)
+	require.Equal(t, "224.2.0.50", *th1.Destination2)
+	require.Equal(t, [2]int{9000, 9001}, *th1.Ports)
+
+	inTH2 := &headers.Transport{
+		Delivery: ptrOf(headers.TransportDeliveryMulticast),
+		Protocol: headers.TransportProtocolUDP,
+		Mode:     ptrOf(headers.TransportModePlay),
+	}
+
+	_, th2 := doSetup(t, conn, mediaURL(t, desc.BaseURL, desc.Medias[1]).String(), inTH2, session)
+
+	require.Equal(t, headers.TransportProtocolUDP, th2.Protocol)
+	require.Equal(t, headers.TransportDeliveryMulticast, *th2.Delivery)
+	require.Equal(t, "224.2.0.51", *th2.Destination2)
+	require.Equal(t, [2]int{9002, 9003}, *th2.Ports)
+}

server_session.go

@@ -1304,8 +1304,8 @@ func (ss *ServerSession) handleRequestInner(sc *ServerConn, req *base.Request) (
 				} else {
 					th.Delivery = ptrOf(headers.TransportDeliveryMulticast)
 					th.TTL = ptrOf(uint(127))
-					th.Destination2 = ptrOf(stream.medias[medi].multicastWriter.ip().String())
-					th.Ports = &[2]int{ss.s.MulticastRTPPort, ss.s.MulticastRTCPPort}
+					th.Destination2 = ptrOf(stream.medias[medi].multicastWriter.ip.String())
+					th.Ports = &[2]int{stream.medias[medi].multicastWriter.rtpPort, stream.medias[medi].multicastWriter.rtcpPort}
 				}
 
 			default: // TCP

server_stream.go

@@ -3,6 +3,7 @@ package gortsplib
 import (
 	"crypto/rand"
 	"fmt"
+	"net"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -40,14 +41,27 @@ func serverStreamExtractExistingSSRCs(medias map[*description.Media]*serverStrea
 	return ret
 }
 
-// ServerStream represents a data stream.
+// StreamMediaMulticastParams used to request specific Multicast configuration for each media in a stream
+type StreamMediaMulticastParams struct {
+	IP       net.IP
+	RTPPort  int
+	RTCPPort int
+}
+
+// ServerStream represents a media stream.
 // This is in charge of
 // - storing stream description and statistics
 // - distributing the stream to each reader
 // - allocating multicast listeners
 type ServerStream struct {
+	// Parent server.
 	Server *Server
-	Desc   *description.Session
+
+	// Stream description.
+	Desc *description.Session
+
+	// (optional) Stream-specific Multicast settings.
+	MulticastParams map[*description.Media]StreamMediaMulticastParams
 
 	mutex                sync.RWMutex
 	readers              map[*ServerSession]struct{}
@@ -220,8 +234,33 @@ func (st *ServerStream) readerAdd(
 	case ProtocolUDPMulticast:
 		if st.multicastReaderCount == 0 {
 			for _, media := range st.medias {
+				var ip net.IP
+				var rtpPort int
+				var rtcpPort int
+
+				if params, ok := st.MulticastParams[media.media]; ok {
+					ip = params.IP
+					rtpPort = params.RTPPort
+					rtcpPort = params.RTCPPort
+				} else {
+					var err error
+					ip, err = st.Server.getMulticastIP()
+					if err != nil {
+						return err
+					}
+
+					rtpPort = st.Server.MulticastRTPPort
+					rtcpPort = st.Server.MulticastRTCPPort
+				}
+
 				mw := &serverMulticastWriter{
-					s: st.Server,
+					udpReadBufferSize: st.Server.UDPReadBufferSize,
+					listenPacket:      st.Server.ListenPacket,
+					writeQueueSize:    st.Server.WriteQueueSize,
+					writeTimeout:      st.Server.WriteTimeout,
+					ip:                ip,
+					rtpPort:           rtpPort,
+					rtcpPort:          rtcpPort,
 				}
 				err := mw.initialize()
 				if err != nil {

server_udp_listener.go

@@ -120,10 +120,6 @@ func (u *serverUDPListener) close() {
 	<-u.done
 }
 
-func (u *serverUDPListener) ip() net.IP {
-	return u.listenIP
-}
-
 func (u *serverUDPListener) port() int {
 	return u.pc.LocalAddr().(*net.UDPAddr).Port
 }