config to enable jitter buffer (#312)

Author: frostbyte73

pkg/config/config.go

@@ -88,7 +88,8 @@ type Config struct {
 	HideInboundPort bool `yaml:"hide_inbound_port"`
 
 	// AudioDTMF forces SIP to generate audio DTMF tones in addition to digital.
-	AudioDTMF bool `yaml:"audio_dtmf"`
+	AudioDTMF          bool `yaml:"audio_dtmf"`
+	EnableJitterBuffer bool `yaml:"enable_jitter_buffer"`
 
 	// internal
 	ServiceName string `yaml:"-"`

pkg/media/rtp/jitter.go

@@ -23,14 +23,10 @@ import (
 )
 
 const (
-	jitterEnabled    = false
 	jitterMaxLatency = 60 * time.Millisecond // should match mixer's target buffer size
 )
 
 func HandleJitter(clockRate int, h Handler) Handler {
-	if !jitterEnabled {
-		return h
-	}
 	return &jitterHandler{
 		h:   h,
 		buf: jitter.NewBuffer(audioDepacketizer{}, uint32(clockRate), jitterMaxLatency),

pkg/sip/inbound.go

@@ -17,14 +17,15 @@ package sip
 import (
 	"context"
 	"fmt"
-	"github.com/livekit/protocol/rpc"
 	"math"
 	"net/netip"
 	"slices"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/livekit/protocol/rpc"
+
 	"github.com/frostbyte73/core"
 	"github.com/icholy/digest"
 	"github.com/pkg/errors"
@@ -544,6 +545,7 @@ func (c *inboundCall) runMediaConn(offerData []byte, conf *config.Config, featur
 		Ports:               conf.RTPPort,
 		MediaTimeoutInitial: c.s.conf.MediaTimeoutInitial,
 		MediaTimeout:        c.s.conf.MediaTimeout,
+		EnableJitterBuffer:  c.s.conf.EnableJitterBuffer,
 	}, RoomSampleRate)
 	if err != nil {
 		return nil, err

pkg/sip/media_port.go

@@ -43,6 +43,7 @@ type MediaConfig struct {
 	Ports               rtcconfig.PortRange
 	MediaTimeoutInitial time.Duration
 	MediaTimeout        time.Duration
+	EnableJitterBuffer  bool
 }
 
 func NewMediaPort(log logger.Logger, mon *stats.CallMonitor, conf *MediaConfig, sampleRate int) (*MediaPort, error) {
@@ -52,10 +53,11 @@ func NewMediaPort(log logger.Logger, mon *stats.CallMonitor, conf *MediaConfig,
 func NewMediaPortWith(log logger.Logger, mon *stats.CallMonitor, conn rtp.UDPConn, conf *MediaConfig, sampleRate int) (*MediaPort, error) {
 	mediaTimeout := make(chan struct{})
 	p := &MediaPort{
-		log:          log,
-		mon:          mon,
-		externalIP:   conf.IP,
-		mediaTimeout: mediaTimeout,
+		log:           log,
+		mon:           mon,
+		externalIP:    conf.IP,
+		mediaTimeout:  mediaTimeout,
+		jitterEnabled: conf.EnableJitterBuffer,
 		conn: rtp.NewConnWith(conn, &rtp.ConnConfig{
 			MediaTimeoutInitial: conf.MediaTimeoutInitial,
 			MediaTimeout:        conf.MediaTimeout,
@@ -81,6 +83,7 @@ type MediaPort struct {
 	conn             *rtp.Conn
 	mediaTimeout     <-chan struct{}
 	dtmfAudioEnabled bool
+	jitterEnabled    bool
 	closed           atomic.Bool
 
 	mu           sync.Mutex
@@ -236,7 +239,9 @@ func (p *MediaPort) setupInput() {
 	// Decoding pipeline (SIP -> LK)
 	audioHandler := p.conf.Audio.Codec.DecodeRTP(p.audioIn, p.conf.Audio.Type)
 	p.audioInHandler = audioHandler
-	audioHandler = rtp.HandleJitter(p.conf.Audio.Codec.Info().RTPClockRate, audioHandler)
+	if p.jitterEnabled {
+		audioHandler = rtp.HandleJitter(p.conf.Audio.Codec.Info().RTPClockRate, audioHandler)
+	}
 	mux := rtp.NewMux(nil)
 	mux.SetDefault(newRTPStatsHandler(p.mon, "", nil))
 	mux.Register(p.conf.Audio.Type, newRTPStatsHandler(p.mon, p.conf.Audio.Codec.Info().SDPName, audioHandler))

pkg/sip/outbound.go

@@ -122,6 +122,7 @@ func (c *Client) newCall(ctx context.Context, conf *config.Config, log logger.Lo
 		Ports:               conf.RTPPort,
 		MediaTimeoutInitial: c.conf.MediaTimeoutInitial,
 		MediaTimeout:        c.conf.MediaTimeout,
+		EnableJitterBuffer:  c.conf.EnableJitterBuffer,
 	}, RoomSampleRate)
 	if err != nil {
 		call.close(errors.Wrap(err, "media failed"), callDropped, "media-failed", livekit.DisconnectReason_UNKNOWN_REASON)

pkg/sip/room.go

@@ -207,7 +207,9 @@ func (r *Room) Connect(conf *config.Config, rconf RoomConfig) error {
 					defer odec.Close()
 
 					var h rtp.Handler = rtp.NewMediaStreamIn[opus.Sample](odec)
-					h = rtp.HandleJitter(int(track.Codec().ClockRate), h)
+					if conf.EnableJitterBuffer {
+						h = rtp.HandleJitter(int(track.Codec().ClockRate), h)
+					}
 					err = rtp.HandleLoop(track, h)
 					if err != nil && !errors.Is(err, io.EOF) {
 						log.Infow("room track rtp handler returned with failure", "error", err)

test/integration/sip_test.go

@@ -59,18 +59,19 @@ func runSIPServer(t testing.TB, lk *LiveKit) *SIPServer {
 	local, err := config.GetLocalIP()
 	require.NoError(t, err)
 	conf := &config.Config{
-		NodeID:            utils.NewGuid("NS_"),
-		ApiKey:            lk.ApiKey,
-		ApiSecret:         lk.ApiSecret,
-		WsUrl:             lk.WsUrl,
-		Redis:             lk.Redis,
-		SIPPort:           sipPort,
-		SIPPortListen:     sipPort,
-		ListenIP:          local.String(),
-		RTPPort:           rtcconfig.PortRange{Start: 20000, End: 20010},
-		UseExternalIP:     false,
-		MaxCpuUtilization: 0.9,
-		Logging:           logger.Config{Level: "debug"},
+		NodeID:             utils.NewGuid("NS_"),
+		ApiKey:             lk.ApiKey,
+		ApiSecret:          lk.ApiSecret,
+		WsUrl:              lk.WsUrl,
+		Redis:              lk.Redis,
+		SIPPort:            sipPort,
+		SIPPortListen:      sipPort,
+		ListenIP:           local.String(),
+		RTPPort:            rtcconfig.PortRange{Start: 20000, End: 20010},
+		UseExternalIP:      false,
+		MaxCpuUtilization:  0.9,
+		Logging:            logger.Config{Level: "debug"},
+		EnableJitterBuffer: true,
 	}
 	_ = conf.InitLogger()
 	log := logger.GetLogger()

test/lktest/sip.go

@@ -340,5 +340,8 @@ Check logs for call:
 	if !params.NoDMTF {
 		t.Log("testing dtmf")
 		CheckDTMFForParticipants(t, ctx, pOut, pIn, dataOut, dataIn)
+
+		t.Log("retesting audio")
+		CheckAudioForParticipants(t, ctx, pOut, pIn)
 	}
 }