Repository: bluenviron/gortsplib
Pull request #457 (2023-11-03 20:48:01)

add BytesSent to ServerStream (to allow MediaMTX to gather bytes sent on all paths)
add BytesSent to ServerStream (to allow MediaMTX to gather bytes sent on all paths)

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/bluenviron/gortsplib/pull/457?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) Report
> Merging [#457](https://app.codecov.io/gh/bluenviron/gortsplib/pull/457?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) (1f2252a) into [main](https://app.codecov.io/gh/bluenviron/gortsplib/commit/35bf96c5ecf652d801c9dddeb74056e33ce0c284?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) (35bf96c) will **increase** coverage by `0.01%`.
> The diff coverage is `50.00%`.

```diff
@@            Coverage Diff             @@
##             main     #457      +/-   ##
==========================================
+ Coverage   84.50%   84.52%   +0.01%     
==========================================
  Files         106      106              
  Lines       10913    10919       +6     
==========================================
+ Hits         9222     9229       +7     
+ Misses       1295     1294       -1     
  Partials      396      396              
```


| [Files](https://app.codecov.io/gh/bluenviron/gortsplib/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron) | Coverage Δ | |
|---|---|---|
| [server\_stream\_format.go](https://app.codecov.io/gh/bluenviron/gortsplib/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron#diff-c2VydmVyX3N0cmVhbV9mb3JtYXQuZ28=) | `82.85% <66.66%> (+1.60%)` | :arrow_up: |
| [server\_stream.go](https://app.codecov.io/gh/bluenviron/gortsplib/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron#diff-c2VydmVyX3N0cmVhbS5nbw==) | `84.89% <33.33%> (-0.82%)` | :arrow_down: |

... and [1 file with indirect coverage changes](https://app.codecov.io/gh/bluenviron/gortsplib/pull/457/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bluenviron)

:mega: Codecov offers a browser extension for seamless coverage viewing on GitHub. Try it in [Chrome](https://chrome.google.com/webstore/detail/codecov/gedikamndpbemklijjkncpnolildpbgo) or [Firefox](https://addons.mozilla.org/en-US/firefox/addon/codecov/) today!

rse: I've updated the PR to make the linter happy again.
rse: Ok, the go linter is rather picky. I had to adjust the code once again to let the "make lint" no longer complain. Now it should be acceptable.

Patch:
diff --git a/server_stream.go b/server_stream.go
index 107f0baa..cce0b817 100644
--- a/server_stream.go
+++ b/server_stream.go
@@ -2,6 +2,7 @@ package gortsplib
 
 import (
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/pion/rtcp"
@@ -37,6 +38,7 @@ type ServerStream struct {
 	activeUnicastReaders map[*ServerSession]struct{}
 	streamMedias         map[*description.Media]*serverStreamMedia
 	closed               bool
+	bytesSent            *uint64
 }
 
 // NewServerStream allocates a ServerStream.
@@ -46,6 +48,7 @@ func NewServerStream(s *Server, desc *description.Session) *ServerStream {
 		desc:                 desc,
 		readers:              make(map[*ServerSession]struct{}),
 		activeUnicastReaders: make(map[*ServerSession]struct{}),
+		bytesSent:            new(uint64),
 	}
 
 	st.streamMedias = make(map[*description.Media]*serverStreamMedia, len(desc.Medias))
@@ -71,6 +74,11 @@ func (st *ServerStream) Close() {
 	}
 }
 
+// BytesSent returns the number of written bytes.
+func (st *ServerStream) BytesSent() uint64 {
+	return atomic.LoadUint64(st.bytesSent)
+}
+
 // Description returns the description of the stream.
 func (st *ServerStream) Description() *description.Session {
 	return st.desc
diff --git a/server_stream_format.go b/server_stream_format.go
index 6afa385f..09114dba 100644
--- a/server_stream_format.go
+++ b/server_stream_format.go
@@ -1,6 +1,7 @@
 package gortsplib
 
 import (
+	"sync/atomic"
 	"time"
 
 	"github.com/pion/rtcp"
@@ -39,6 +40,8 @@ func newServerStreamFormat(sm *serverStreamMedia, forma format.Format) *serverSt
 func (sf *serverStreamFormat) writePacketRTP(byts []byte, pkt *rtp.Packet, ntp time.Time) error {
 	sf.rtcpSender.ProcessPacket(pkt, ntp, sf.format.PTSEqualsDTS(pkt))
 
+	le := uint64(len(byts))
+
 	// send unicast
 	for r := range sf.sm.st.activeUnicastReaders {
 		sm, ok := r.setuppedMedias[sf.sm.media]
@@ -46,6 +49,8 @@ func (sf *serverStreamFormat) writePacketRTP(byts []byte, pkt *rtp.Packet, ntp t
 			err := sm.writePacketRTP(byts)
 			if err != nil {
 				r.onStreamWriteError(err)
+			} else {
+				atomic.AddUint64(sf.sm.st.bytesSent, le)
 			}
 		}
 	}
@@ -56,6 +61,7 @@ func (sf *serverStreamFormat) writePacketRTP(byts []byte, pkt *rtp.Packet, ntp t
 		if err != nil {
 			return err
 		}
+		atomic.AddUint64(sf.sm.st.bytesSent, le)
 	}
 
 	return nil


Test patch:
diff --git a/server_play_test.go b/server_play_test.go
index a0ee42c1..c03616e2 100644
--- a/server_play_test.go
+++ b/server_play_test.go
@@ -673,7 +673,7 @@ func TestServerPlay(t *testing.T) {
 				v := headers.TransportDeliveryUnicast
 				inTH.Delivery = &v
 				inTH.Protocol = headers.TransportProtocolTCP
-				inTH.InterleavedIDs = &[2]int{5, 6} // off value
+				inTH.InterleavedIDs = &[2]int{5, 6} // odd value
 			}
 
 			res, th := doSetup(t, conn, absoluteControlAttribute(desc.MediaDescriptions[0]), inTH, "")
@@ -2122,3 +2122,76 @@ func TestServerPlayNoInterleavedIDs(t *testing.T) {
 		require.Equal(t, testRTPPacketMarshaled, f.Payload)
 	}
 }
+
+func TestServerPlayBytesSent(t *testing.T) {
+	var stream *ServerStream
+
+	s := &Server{
+		RTSPAddress:       "localhost:8554",
+		MulticastIPRange:  "224.1.0.0/16",
+		MulticastRTPPort:  8000,
+		MulticastRTCPPort: 8001,
+		Handler: &testServerHandler{
+			onDescribe: func(ctx *ServerHandlerOnDescribeCtx) (*base.Response, *ServerStream, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, stream, nil
+			},
+			onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, stream, nil
+			},
+			onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, nil
+			},
+		},
+	}
+
+	err := s.Start()
+	require.NoError(t, err)
+	defer s.Close()
+
+	stream = NewServerStream(s, &description.Session{Medias: []*description.Media{testH264Media}})
+	defer stream.Close()
+
+	for _, transport := range []string{"tcp", "multicast"} {
+		nconn, err := net.Dial("tcp", "localhost:8554")
+		require.NoError(t, err)
+		defer nconn.Close()
+		conn := conn.NewConn(nconn)
+
+		desc := doDescribe(t, conn)
+
+		inTH := &headers.Transport{
+			Mode: func() *headers.TransportMode {
+				v := headers.TransportModePlay
+				return &v
+			}(),
+		}
+
+		if transport == "multicast" {
+			v := headers.TransportDeliveryMulticast
+			inTH.Delivery = &v
+			inTH.Protocol = headers.TransportProtocolUDP
+		} else {
+			v := headers.TransportDeliveryUnicast
+			inTH.Delivery = &v
+			inTH.Protocol = headers.TransportProtocolTCP
+			inTH.InterleavedIDs = &[2]int{0, 1}
+		}
+
+		res, _ := doSetup(t, conn, absoluteControlAttribute(desc.MediaDescriptions[0]), inTH, "")
+
+		session := readSession(t, res)
+
+		doPlay(t, conn, "rtsp://localhost:8554/teststream", session)
+	}
+
+	err = stream.WritePacketRTP(stream.Description().Medias[0], &testRTPPacket)
+	require.NoError(t, err)
+
+	require.Equal(t, uint64(16*2), stream.BytesSent())
+}