use UDP in sender / receiver report tests

client_publish_test.go

@@ -72,6 +72,18 @@ func TestClientPublishSerial(t *testing.T) {
 				err = inTH.Read(req.Header["Transport"])
 				require.NoError(t, err)
 
+				var l1 net.PacketConn
+				var l2 net.PacketConn
+				if transport == "udp" {
+					l1, err = net.ListenPacket("udp", "localhost:34556")
+					require.NoError(t, err)
+					defer l1.Close()
+
+					l2, err = net.ListenPacket("udp", "localhost:34557")
+					require.NoError(t, err)
+					defer l2.Close()
+				}
+
 				th := headers.Transport{
 					Delivery: func() *headers.TransportDelivery {
 						v := headers.TransportDeliveryUnicast
@@ -96,18 +108,6 @@ func TestClientPublishSerial(t *testing.T) {
 				}.Write(bconn.Writer)
 				require.NoError(t, err)
 
-				var l1 net.PacketConn
-				var l2 net.PacketConn
-				if transport == "udp" {
-					l1, err = net.ListenPacket("udp", "localhost:34556")
-					require.NoError(t, err)
-					defer l1.Close()
-
-					l2, err = net.ListenPacket("udp", "localhost:34557")
-					require.NoError(t, err)
-					defer l2.Close()
-				}
-
 				req, err = readRequest(bconn.Reader)
 				require.NoError(t, err)
 				require.Equal(t, base.Record, req.Method)
@@ -755,6 +755,8 @@ func TestClientPublishAutomaticProtocol(t *testing.T) {
 }
 
 func TestClientPublishRTCPReport(t *testing.T) {
+	reportReceived := make(chan struct{})
+
 	l, err := net.Listen("tcp", "localhost:8554")
 	require.NoError(t, err)
 	defer l.Close()
@@ -802,19 +804,26 @@ func TestClientPublishRTCPReport(t *testing.T) {
 		err = inTH.Read(req.Header["Transport"])
 		require.NoError(t, err)
 
-		th := headers.Transport{
-			Delivery: func() *headers.TransportDelivery {
-				v := headers.TransportDeliveryUnicast
-				return &v
-			}(),
-			Protocol:       headers.TransportProtocolTCP,
-			InterleavedIDs: inTH.InterleavedIDs,
-		}
+		l1, err := net.ListenPacket("udp", "localhost:34556")
+		require.NoError(t, err)
+		defer l1.Close()
+
+		l2, err := net.ListenPacket("udp", "localhost:34557")
+		require.NoError(t, err)
+		defer l2.Close()
 
 		err = base.Response{
 			StatusCode: base.StatusOK,
 			Header: base.Header{
-				"Transport": th.Write(),
+				"Transport": headers.Transport{
+					Delivery: func() *headers.TransportDelivery {
+						v := headers.TransportDeliveryUnicast
+						return &v
+					}(),
+					Protocol:    headers.TransportProtocolUDP,
+					ClientPorts: inTH.ClientPorts,
+					ServerPorts: &[2]int{34556, 34557},
+				}.Write(),
 			},
 		}.Write(bconn.Writer)
 		require.NoError(t, err)
@@ -830,18 +839,15 @@ func TestClientPublishRTCPReport(t *testing.T) {
 
 		rr := rtcpreceiver.New(nil, 90000)
 
-		var f base.InterleavedFrame
-		f.Payload = make([]byte, 2048)
-		err = f.Read(bconn.Reader)
+		buf := make([]byte, 2048)
+		n, _, err := l1.ReadFrom(buf)
 		require.NoError(t, err)
-		require.Equal(t, 0, f.Channel)
-		rr.ProcessPacketRTP(time.Now(), f.Payload)
+		rr.ProcessPacketRTP(time.Now(), buf[:n])
 
-		f.Payload = make([]byte, 2048)
-		err = f.Read(bconn.Reader)
+		buf = make([]byte, 2048)
+		n, _, err = l2.ReadFrom(buf)
 		require.NoError(t, err)
-		require.Equal(t, 1, f.Channel)
-		pkt, err := rtcp.Unmarshal(f.Payload)
+		pkt, err := rtcp.Unmarshal(buf[:n])
 		require.NoError(t, err)
 		sr, ok := pkt[0].(*rtcp.SenderReport)
 		require.True(t, ok)
@@ -852,18 +858,9 @@ func TestClientPublishRTCPReport(t *testing.T) {
 			PacketCount: 1,
 			OctetCount:  4,
 		}, sr)
-		rr.ProcessPacketRTCP(time.Now(), f.Payload)
+		rr.ProcessPacketRTCP(time.Now(), buf[:n])
 
-		err = base.InterleavedFrame{
-			Channel: 1,
-			Payload: rr.Report(time.Now()),
-		}.Write(bconn.Writer)
-		require.NoError(t, err)
-
-		f.Payload = make([]byte, 2048)
-		err = f.Read(bconn.Reader)
-		require.NoError(t, err)
-		require.Equal(t, 0, f.Channel)
+		close(reportReceived)
 
 		req, err = readRequest(bconn.Reader)
 		require.NoError(t, err)
@@ -875,10 +872,6 @@ func TestClientPublishRTCPReport(t *testing.T) {
 	}()
 
 	c := &Client{
-		Transport: func() *Transport {
-			v := TransportTCP
-			return &v
-		}(),
 		senderReportPeriod: 1 * time.Second,
 	}
 
@@ -904,10 +897,7 @@ func TestClientPublishRTCPReport(t *testing.T) {
 	err = c.WritePacketRTP(0, byts)
 	require.NoError(t, err)
 
-	time.Sleep(1300 * time.Millisecond)
-
-	err = c.WritePacketRTP(0, byts)
-	require.NoError(t, err)
+	<-reportReceived
 }
 
 func TestClientPublishIgnoreTCPRTPPackets(t *testing.T) {

client_read_test.go

@@ -891,7 +891,7 @@ func TestClientReadAnyPort(t *testing.T) {
 				}.Write(bconn.Writer)
 				require.NoError(t, err)
 
-				time.Sleep(1 * time.Second)
+				time.Sleep(500 * time.Millisecond)
 
 				l1, err := net.ListenPacket("udp", "localhost:13344")
 				require.NoError(t, err)
@@ -1480,7 +1480,7 @@ func TestClientReadRedirect(t *testing.T) {
 		}.Write(bconn.Writer)
 		require.NoError(t, err)
 
-		time.Sleep(1 * time.Second)
+		time.Sleep(500 * time.Millisecond)
 
 		l1, err := net.ListenPacket("udp", "localhost:34556")
 		require.NoError(t, err)
@@ -1723,6 +1723,8 @@ func TestClientReadPause(t *testing.T) {
 }
 
 func TestClientReadRTCPReport(t *testing.T) {
+	reportReceived := make(chan struct{})
+
 	l, err := net.Listen("tcp", "localhost:8554")
 	require.NoError(t, err)
 	defer l.Close()
@@ -1776,21 +1778,29 @@ func TestClientReadRTCPReport(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, base.Setup, req.Method)
 
-		var th headers.Transport
-		err = th.Read(req.Header["Transport"])
+		var inTH headers.Transport
+		err = inTH.Read(req.Header["Transport"])
 		require.NoError(t, err)
 
+		l1, err := net.ListenPacket("udp", "localhost:27556")
+		require.NoError(t, err)
+		defer l1.Close()
+
+		l2, err := net.ListenPacket("udp", "localhost:27557")
+		require.NoError(t, err)
+		defer l2.Close()
+
 		err = base.Response{
 			StatusCode: base.StatusOK,
 			Header: base.Header{
 				"Transport": headers.Transport{
-					Protocol: headers.TransportProtocolTCP,
+					Protocol: headers.TransportProtocolUDP,
 					Delivery: func() *headers.TransportDelivery {
 						v := headers.TransportDeliveryUnicast
 						return &v
 					}(),
-					ClientPorts:    th.ClientPorts,
-					InterleavedIDs: &[2]int{0, 1},
+					ServerPorts: &[2]int{27556, 27557},
+					ClientPorts: inTH.ClientPorts,
 				}.Write(),
 			},
 		}.Write(bconn.Writer)
@@ -1805,6 +1815,11 @@ func TestClientReadRTCPReport(t *testing.T) {
 		}.Write(bconn.Writer)
 		require.NoError(t, err)
 
+		// skip firewall opening
+		buf := make([]byte, 2048)
+		_, _, err = l2.ReadFrom(buf)
+		require.NoError(t, err)
+
 		rs := rtcpsender.New(90000)
 
 		byts, _ := (&rtp.Packet{
@@ -1818,25 +1833,23 @@ func TestClientReadRTCPReport(t *testing.T) {
 			},
 			Payload: []byte{0x01, 0x02, 0x03, 0x04},
 		}).Marshal()
-		err = base.InterleavedFrame{
-			Channel: 0,
-			Payload: byts,
-		}.Write(bconn.Writer)
+		_, err = l1.WriteTo(byts, &net.UDPAddr{
+			IP:   net.ParseIP("127.0.0.1"),
+			Port: inTH.ClientPorts[0],
+		})
 		require.NoError(t, err)
 		rs.ProcessPacketRTP(time.Now(), byts)
 
-		err = base.InterleavedFrame{
-			Channel: 1,
-			Payload: rs.Report(time.Now()),
-		}.Write(bconn.Writer)
+		_, err = l2.WriteTo(rs.Report(time.Now()), &net.UDPAddr{
+			IP:   net.ParseIP("127.0.0.1"),
+			Port: inTH.ClientPorts[1],
+		})
 		require.NoError(t, err)
 
-		var f base.InterleavedFrame
-		f.Payload = make([]byte, 2048)
-		err = f.Read(bconn.Reader)
+		buf = make([]byte, 2048)
+		n, _, err := l2.ReadFrom(buf)
 		require.NoError(t, err)
-		require.Equal(t, 1, f.Channel)
-		pkt, err := rtcp.Unmarshal(f.Payload)
+		pkt, err := rtcp.Unmarshal(buf[:n])
 		require.NoError(t, err)
 		rr, ok := pkt[0].(*rtcp.ReceiverReport)
 		require.True(t, ok)
@@ -1853,33 +1866,18 @@ func TestClientReadRTCPReport(t *testing.T) {
 			ProfileExtensions: []uint8{},
 		}, rr)
 
-		err = base.InterleavedFrame{
-			Channel: 0,
-			Payload: byts,
-		}.Write(bconn.Writer)
+		close(reportReceived)
+
+		req, err = readRequest(bconn.Reader)
 		require.NoError(t, err)
+		require.Equal(t, base.Teardown, req.Method)
+
+		base.Response{
+			StatusCode: base.StatusOK,
+		}.Write(bconn.Writer)
 	}()
 
-	recv := 0
-	recvDone := make(chan struct{})
-
 	c := &Client{
-		Transport: func() *Transport {
-			v := TransportTCP
-			return &v
-		}(),
-		OnPacketRTP: func(trackID int, payload []byte) {
-			recv++
-			if recv >= 3 {
-				close(recvDone)
-			}
-		},
-		OnPacketRTCP: func(trackID int, payload []byte) {
-			recv++
-			if recv >= 3 {
-				close(recvDone)
-			}
-		},
 		receiverReportPeriod: 1 * time.Second,
 	}
 
@@ -1887,7 +1885,7 @@ func TestClientReadRTCPReport(t *testing.T) {
 	require.NoError(t, err)
 	defer c.Close()
 
-	<-recvDone
+	<-reportReceived
 }
 
 func TestClientReadErrorTimeout(t *testing.T) {
@@ -1964,7 +1962,7 @@ func TestClientReadErrorTimeout(t *testing.T) {
 				var l1 net.PacketConn
 				if transport == "udp" || transport == "auto" {
 					var err error
-					l1, err = net.ListenPacket("udp", "localhost:34557")
+					l1, err = net.ListenPacket("udp", "localhost:34556")
 					require.NoError(t, err)
 					defer l1.Close()
 
@@ -1994,12 +1992,6 @@ func TestClientReadErrorTimeout(t *testing.T) {
 				require.NoError(t, err)
 
 				if transport == "udp" || transport == "auto" {
-					time.Sleep(500 * time.Millisecond)
-
-					l1, err := net.ListenPacket("udp", "localhost:34556")
-					require.NoError(t, err)
-					defer l1.Close()
-
 					// write a packet to skip the protocol autodetection feature
 					l1.WriteTo([]byte("\x01\x02\x03\x04"), &net.UDPAddr{
 						IP:   net.ParseIP("127.0.0.1"),

server_publish_test.go

@@ -722,9 +722,20 @@ func TestServerPublish(t *testing.T) {
 				}(),
 			}
 
+			var l1 net.PacketConn
+			var l2 net.PacketConn
+
 			if transport == "udp" {
 				inTH.Protocol = headers.TransportProtocolUDP
 				inTH.ClientPorts = &[2]int{35466, 35467}
+
+				l1, err = net.ListenPacket("udp", "localhost:35466")
+				require.NoError(t, err)
+				defer l1.Close()
+
+				l2, err = net.ListenPacket("udp", "localhost:35467")
+				require.NoError(t, err)
+				defer l2.Close()
 			} else {
 				inTH.Protocol = headers.TransportProtocolTCP
 				inTH.InterleavedIDs = &[2]int{0, 1}
@@ -746,18 +757,6 @@ func TestServerPublish(t *testing.T) {
 			err = th.Read(res.Header["Transport"])
 			require.NoError(t, err)
 
-			var l1 net.PacketConn
-			var l2 net.PacketConn
-			if transport == "udp" {
-				l1, err = net.ListenPacket("udp", "localhost:35466")
-				require.NoError(t, err)
-				defer l1.Close()
-
-				l2, err = net.ListenPacket("udp", "localhost:35467")
-				require.NoError(t, err)
-				defer l2.Close()
-			}
-
 			res, err = writeReqReadRes(bconn, base.Request{
 				Method: base.Record,
 				URL:    mustParseURL("rtsp://localhost:8554/teststream"),
@@ -1068,6 +1067,8 @@ func TestServerPublishRTCPReport(t *testing.T) {
 			},
 		},
 		receiverReportPeriod: 1 * time.Second,
+		UDPRTPAddress:        "127.0.0.1:8000",
+		UDPRTCPAddress:       "127.0.0.1:8001",
 		RTSPAddress:          "localhost:8554",
 	}
 
@@ -1103,26 +1104,32 @@ func TestServerPublishRTCPReport(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, base.StatusOK, res.StatusCode)
 
-	inTH := &headers.Transport{
-		Delivery: func() *headers.TransportDelivery {
-			v := headers.TransportDeliveryUnicast
-			return &v
-		}(),
-		Mode: func() *headers.TransportMode {
-			v := headers.TransportModeRecord
-			return &v
-		}(),
-		Protocol:       headers.TransportProtocolTCP,
-		InterleavedIDs: &[2]int{0, 1},
-	}
+	l1, err := net.ListenPacket("udp", "localhost:34556")
+	require.NoError(t, err)
+	defer l1.Close()
+
+	l2, err := net.ListenPacket("udp", "localhost:34557")
+	require.NoError(t, err)
+	defer l2.Close()
 
 	res, err = writeReqReadRes(bconn, base.Request{
 		Method: base.Setup,
 		URL:    mustParseURL("rtsp://localhost:8554/teststream/trackID=0"),
 		Header: base.Header{
-			"CSeq":      base.HeaderValue{"2"},
-			"Transport": inTH.Write(),
-			"Session":   res.Header["Session"],
+			"CSeq": base.HeaderValue{"2"},
+			"Transport": headers.Transport{
+				Delivery: func() *headers.TransportDelivery {
+					v := headers.TransportDeliveryUnicast
+					return &v
+				}(),
+				Mode: func() *headers.TransportMode {
+					v := headers.TransportModeRecord
+					return &v
+				}(),
+				Protocol:    headers.TransportProtocolUDP,
+				ClientPorts: &[2]int{34556, 34557},
+			}.Write(),
+			"Session": res.Header["Session"],
 		},
 	})
 	require.NoError(t, err)
@@ -1154,18 +1161,21 @@ func TestServerPublishRTCPReport(t *testing.T) {
 		},
 		Payload: []byte{0x01, 0x02, 0x03, 0x04},
 	}).Marshal()
-	err = base.InterleavedFrame{
-		Channel: 0,
-		Payload: byts,
-	}.Write(bconn.Writer)
+	_, err = l1.WriteTo(byts, &net.UDPAddr{
+		IP:   net.ParseIP("127.0.0.1"),
+		Port: th.ServerPorts[0],
+	})
 	require.NoError(t, err)
 
-	var f base.InterleavedFrame
-	f.Payload = make([]byte, 2048)
-	f.Read(bconn.Reader)
+	// skip firewall opening
+	buf := make([]byte, 2048)
+	_, _, err = l2.ReadFrom(buf)
 	require.NoError(t, err)
-	require.Equal(t, 1, f.Channel)
-	pkt, err := rtcp.Unmarshal(f.Payload)
+
+	buf = make([]byte, 2048)
+	n, _, err := l2.ReadFrom(buf)
+	require.NoError(t, err)
+	pkt, err := rtcp.Unmarshal(buf[:n])
 	require.NoError(t, err)
 	rr, ok := pkt[0].(*rtcp.ReceiverReport)
 	require.True(t, ok)
@@ -1181,12 +1191,6 @@ func TestServerPublishRTCPReport(t *testing.T) {
 		},
 		ProfileExtensions: []uint8{},
 	}, rr)
-
-	err = base.InterleavedFrame{
-		Channel: 0,
-		Payload: byts,
-	}.Write(bconn.Writer)
-	require.NoError(t, err)
 }
 
 func TestServerPublishTimeout(t *testing.T) {

server_read_test.go

@@ -399,26 +399,14 @@ func TestServerRead(t *testing.T) {
 			err = th.Read(res.Header["Transport"])
 			require.NoError(t, err)
 
-			switch transport {
-			case "udp":
-				require.Equal(t, headers.TransportProtocolUDP, th.Protocol)
-				require.Equal(t, headers.TransportDeliveryUnicast, *th.Delivery)
-
-			case "multicast":
-				require.Equal(t, headers.TransportProtocolUDP, th.Protocol)
-				require.Equal(t, headers.TransportDeliveryMulticast, *th.Delivery)
-
-			default:
-				require.Equal(t, headers.TransportProtocolTCP, th.Protocol)
-				require.Equal(t, headers.TransportDeliveryUnicast, *th.Delivery)
-			}
-
-			<-sessionOpened
-
 			var l1 net.PacketConn
 			var l2 net.PacketConn
+
 			switch transport {
 			case "udp":
+				require.Equal(t, headers.TransportProtocolUDP, th.Protocol)
+				require.Equal(t, headers.TransportDeliveryUnicast, *th.Delivery)
+
 				l1, err = net.ListenPacket("udp", listenIP+":35466")
 				require.NoError(t, err)
 				defer l1.Close()
@@ -428,6 +416,9 @@ func TestServerRead(t *testing.T) {
 				defer l2.Close()
 
 			case "multicast":
+				require.Equal(t, headers.TransportProtocolUDP, th.Protocol)
+				require.Equal(t, headers.TransportDeliveryMulticast, *th.Delivery)
+
 				l1, err = net.ListenPacket("udp", "224.0.0.0:"+strconv.FormatInt(int64(th.Ports[0]), 10))
 				require.NoError(t, err)
 				defer l1.Close()
@@ -455,8 +446,14 @@ func TestServerRead(t *testing.T) {
 					err := p.JoinGroup(&intf, &net.UDPAddr{IP: *th.Destination})
 					require.NoError(t, err)
 				}
+
+			default:
+				require.Equal(t, headers.TransportProtocolTCP, th.Protocol)
+				require.Equal(t, headers.TransportDeliveryUnicast, *th.Delivery)
 			}
 
+			<-sessionOpened
+
 			res, err = writeReqReadRes(bconn, base.Request{
 				Method: base.Play,
 				URL:    mustParseURL("rtsp://" + listenIP + ":8554/teststream"),