diff --git a/pkg/config/config.go b/pkg/config/config.go index 461f934c..6bbd2869 100644 --- a/pkg/config/config.go +++ b/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:"-"` diff --git a/pkg/media/rtp/jitter.go b/pkg/media/rtp/jitter.go index 9a1f3874..886ac208 100644 --- a/pkg/media/rtp/jitter.go +++ b/pkg/media/rtp/jitter.go @@ -17,19 +17,16 @@ package rtp import ( "time" - "github.com/livekit/server-sdk-go/v2/pkg/jitter" "github.com/pion/rtp" + + "github.com/livekit/server-sdk-go/v2/pkg/jitter" ) 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), diff --git a/pkg/sip/inbound.go b/pkg/sip/inbound.go index 332a5a0e..544260df 100644 --- a/pkg/sip/inbound.go +++ b/pkg/sip/inbound.go @@ -17,7 +17,6 @@ package sip import ( "context" "fmt" - "github.com/livekit/protocol/rpc" "math" "net/netip" "slices" @@ -25,6 +24,8 @@ import ( "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 diff --git a/pkg/sip/media_port.go b/pkg/sip/media_port.go index 2e7f7c76..90c50020 100644 --- a/pkg/sip/media_port.go +++ b/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)) diff --git a/pkg/sip/outbound.go b/pkg/sip/outbound.go index e93a54b0..b20332e7 100644 --- a/pkg/sip/outbound.go +++ b/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) diff --git a/pkg/sip/room.go b/pkg/sip/room.go index 074be5f0..3f17f820 100644 --- a/pkg/sip/room.go +++ b/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) diff --git a/test/integration/sip_test.go b/test/integration/sip_test.go index cbab2002..a79ed47a 100644 --- a/test/integration/sip_test.go +++ b/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() diff --git a/test/lktest/sip.go b/test/lktest/sip.go index f37f9c38..b5c72ab6 100644 --- a/test/lktest/sip.go +++ b/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) } }