Add bitrate to webrtc/mse/mp4 consumer info (#5)

.gitignore

@@ -5,3 +5,5 @@ go2rtc.yaml
 go2rtc.json
 
 0_test.go
+
+.goreload

Makefile

@@ -0,0 +1,2 @@
+dev:
+	goreload .

pkg/core/core.go

@@ -100,6 +100,7 @@ type Info struct {
 	Senders    []*Sender   `json:"senders,omitempty"`
 	Recv       int         `json:"recv,omitempty"`
 	Send       int         `json:"send,omitempty"`
+	Bitrate    int         `json:"bitrate,omitempty"` // bytes per second
 }
 
 const (
@@ -147,6 +148,7 @@ type SuperConsumer struct {
 	Medias     []*Media  `json:"medias,omitempty"`
 	Senders    []*Sender `json:"senders,omitempty"`
 	Send       int       `json:"send,omitempty"`
+	Bitrate    int       `json:"bitrate,omitempty"` // bytes per second
 }
 
 func (s *SuperConsumer) GetMedias() []*Media {

pkg/custom/custom.go

@@ -0,0 +1,21 @@
+package custom
+
+import (
+	"time"
+)
+
+func StartBitrateWorker(send, bitrate *int, stop chan struct{}) {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	prevSendBytes := *send
+	for {
+		select {
+		case <-ticker.C:
+			*bitrate = *send - prevSendBytes
+			prevSendBytes = *send
+		case <-stop:
+			return
+		}
+	}
+}

pkg/isapi/client.go

@@ -2,12 +2,13 @@ package isapi
 
 import (
 	"errors"
-	"github.com/AlexxIT/go2rtc/pkg/core"
-	"github.com/AlexxIT/go2rtc/pkg/tcp"
 	"io"
 	"net"
 	"net/http"
 	"net/url"
+
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/tcp"
 )
 
 type Client struct {
@@ -20,6 +21,9 @@ type Client struct {
 	medias []*core.Media
 	sender *core.Sender
 	send   int
+
+	bitrate           int // bytes per second
+	stopBitrateWorker chan struct{}
 }
 
 func NewClient(rawURL string) (*Client, error) {
@@ -32,7 +36,9 @@ func NewClient(rawURL string) (*Client, error) {
 	u.Scheme = "http"
 	u.Path = ""
 
-	return &Client{url: u.String()}, nil
+	c := &Client{url: u.String()}
+	c.startBitrateWorker()
+	return c, nil
 }
 
 func (c *Client) Dial() (err error) {
@@ -137,6 +143,8 @@ func (c *Client) Close() (err error) {
 
 	tcp.Close(res)
 
+	c.stopBitrateWorker <- struct{}{}
+
 	return nil
 }
 

pkg/isapi/consumer.go

@@ -2,6 +2,7 @@ package isapi
 
 import (
 	"encoding/json"
+
 	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/pion/rtp"
 )
@@ -52,9 +53,10 @@ func (c *Client) Stop() (err error) {
 
 func (c *Client) MarshalJSON() ([]byte, error) {
 	info := &core.Info{
-		Type:   "ISAPI active consumer",
-		Medias: c.medias,
-		Send:   c.send,
+		Type:    "ISAPI active consumer",
+		Medias:  c.medias,
+		Send:    c.send,
+		Bitrate: c.bitrate,
 	}
 	if c.sender != nil {
 		info.Senders = []*core.Sender{c.sender}

pkg/isapi/custom.go

@@ -0,0 +1,8 @@
+package isapi
+
+import "github.com/AlexxIT/go2rtc/pkg/custom"
+
+func (c *Client) startBitrateWorker() {
+	c.stopBitrateWorker = make(chan struct{})
+	go custom.StartBitrateWorker(&c.send, &c.bitrate, c.stopBitrateWorker)
+}

pkg/mp4/consumer.go

@@ -23,6 +23,8 @@ type Consumer struct {
 	Rotate int `json:"-"`
 	ScaleX int `json:"-"`
 	ScaleY int `json:"-"`
+
+	stopBitrateWorker chan struct{}
 }
 
 func NewConsumer(medias []*core.Media) *Consumer {
@@ -52,6 +54,7 @@ func NewConsumer(medias []*core.Media) *Consumer {
 		wr:    core.NewWriteBuffer(nil),
 	}
 	cons.Medias = medias
+	cons.startBitrateWorker()
 	return cons
 }
 
@@ -185,5 +188,6 @@ func (c *Consumer) WriteTo(wr io.Writer) (int64, error) {
 
 func (c *Consumer) Stop() error {
 	_ = c.SuperConsumer.Close()
+	c.stopBitrateWorker <- struct{}{}
 	return c.wr.Close()
 }

pkg/mp4/custom.go

@@ -0,0 +1,8 @@
+package mp4
+
+import "github.com/AlexxIT/go2rtc/pkg/custom"
+
+func (c *Consumer) startBitrateWorker() {
+	c.stopBitrateWorker = make(chan struct{})
+	go custom.StartBitrateWorker(&c.SuperConsumer.Send, &c.SuperConsumer.Bitrate, c.stopBitrateWorker)
+}

pkg/webrtc/conn.go

@@ -28,6 +28,9 @@ type Conn struct {
 	offer  string
 	remote string
 	closed core.Waiter
+
+	bitrate           int // bytes per second
+	stopBitrateWorker chan struct{}
 }
 
 func NewConn(pc *webrtc.PeerConnection) *Conn {
@@ -127,11 +130,14 @@ func NewConn(pc *webrtc.PeerConnection) *Conn {
 		}
 	})
 
+	c.startBitrateWorker()
+
 	return c
 }
 
 func (c *Conn) Close() error {
 	c.closed.Done(nil)
+	c.stopBitrateWorker <- struct{}{}
 	return c.pc.Close()
 }
 

pkg/webrtc/consumer.go

@@ -93,6 +93,7 @@ func (c *Conn) MarshalJSON() ([]byte, error) {
 		Senders:    c.senders,
 		Recv:       c.recv,
 		Send:       c.send,
+		Bitrate:    c.bitrate,
 	}
 	return json.Marshal(info)
 }

pkg/webrtc/custom.go

@@ -0,0 +1,10 @@
+package webrtc
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/custom"
+)
+
+func (c *Conn) startBitrateWorker() {
+	c.stopBitrateWorker = make(chan struct{})
+	go custom.StartBitrateWorker(&c.send, &c.bitrate, c.stopBitrateWorker)
+}