Repository: bluenviron/gortsplib
Pull request #357 (2023-08-14 09:35:48)

server: allow WritePacket() to return an error

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/bluenviron/gortsplib/pull/357?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) Report
> Merging [#357](https://app.codecov.io/gh/bluenviron/gortsplib/pull/357?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) (b475c18) into [main](https://app.codecov.io/gh/bluenviron/gortsplib/commit/18ddae4c507e5536443ee5f29d3388b1de5cfe30?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) (18ddae4) will **increase** coverage by `0.00%`.
> The diff coverage is `73.33%`.

```diff
@@           Coverage Diff           @@
##             main     #357   +/-   ##
=======================================
  Coverage   82.63%   82.63%           
=======================================
  Files         101      101           
  Lines       10457    10459    +2     
=======================================
+ Hits         8641     8643    +2     
  Misses       1435     1435           
  Partials      381      381           
```


| [Files Changed](https://app.codecov.io/gh/bluenviron/gortsplib/pull/357?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) | Coverage Δ | |
|---|---|---|
| [server\_stream\_media.go](https://app.codecov.io/gh/bluenviron/gortsplib/pull/357?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron#diff-c2VydmVyX3N0cmVhbV9tZWRpYS5nbw==) | `83.33% <71.42%> (+0.47%)` | :arrow_up: |
| [server\_stream.go](https://app.codecov.io/gh/bluenviron/gortsplib/pull/357?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron#diff-c2VydmVyX3N0cmVhbS5nbw==) | `83.87% <75.00%> (ø)` | |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron)


Patch:
diff --git a/server_stream.go b/server_stream.go
index 83edcb8a..70205a6e 100644
--- a/server_stream.go
+++ b/server_stream.go
@@ -252,34 +252,34 @@ func (st *ServerStream) readerSetInactive(ss *ServerSession) {
 }
 
 // WritePacketRTP writes a RTP packet to all the readers of the stream.
-func (st *ServerStream) WritePacketRTP(medi *media.Media, pkt *rtp.Packet) {
-	st.WritePacketRTPWithNTP(medi, pkt, time.Now())
+func (st *ServerStream) WritePacketRTP(medi *media.Media, pkt *rtp.Packet) error {
+	return st.WritePacketRTPWithNTP(medi, pkt, time.Now())
 }
 
 // WritePacketRTPWithNTP writes a RTP packet to all the readers of the stream.
 // ntp is the absolute time of the packet, and is needed to generate RTCP sender reports
 // that allows the receiver to reconstruct the absolute time of the packet.
-func (st *ServerStream) WritePacketRTPWithNTP(medi *media.Media, pkt *rtp.Packet, ntp time.Time) {
+func (st *ServerStream) WritePacketRTPWithNTP(medi *media.Media, pkt *rtp.Packet, ntp time.Time) error {
 	st.mutex.RLock()
 	defer st.mutex.RUnlock()
 
 	if st.closed {
-		return
+		return fmt.Errorf("stream is closed")
 	}
 
 	sm := st.streamMedias[medi]
-	sm.WritePacketRTPWithNTP(st, pkt, ntp)
+	return sm.writePacketRTPWithNTP(st, pkt, ntp)
 }
 
 // WritePacketRTCP writes a RTCP packet to all the readers of the stream.
-func (st *ServerStream) WritePacketRTCP(medi *media.Media, pkt rtcp.Packet) {
+func (st *ServerStream) WritePacketRTCP(medi *media.Media, pkt rtcp.Packet) error {
 	st.mutex.RLock()
 	defer st.mutex.RUnlock()
 
 	if st.closed {
-		return
+		return fmt.Errorf("stream is closed")
 	}
 
 	sm := st.streamMedias[medi]
-	sm.writePacketRTCP(st, pkt)
+	return sm.writePacketRTCP(st, pkt)
 }
diff --git a/server_stream_media.go b/server_stream_media.go
index f9f711b0..9420cf61 100644
--- a/server_stream_media.go
+++ b/server_stream_media.go
@@ -33,7 +33,7 @@ func newServerStreamMedia(st *ServerStream, medi *media.Media, trackID int) *ser
 		tr.rtcpSender = rtcpsender.New(
 			forma.ClockRate(),
 			func(pkt rtcp.Packet) {
-				st.WritePacketRTCP(cmedia, pkt)
+				st.WritePacketRTCP(cmedia, pkt) //nolint:errcheck
 			},
 		)
 
@@ -67,11 +67,11 @@ func (sm *serverStreamMedia) allocateMulticastHandler(s *Server) error {
 	return nil
 }
 
-func (sm *serverStreamMedia) WritePacketRTPWithNTP(ss *ServerStream, pkt *rtp.Packet, ntp time.Time) {
+func (sm *serverStreamMedia) writePacketRTPWithNTP(ss *ServerStream, pkt *rtp.Packet, ntp time.Time) error {
 	byts := make([]byte, udpMaxPayloadSize)
 	n, err := pkt.MarshalTo(byts)
 	if err != nil {
-		return
+		return err
 	}
 	byts = byts[:n]
 
@@ -91,12 +91,14 @@ func (sm *serverStreamMedia) WritePacketRTPWithNTP(ss *ServerStream, pkt *rtp.Pa
 	if sm.multicastWriter != nil {
 		sm.multicastWriter.writePacketRTP(byts)
 	}
+
+	return nil
 }
 
-func (sm *serverStreamMedia) writePacketRTCP(ss *ServerStream, pkt rtcp.Packet) {
+func (sm *serverStreamMedia) writePacketRTCP(ss *ServerStream, pkt rtcp.Packet) error {
 	byts, err := pkt.Marshal()
 	if err != nil {
-		return
+		return err
 	}
 
 	// send unicast
@@ -111,4 +113,6 @@ func (sm *serverStreamMedia) writePacketRTCP(ss *ServerStream, pkt rtcp.Packet)
 	if sm.multicastWriter != nil {
 		sm.multicastWriter.writePacketRTCP(byts)
 	}
+
+	return nil
 }


Test patch:
diff --git a/server_play_test.go b/server_play_test.go
index dbdee90b..24c467fc 100644
--- a/server_play_test.go
+++ b/server_play_test.go
@@ -568,9 +568,11 @@ func TestServerPlay(t *testing.T) {
 						// with StatusOK; therefore we must wait before calling
 						// ServerStream.WritePacket*()
 						go func() {
-							time.Sleep(1 * time.Second)
-							stream.WritePacketRTCP(stream.Medias()[0], &testRTCPPacket)
-							stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+							time.Sleep(500 * time.Millisecond)
+							err := stream.WritePacketRTCP(stream.Medias()[0], &testRTCPPacket)
+							require.NoError(t, err)
+							err = stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+							require.NoError(t, err)
 						}()
 
 						return &base.Response{
@@ -1042,7 +1044,7 @@ func TestServerPlayRTCPReport(t *testing.T) {
 			doPlay(t, conn, "rtsp://localhost:8554/teststream", session)
 
 			for i := 0; i < 2; i++ {
-				stream.WritePacketRTP(stream.Medias()[0], &rtp.Packet{
+				err := stream.WritePacketRTP(stream.Medias()[0], &rtp.Packet{
 					Header: rtp.Header{
 						Version:     2,
 						PayloadType: 96,
@@ -1050,6 +1052,7 @@ func TestServerPlayRTCPReport(t *testing.T) {
 					},
 					Payload: []byte{0x05}, // IDR
 				})
+				require.NoError(t, err)
 			}
 
 			var buf []byte
@@ -1161,15 +1164,17 @@ func TestServerPlayTCPResponseBeforeFrames(t *testing.T) {
 				go func() {
 					defer close(writerDone)
 
-					stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+					err := stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+					require.NoError(t, err)
 
-					t := time.NewTicker(50 * time.Millisecond)
-					defer t.Stop()
+					ti := time.NewTicker(50 * time.Millisecond)
+					defer ti.Stop()
 
 					for {
 						select {
-						case <-t.C:
-							stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+						case <-ti.C:
+							err := stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+							require.NoError(t, err)
 						case <-writerTerminate:
 							return
 						}
@@ -1308,13 +1313,14 @@ func TestServerPlayPlayPausePlay(t *testing.T) {
 					go func() {
 						defer close(writerDone)
 
-						t := time.NewTicker(50 * time.Millisecond)
-						defer t.Stop()
+						ti := time.NewTicker(50 * time.Millisecond)
+						defer ti.Stop()
 
 						for {
 							select {
-							case <-t.C:
-								stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+							case <-ti.C:
+								err := stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+								require.NoError(t, err)
 							case <-writerTerminate:
 								return
 							}
@@ -1395,13 +1401,14 @@ func TestServerPlayPlayPausePause(t *testing.T) {
 				go func() {
 					defer close(writerDone)
 
-					t := time.NewTicker(50 * time.Millisecond)
-					defer t.Stop()
+					ti := time.NewTicker(50 * time.Millisecond)
+					defer ti.Stop()
 
 					for {
 						select {
-						case <-t.C:
-							stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+						case <-ti.C:
+							err := stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+							require.NoError(t, err)
 						case <-writerTerminate:
 							return
 						}
@@ -1743,9 +1750,11 @@ func TestServerPlayPartialMedias(t *testing.T) {
 			},
 			onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
 				go func() {
-					time.Sleep(1 * time.Second)
-					stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
-					stream.WritePacketRTP(stream.Medias()[1], &testRTPPacket)
+					time.Sleep(500 * time.Millisecond)
+					err := stream.WritePacketRTP(stream.Medias()[0], &testRTPPacket)
+					require.NoError(t, err)
+					err = stream.WritePacketRTP(stream.Medias()[1], &testRTPPacket)
+					require.NoError(t, err)
 				}()
 
 				return &base.Response{
@@ -1891,7 +1900,7 @@ func TestServerPlayAdditionalInfos(t *testing.T) {
 	require.NoError(t, err)
 	defer s.Close()
 
-	stream.WritePacketRTP(stream.Medias()[0], &rtp.Packet{
+	err = stream.WritePacketRTP(stream.Medias()[0], &rtp.Packet{
 		Header: rtp.Header{
 			Version:        2,
 			PayloadType:    96,
@@ -1901,6 +1910,7 @@ func TestServerPlayAdditionalInfos(t *testing.T) {
 		},
 		Payload: []byte{0x01, 0x02, 0x03, 0x04},
 	})
+	require.NoError(t, err)
 
 	rtpInfo, ssrcs := getInfos()
 	require.True(t, strings.HasPrefix(mustParseURL((*rtpInfo)[0].URL).Path, "/teststream/trackID="))
@@ -1920,7 +1930,7 @@ func TestServerPlayAdditionalInfos(t *testing.T) {
 		nil,
 	}, ssrcs)
 
-	stream.WritePacketRTP(stream.Medias()[1], &rtp.Packet{
+	err = stream.WritePacketRTP(stream.Medias()[1], &rtp.Packet{
 		Header: rtp.Header{
 			Version:        2,
 			PayloadType:    96,
@@ -1930,6 +1940,7 @@ func TestServerPlayAdditionalInfos(t *testing.T) {
 		},
 		Payload: []byte{0x01, 0x02, 0x03, 0x04},
 	})
+	require.NoError(t, err)
 
 	rtpInfo, ssrcs = getInfos()
 	require.True(t, strings.HasPrefix(mustParseURL((*rtpInfo)[0].URL).Path, "/teststream/trackID="))
@@ -2036,7 +2047,8 @@ func TestServerPlayNoInterleavedIDs(t *testing.T) {
 	doPlay(t, conn, "rtsp://localhost:8554/teststream", session)
 
 	for i := 0; i < 2; i++ {
-		stream.WritePacketRTP(stream.Medias()[i], &testRTPPacket)
+		err := stream.WritePacketRTP(stream.Medias()[i], &testRTPPacket)
+		require.NoError(t, err)
 
 		f, err := conn.ReadInterleavedFrame()
 		require.NoError(t, err)