diff --git a/client_publish_test.go b/client_publish_test.go index f36bc714..5d4af72a 100644 --- a/client_publish_test.go +++ b/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) { diff --git a/client_read_test.go b/client_read_test.go index f4c83ef9..b0418384 100644 --- a/client_read_test.go +++ b/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"), diff --git a/server_publish_test.go b/server_publish_test.go index 9132819f..6d9111ce 100644 --- a/server_publish_test.go +++ b/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) { diff --git a/server_read_test.go b/server_read_test.go index 16b378e2..e7837364 100644 --- a/server_read_test.go +++ b/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"),