feat(gateway): relay complete encoded frames

gateway/apollo_native.go

@@ -28,7 +28,9 @@ import (
 )
 
 const (
-	nativeApolloVideoQueuePackets = 256
+	nativeApolloVideoQueuePackets = 16
+	nativeApolloVideoQueueBytes   = 4 << 20
+	nativeApolloVideoQueueLatency = 250 * time.Millisecond
 	nativeApolloAudioQueuePackets = 16
 	nativeApolloEventQueuePackets = 16
 )
@@ -316,6 +318,9 @@ type nativeApolloSession struct {
 	mediaRecovered              atomic.Uint64
 	mediaEnqueued               atomic.Uint64
 	mediaQueueMaximum           atomic.Uint64
+	mediaQueueBytes             atomic.Int64
+	mediaQueueMaximumBytes      atomic.Uint64
+	mediaQueueSequence          atomic.Uint64
 }
 
 func newNativeApolloSession(sessionID string) *nativeApolloSession {
@@ -762,10 +767,22 @@ func (s *nativeApolloSession) quiesceMedia() {
 
 func (s *nativeApolloSession) closeMediaChannels() {
 	s.channelsOnce.Do(func() {
+		s.mediaQuiesced.Store(true)
 		s.mediaMu.Lock()
 		defer s.mediaMu.Unlock()
-		close(s.video)
-		close(s.audio)
+		for {
+			select {
+			case media := <-s.video:
+				if media.expiry != nil {
+					media.expiry.Stop()
+				}
+				media.releaseQueue()
+			default:
+				close(s.video)
+				close(s.audio)
+				return
+			}
+		}
 	})
 }
 
@@ -775,9 +792,66 @@ func (s *nativeApolloSession) enqueueMedia(output chan ProviderMedia, payload []
 	if len(payload) == 0 || s.mediaQuiesced.Load() {
 		return false
 	}
+	if output == s.video && len(payload) > maxCompleteFrameBytes {
+		s.mediaDrops.Add(1)
+		return false
+	}
 	s.mediaRecovered.Add(1)
 	media := ProviderMedia{Payload: payload, ReceivedAt: receivedAt, EnqueuedAt: time.Now()}
-	if pushLatest(output, media) {
+	if output == s.video {
+		dropped := uint64(0)
+		for s.mediaQueueBytes.Load()+int64(len(payload)) > nativeApolloVideoQueueBytes {
+			select {
+			case replaced := <-output:
+				if replaced.expiry != nil {
+					replaced.expiry.Stop()
+				}
+				replaced.releaseQueue()
+				dropped++
+			default:
+				s.mediaDrops.Add(dropped + 1)
+				return false
+			}
+		}
+		media.queueID = s.mediaQueueSequence.Add(1)
+		media.accounting = &providerMediaQueueAccounting{
+			bytes: int64(len(payload)), total: &s.mediaQueueBytes,
+		}
+		currentBytes := uint64(s.mediaQueueBytes.Add(int64(len(payload))))
+		media.expiry = time.AfterFunc(nativeApolloVideoQueueLatency, func() {
+			s.expireVideo(media.queueID)
+			media.releaseQueue()
+		})
+		for maximum := s.mediaQueueMaximumBytes.Load(); currentBytes > maximum && !s.mediaQueueMaximumBytes.CompareAndSwap(maximum, currentBytes); maximum = s.mediaQueueMaximumBytes.Load() {
+		}
+		if dropped > 0 {
+			s.mediaDrops.Add(dropped)
+		}
+	}
+	dropped := false
+	select {
+	case output <- media:
+	default:
+		select {
+		case replaced := <-output:
+			if replaced.expiry != nil {
+				replaced.expiry.Stop()
+			}
+			replaced.releaseQueue()
+		default:
+		}
+		select {
+		case output <- media:
+			dropped = true
+		default:
+			if media.expiry != nil {
+				media.expiry.Stop()
+			}
+			media.releaseQueue()
+			dropped = true
+		}
+	}
+	if dropped {
 		s.mediaDrops.Add(1)
 	}
 	s.mediaEnqueued.Add(1)
@@ -787,6 +861,35 @@ func (s *nativeApolloSession) enqueueMedia(output chan ProviderMedia, payload []
 	return true
 }
 
+func (s *nativeApolloSession) expireVideo(queueID uint64) {
+	s.mediaMu.Lock()
+	defer s.mediaMu.Unlock()
+	if s.mediaQuiesced.Load() {
+		return
+	}
+	retained := make([]ProviderMedia, 0, cap(s.video))
+	removed := false
+	for {
+		select {
+		case media := <-s.video:
+			if media.queueID == queueID {
+				removed = true
+				media.releaseQueue()
+				continue
+			}
+			retained = append(retained, media)
+		default:
+			for _, media := range retained {
+				s.video <- media
+			}
+			if removed {
+				s.mediaDrops.Add(1)
+			}
+			return
+		}
+	}
+}
+
 func (s *nativeApolloSession) readUDPMedia() {
 	if s.media == nil {
 		close(s.readDone)