Make refactored BWE compatible with the old API

go.mod

@@ -1,6 +1,6 @@
 module github.com/pion/interceptor
 
-go 1.20
+go 1.21
 
 require (
 	github.com/pion/logging v0.2.3

internal/test/mock_stream.go

@@ -41,6 +41,7 @@ type RTPWithError struct {
 // RTCPWithError is used to send a batch of rtcp packets or an error on a channel
 type RTCPWithError struct {
 	Packets []rtcp.Packet
+	Attr    interceptor.Attributes
 	Err     error
 }
 
@@ -107,21 +108,21 @@ func NewMockStream(info *interceptor.StreamInfo, i interceptor.Interceptor) *Moc
 	go func() {
 		buf := make([]byte, 1500)
 		for {
-			i, _, err := s.rtcpReader.Read(buf, interceptor.Attributes{})
+			i, attr, err := s.rtcpReader.Read(buf, interceptor.Attributes{})
 			if err != nil {
 				if !errors.Is(err, io.EOF) {
-					s.rtcpInModified <- RTCPWithError{Err: err}
+					s.rtcpInModified <- RTCPWithError{Attr: attr, Err: err}
 				}
 				return
 			}
 
 			pkts, err := rtcp.Unmarshal(buf[:i])
 			if err != nil {
-				s.rtcpInModified <- RTCPWithError{Err: err}
+				s.rtcpInModified <- RTCPWithError{Attr: attr, Err: err}
 				return
 			}
 
-			s.rtcpInModified <- RTCPWithError{Packets: pkts}
+			s.rtcpInModified <- RTCPWithError{Attr: attr, Packets: pkts}
 		}
 	}()
 	go func() {

pkg/bwe/acknowledgment.go

@@ -0,0 +1,38 @@
+package bwe
+
+import (
+	"fmt"
+	"time"
+)
+
+type ECN uint8
+
+const (
+	//nolint:misspell
+	// ECNNonECT signals Non ECN-Capable Transport, Non-ECT
+	ECNNonECT ECN = iota // 00
+
+	//nolint:misspell
+	// ECNECT1 signals ECN Capable Transport, ECT(0)
+	ECNECT1 // 01
+
+	//nolint:misspell
+	// ECNECT0 signals ECN Capable Transport, ECT(1)
+	ECNECT0 // 10
+
+	// ECNCE signals ECN Congestion Encountered, CE
+	ECNCE // 11
+)
+
+type Acknowledgment struct {
+	SeqNr     int64
+	Size      uint16
+	Departure time.Time
+	Arrived   bool
+	Arrival   time.Time
+	ECN       ECN
+}
+
+func (a Acknowledgment) String() string {
+	return fmt.Sprintf("seq=%v, departure=%v, arrival=%v", a.SeqNr, a.Departure, a.Arrival)
+}

pkg/bwe/arrival_group_accumulator.go

@@ -0,0 +1,48 @@
+package bwe
+
+import (
+	"time"
+)
+
+type arrivalGroup []Acknowledgment
+
+type arrivalGroupAccumulator struct {
+	next             arrivalGroup
+	burstInterval    time.Duration
+	maxBurstDuration time.Duration
+}
+
+func newArrivalGroupAccumulator() *arrivalGroupAccumulator {
+	return &arrivalGroupAccumulator{
+		next:             make([]Acknowledgment, 0),
+		burstInterval:    5 * time.Millisecond,
+		maxBurstDuration: 100 * time.Millisecond,
+	}
+}
+
+func (a *arrivalGroupAccumulator) onPacketAcked(ack Acknowledgment) arrivalGroup {
+	if len(a.next) == 0 {
+		a.next = append(a.next, ack)
+		return nil
+	}
+
+	if ack.Departure.Sub(a.next[0].Departure) < a.burstInterval {
+		a.next = append(a.next, ack)
+		return nil
+	}
+
+	sendTimeDelta := ack.Departure.Sub(a.next[0].Departure)
+	arrivalTimeDeltaLast := ack.Arrival.Sub(a.next[len(a.next)-1].Arrival)
+	arrivalTimeDeltaFirst := ack.Arrival.Sub(a.next[0].Arrival)
+	propagationDelta := arrivalTimeDeltaFirst - sendTimeDelta
+
+	if propagationDelta < 0 && arrivalTimeDeltaLast <= a.burstInterval && arrivalTimeDeltaFirst < a.maxBurstDuration {
+		a.next = append(a.next, ack)
+		return nil
+	}
+
+	group := make(arrivalGroup, len(a.next))
+	copy(group, a.next)
+	a.next = arrivalGroup{ack}
+	return group
+}

pkg/bwe/arrival_group_accumulator_test.go

@@ -0,0 +1,239 @@
+package bwe
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestArrivalGroupAccumulator(t *testing.T) {
+	triggerNewGroupElement := Acknowledgment{
+		Departure: time.Time{}.Add(time.Second),
+		Arrival:   time.Time{}.Add(time.Second),
+	}
+	cases := []struct {
+		name string
+		log  []Acknowledgment
+		exp  []arrivalGroup
+	}{
+		{
+			name: "emptyCreatesNoGroups",
+			log:  []Acknowledgment{},
+			exp:  []arrivalGroup{},
+		},
+		{
+			name: "createsSingleElementGroup",
+			log: []Acknowledgment{
+				{
+					Departure: time.Time{},
+					Arrival:   time.Time{}.Add(time.Millisecond),
+				},
+				triggerNewGroupElement,
+			},
+			exp: []arrivalGroup{{
+				{
+					Departure: time.Time{},
+					Arrival:   time.Time{}.Add(time.Millisecond),
+				},
+			},
+			},
+		},
+		{
+			name: "createsTwoElementGroup",
+			log: []Acknowledgment{
+				{
+					Arrival: time.Time{}.Add(15 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(3 * time.Millisecond),
+					Arrival:   time.Time{}.Add(20 * time.Millisecond),
+				},
+				triggerNewGroupElement,
+			},
+			exp: []arrivalGroup{{
+				{
+					Departure: time.Time{},
+					Arrival:   time.Time{}.Add(15 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(3 * time.Millisecond),
+					Arrival:   time.Time{}.Add(20 * time.Millisecond),
+				},
+			}},
+		},
+		{
+			name: "createsTwoArrivalGroups1",
+			log: []Acknowledgment{
+				{
+					Departure: time.Time{},
+					Arrival:   time.Time{}.Add(15 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(3 * time.Millisecond),
+					Arrival:   time.Time{}.Add(20 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(9 * time.Millisecond),
+					Arrival:   time.Time{}.Add(24 * time.Millisecond),
+				},
+				triggerNewGroupElement,
+			},
+			exp: []arrivalGroup{
+				{
+					{
+						Arrival: time.Time{}.Add(15 * time.Millisecond),
+					},
+					{
+						Departure: time.Time{}.Add(3 * time.Millisecond),
+						Arrival:   time.Time{}.Add(20 * time.Millisecond),
+					},
+				},
+				{
+					{
+						Departure: time.Time{}.Add(9 * time.Millisecond),
+						Arrival:   time.Time{}.Add(24 * time.Millisecond),
+					},
+				},
+			},
+		},
+		{
+			name: "createsTwoArrivalGroups2",
+			log: []Acknowledgment{
+				{
+					Departure: time.Time{},
+					Arrival:   time.Time{}.Add(15 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(3 * time.Millisecond),
+					Arrival:   time.Time{}.Add(20 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(9 * time.Millisecond),
+					Arrival:   time.Time{}.Add(30 * time.Millisecond),
+				},
+				triggerNewGroupElement,
+			},
+			exp: []arrivalGroup{
+				{
+					{
+						Arrival: time.Time{}.Add(15 * time.Millisecond),
+					},
+					{
+						Departure: time.Time{}.Add(3 * time.Millisecond),
+						Arrival:   time.Time{}.Add(20 * time.Millisecond),
+					},
+				},
+				{
+					{
+						Departure: time.Time{}.Add(9 * time.Millisecond),
+						Arrival:   time.Time{}.Add(30 * time.Millisecond),
+					},
+				},
+			},
+		},
+		{
+			name: "ignoresOutOfOrderPackets",
+			log: []Acknowledgment{
+				{
+					Departure: time.Time{},
+					Arrival:   time.Time{}.Add(15 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(6 * time.Millisecond),
+					Arrival:   time.Time{}.Add(34 * time.Millisecond),
+				},
+				{
+					Departure: time.Time{}.Add(8 * time.Millisecond),
+					Arrival:   time.Time{}.Add(30 * time.Millisecond),
+				},
+				triggerNewGroupElement,
+			},
+			exp: []arrivalGroup{
+				{
+					{
+						Departure: time.Time{},
+						Arrival:   time.Time{}.Add(15 * time.Millisecond),
+					},
+				},
+				{
+					{
+						Departure: time.Time{}.Add(6 * time.Millisecond),
+						Arrival:   time.Time{}.Add(34 * time.Millisecond),
+					},
+					{
+						Departure: time.Time{}.Add(8 * time.Millisecond),
+						Arrival:   time.Time{}.Add(30 * time.Millisecond),
+					},
+				},
+			},
+		},
+		{
+			name: "newGroupBecauseOfInterDepartureTime",
+			log: []Acknowledgment{
+				{
+					SeqNr:     0,
+					Departure: time.Time{},
+					Arrival:   time.Time{}.Add(4 * time.Millisecond),
+				},
+				{
+					SeqNr:     1,
+					Departure: time.Time{}.Add(3 * time.Millisecond),
+					Arrival:   time.Time{}.Add(4 * time.Millisecond),
+				},
+				{
+					SeqNr:     2,
+					Departure: time.Time{}.Add(6 * time.Millisecond),
+					Arrival:   time.Time{}.Add(10 * time.Millisecond),
+				},
+				{
+					SeqNr:     3,
+					Departure: time.Time{}.Add(9 * time.Millisecond),
+					Arrival:   time.Time{}.Add(10 * time.Millisecond),
+				},
+				triggerNewGroupElement,
+			},
+			exp: []arrivalGroup{
+				{
+					{
+						SeqNr:     0,
+						Departure: time.Time{},
+						Arrival:   time.Time{}.Add(4 * time.Millisecond),
+					},
+					{
+						SeqNr:     1,
+						Departure: time.Time{}.Add(3 * time.Millisecond),
+						Arrival:   time.Time{}.Add(4 * time.Millisecond),
+					},
+				},
+				{
+					{
+						SeqNr:     2,
+						Departure: time.Time{}.Add(6 * time.Millisecond),
+						Arrival:   time.Time{}.Add(10 * time.Millisecond),
+					},
+					{
+						SeqNr:     3,
+						Departure: time.Time{}.Add(9 * time.Millisecond),
+						Arrival:   time.Time{}.Add(10 * time.Millisecond),
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			aga := newArrivalGroupAccumulator()
+			received := []arrivalGroup{}
+			for _, ack := range tc.log {
+				next := aga.onPacketAcked(ack)
+				if next != nil {
+					received = append(received, next)
+				}
+			}
+			assert.Equal(t, tc.exp, received)
+		})
+	}
+}