Skip to content

Clean up writer interfaces. #793

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexfish8 merged 9 commits into from
Aug 14, 2026
Merged

Clean up writer interfaces. #793

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d0ea088
Remove audio input processor.
alexfish8 Aug 14, 2026
8d011d6
Rename MediaSegment methods.
alexfish8 Aug 13, 2026
a28910f
Implement MediaSegment for *Room.
alexfish8 Aug 13, 2026
e9f9a1b
Fix outbound audio issue when transferring calls.
alexfish8 Aug 14, 2026
25ebe74
Respond to PR feedback.
alexfish8 Aug 14, 2026
f55da47
Get tests compiling.
alexfish8 Aug 14, 2026
e874973
Add back in DTMF forwarding check.
alexfish8 Aug 14, 2026
4c0d781
Move DTMF wiring.
alexfish8 Aug 14, 2026
ad3eef4
Fix test.
alexfish8 Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 46 additions & 44 deletions pkg/sip/inbound.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,6 @@ import (
lksip "github.com/livekit/protocol/sip"
"github.com/livekit/protocol/utils/traceid"
"github.com/livekit/psrpc"
lksdk "github.com/livekit/server-sdk-go/v2"
"github.com/livekit/sipgo/sip"

"github.com/livekit/sip/pkg/config"
Expand Down Expand Up @@ -718,7 +717,6 @@ type inboundCall struct {
lkRoom RoomInterface // LiveKit room; only active after correct pin is entered
callDur func() time.Duration
joinDur func() time.Duration
forwardDTMF atomic.Bool
done atomic.Bool
started core.Fuse
stats Stats
Expand Down Expand Up @@ -978,7 +976,7 @@ func (c *inboundCall) handleInvite(ctx context.Context, tid traceid.ID, req *sip
// Start this timer right after the Accept.
ackTimeout = time.After(inviteOkAckLateTimeout)
}
if old := c.audioOut.Swap(c.media.GetAudioWriter()); old != nil {
if old := c.audioOut.Swap(c.media.GetOutboundAudioWriter()); old != nil {
c.log().Warnw("unexpected audio out writer", nil)
old.Close()
}
Expand Down Expand Up @@ -1015,6 +1013,7 @@ func (c *inboundCall) handleInvite(ctx context.Context, tid traceid.ID, req *sip
return err // already sent a response
}
}

p := &disp.Room.Participant
p.Attributes = HeadersToAttrs(p.Attributes, disp.HeadersToAttributes, disp.IncludeHeaders, c.cc, nil)
if disp.MaxCallDuration <= 0 || disp.MaxCallDuration > maxCallDuration {
Expand Down Expand Up @@ -1113,25 +1112,33 @@ func (c *inboundCall) waitForCallEnd(ctx context.Context, ackReceived <-chan str
}
}

// TODO(alexfish): Update Room so that we don't need this adapater.
type dtmfEventWriter struct {
handler func(msg *livekit.SipDTMF)
type pinDTMFWriter struct {
dtmfEvents chan<- dtmf.Event
}

func (w *dtmfEventWriter) String() string {
return "dtmfEventWriter"
func (w *pinDTMFWriter) String() string {
return "pinDTMFWriter"
}

func (w *dtmfEventWriter) SampleRate() int {
func (w *pinDTMFWriter) SampleRate() int {
return dtmf.SampleRate
}

func (w *dtmfEventWriter) Close() error {
func (w *pinDTMFWriter) Close() error {
return nil
}

func (w *dtmfEventWriter) WriteSample(sample *livekit.SipDTMF) error {
w.handler(sample)
func (w *pinDTMFWriter) WriteSample(msg *livekit.SipDTMF) error {
if msg == nil {
return nil
}

event := dtmfEventFromSipDTMF(msg)
// We should have enough buffer here.
select {
case w.dtmfEvents <- event:
default:
}
return nil
}

Expand Down Expand Up @@ -1173,13 +1180,20 @@ func (c *inboundCall) runMediaConn(tid traceid.ID, offerData []byte, mconf *sipM
c.mon.SDPSize(len(answerData), false)
c.log().Debugw("SDP answer", "sdp", string(answerData))

mp.WriteDTMFTo(&dtmfEventWriter{handler: c.handleDTMF})
if old := mp.WriteInboundDTMFTo(&pinDTMFWriter{c.dtmf}); old != nil {
c.log().Warnw("media port has unexpected inbound DTMF writer", nil)
}

// Must be set earlier to send the pin prompts.
if w := c.lkRoom.SwapOutput(c.audioOut); w != nil {
_ = w.Close()
if old := c.lkRoom.WriteOutboundAudioTo(c.audioOut); old != nil {
c.log().Warnw("room has unexpected outbound audio writer", nil)
old.Close()
}

if old := c.lkRoom.WriteOutboundDTMFTo(c.media.GetOutboundDTMFWriter()); old != nil {
c.log().Warnw("room has unexpected outbound audio DTMF writer", nil)
old.Close()
}
c.lkRoom.SetDTMFOutput(c.media.GetDTMFWriter())

audio := mp.NegotiatedAudio()
if audio == nil {
Expand Down Expand Up @@ -1576,7 +1590,6 @@ func (c *inboundCall) createLiveKitParticipant(ctx context.Context, rconf RoomCo
partConf.Attributes[k] = v
}
partConf.Attributes[livekit.AttrSIPCallStatus] = status.Attribute()
c.forwardDTMF.Store(true)
select {
case <-ctx.Done():
return ctx.Err()
Expand Down Expand Up @@ -1607,16 +1620,23 @@ func (c *inboundCall) createLiveKitParticipant(ctx context.Context, rconf RoomCo

func (c *inboundCall) publishTrack(features []livekit.SIPFeature, featureFlags map[string]string) error {
defer c.mon.StageDurTimer("track-publish")()
local, err := c.lkRoom.NewParticipantTrack(RoomSampleRate)
inboundAudio, err := c.lkRoom.GetInboundAudioWriter()
if err != nil {
_ = c.lkRoom.Close()
return err
}

if audioInProcessor := c.s.handler.GetMediaProcessor(features, featureFlags, string(c.cc.ID()), MediaProcessorOpts{InputSampleRate: RoomSampleRate}); audioInProcessor != nil {
local = audioInProcessor(local)
inboundAudio = audioInProcessor(inboundAudio)
}
if old := c.media.WriteInboundAudioTo(inboundAudio); old != nil {
c.log().Warnw("media port has unexpected inbound audio writer", nil)
old.Close()
}
if old := c.media.WriteInboundDTMFTo(c.lkRoom.GetInboundDTMFWriter()); old != nil {
c.log().Warnw("media port has unexpected inbound dtmf writer", nil)
old.Close()
Comment on lines +1635 to +1638

@devin-ai-integration devin-ai-integration Bot Aug 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Every inbound call logs a false warning about an unexpected DTMF destination

The DTMF destination for the call is unconditionally reported as unexpected (Warnw("media port has unexpected inbound dtmf writer") at pkg/sip/inbound.go:1636) even though the code itself always installs a temporary destination beforehand, so operators see a spurious warning on every single inbound call.
Impact: Log noise and false alarms on every inbound call, which can mask genuine warnings.

Mechanism: pin-prompt DTMF writer is always installed first, so the swap always returns a non-nil previous writer

pkg/sip/inbound.go:1183 always installs &pinDTMFWriter{c.dtmf} as the media port's inbound DTMF destination during runMediaConn, for both the pin-prompt and non-pin dispatch paths. publishTrack later hands DTMF over to the room (pkg/sip/inbound.go:1635), so WriteInboundDTMFTo always returns the previously-installed pinDTMFWriter, making old != nil unconditionally true and triggering the warning plus a no-op Close(). The intended handoff should not be treated as an invariant violation; only a non-pinDTMFWriter previous value is unexpected.

Prompt for agents 
In pkg/sip/inbound.go, runMediaConn always installs a pinDTMFWriter as the media port's inbound DTMF destination (used to feed the pin-prompt channel), and publishTrack later swaps it for the room's inbound DTMF writer. The swap in publishTrack logs Warnw("media port has unexpected inbound dtmf writer") whenever the previous writer is non-nil, which is always the case, so every inbound call emits a false warning. Make the handoff explicit: either only warn when the previous writer is not the expected pinDTMFWriter (e.g. type assertion), or drop the warning for this specific expected transition.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

}
c.media.WriteAudioTo(local)
return nil
}

Expand Down Expand Up @@ -1657,35 +1677,18 @@ func (c *inboundCall) playAudio(ctx context.Context, frames []msdk.PCM16Sample)
_ = msdk.PlayAudio[msdk.PCM16Sample](ctx, t, rtp.DefFrameDur, frames)
}

func (c *inboundCall) handleDTMF(msg *livekit.SipDTMF) {
if msg == nil {
return
}

func dtmfEventFromSipDTMF(msg *livekit.SipDTMF) dtmf.Event {
code := byte(msg.Code)
digit := byte(0)
if len(msg.Digit) == 1 {
digit = msg.Digit[0]
} else {
digit = dtmf.CodeToChar(code)
}
event := dtmf.Event{
return dtmf.Event{
Code: code,
Digit: digit,
}

if c.forwardDTMF.Load() {
_ = c.lkRoom.SendData(&livekit.SipDTMF{
Code: uint32(code),
Digit: string([]byte{digit}),
}, lksdk.WithDataPublishReliable(true))
return
}
// We should have enough buffer here.
select {
case c.dtmf <- event:
default:
}
}

func (c *inboundCall) transferCall(ctx context.Context, transferTo string, headers map[string]string, dialtone bool) (retErr error) {
Expand All @@ -1701,14 +1704,13 @@ func (c *inboundCall) transferCall(ctx context.Context, transferTo string, heade
rctx, rcancel := context.WithCancel(ctx)
defer rcancel()

// mute the room audio to the SIP participant
w := c.lkRoom.SwapOutput(nil)
// Mute the room audio to the SIP participant.
// Skip closing the existing writer, which is c.audioOut.
_ = c.lkRoom.WriteOutboundAudioTo(nil)

defer func() {
if retErr != nil && !c.done.Load() {
c.lkRoom.SwapOutput(w)
} else if w != nil {
w.Close()
c.lkRoom.WriteOutboundAudioTo(c.audioOut)
}
}()

Expand Down
27 changes: 14 additions & 13 deletions pkg/sip/media_port.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,20 +390,21 @@ func (o *MediaOptions) ApplyDefaults() {
}
}

type MediaSegment interface {
GetAudioWriter() msdk.PCM16Writer
WriteAudioTo(w msdk.PCM16Writer) msdk.PCM16Writer
GetDTMFWriter() msdk.WriteCloser[*livekit.SipDTMF]
WriteDTMFTo(w msdk.WriteCloser[*livekit.SipDTMF]) msdk.WriteCloser[*livekit.SipDTMF]
}

// MediaPort is the insulated media-plane API: UDP/RTP to the wire, SDP negotiation,
// and audio/DTMF endpoints. It does not know about calls, rooms, or SIP dialogs.
type MediaPort interface {
Close()
CloseWait()

MediaSegment
// GetOutboundAudioWriter returns the LK room -> SIP writer.
GetOutboundAudioWriter() msdk.PCM16Writer
// GetOutboundDTMFWriter returns the LK room -> SIP DTMF writer.
GetOutboundDTMFWriter() msdk.WriteCloser[*livekit.SipDTMF]

// WriteInboundAudioTo tells the MediaSegment where to write inbound SIP audio.
WriteInboundAudioTo(w msdk.PCM16Writer) msdk.PCM16Writer
// WriteInboundDTMFTo tells the MediaSegment where to write inbound SIP DTMF.
WriteInboundDTMFTo(w msdk.WriteCloser[*livekit.SipDTMF]) msdk.WriteCloser[*livekit.SipDTMF]

// If there is no offer, this generates an offer.
// If there is an offer, this simply returns the SDP of that offer.
Expand Down Expand Up @@ -731,20 +732,20 @@ func (p *mediaPort) reportPeerCodecs(d sdp.MediaDesc) {

// Plumbing

func (p *mediaPort) GetAudioWriter() msdk.PCM16Writer {
func (p *mediaPort) GetOutboundAudioWriter() msdk.PCM16Writer {
return p.audioOut
}

// WriteAudioTo sets audio writer that will receive decoded PCM from incoming RTP packets.
func (p *mediaPort) WriteAudioTo(w msdk.PCM16Writer) msdk.PCM16Writer {
// WriteInboundAudioTo sets audio writer that will receive decoded PCM from incoming RTP packets.
func (p *mediaPort) WriteInboundAudioTo(w msdk.PCM16Writer) msdk.PCM16Writer {
return p.audioIn.Swap(w)
}

func (p *mediaPort) GetDTMFWriter() msdk.WriteCloser[*livekit.SipDTMF] {
func (p *mediaPort) GetOutboundDTMFWriter() msdk.WriteCloser[*livekit.SipDTMF] {
return &p.dtmfOut
}

func (p *mediaPort) WriteDTMFTo(w msdk.WriteCloser[*livekit.SipDTMF]) msdk.WriteCloser[*livekit.SipDTMF] {
func (p *mediaPort) WriteInboundDTMFTo(w msdk.WriteCloser[*livekit.SipDTMF]) msdk.WriteCloser[*livekit.SipDTMF] {
return p.dtmfIn.Swap(w)
}

Expand Down
12 changes: 6 additions & 6 deletions pkg/sip/media_port_negotiation_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ func roomFrame() msdk.PCM16Sample {
// writeFrames pushes room audio into the port. Write errors are ignored: the in-memory
// UDP pipe is bounded, and a peer that is mid-renegotiation may not be draining it.
func writeFrames(m *mediaPort, frames int) {
w := m.GetAudioWriter()
w := m.GetOutboundAudioWriter()
frame := roomFrame()
for range frames {
_ = w.WriteSample(frame)
Expand Down Expand Up @@ -169,7 +169,7 @@ func TestMediaPortRenegotiation(t *testing.T) {
m1, m2 := newMediaPair(t, nil, nil, "")

recv2 := &recvBuffer{}
m2.WriteAudioTo(recv2)
m2.WriteInboundAudioTo(recv2)
requireAudioFlows(t, m1, recv2)

for range 3 {
Expand Down Expand Up @@ -203,7 +203,7 @@ func TestMediaPortRenegotiation(t *testing.T) {
require.Equal(t, g711.ULawSDPNameAndRate, answerCodec(t, answerData))

recv2 := &recvBuffer{}
m2.WriteAudioTo(recv2)
m2.WriteInboundAudioTo(recv2)
requireAudioFlows(t, m1, recv2)

// G722 samples at 16k, so the encode leaf changes sample rate under the same
Expand Down Expand Up @@ -247,9 +247,9 @@ func TestMediaPortHold(t *testing.T) {
m1, m2 := newMediaPair(t, nil, nil, "")

recv1 := &recvBuffer{}
m1.WriteAudioTo(recv1)
m1.WriteInboundAudioTo(recv1)
recv2 := &recvBuffer{}
m2.WriteAudioTo(recv2)
m2.WriteInboundAudioTo(recv2)

// Baseline: m1 sends to m2.
require.NotNil(t, m1.audioOut.Get())
Expand All @@ -268,7 +268,7 @@ func TestMediaPortHold(t *testing.T) {
assert.True(t, dst.Addr().IsUnspecified(), "held port kept a destination: %v", dst)
}
assert.Nil(t, m1.audioOut.Get(), "held port still accepts room audio")
assert.NoError(t, m1.GetAudioWriter().WriteSample(roomFrame()))
assert.NoError(t, m1.GetOutboundAudioWriter().WriteSample(roomFrame()))

sent := recv2.count()
writeFrames(m1, 10)
Expand Down
4 changes: 2 additions & 2 deletions pkg/sip/media_port_pending_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,11 +197,11 @@ func TestMediaPort(t *testing.T) {

var aliceRecvBuf msdk.PCM16Sample
aliceHandler := msdk.NewPCM16BufferWriter(&aliceRecvBuf, testRate)
alicePort.WriteAudioTo(aliceHandler)
alicePort.WriteInboundAudioTo(aliceHandler)

var bobRecvBuf msdk.PCM16Sample
bobHandler := msdk.NewPCM16BufferWriter(&bobRecvBuf, testRate)
bobPort.WriteAudioTo(bobHandler)
bobPort.WriteInboundAudioTo(bobHandler)

aliceToBob := alicePort.GetAudioWriter()
bobToAlice := bobPort.GetAudioWriter()
Expand Down
16 changes: 8 additions & 8 deletions pkg/sip/media_port_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,7 +299,7 @@ func newMediaPairWithAddr(t testing.TB, ip1, ip2 netip.Addr, opt1, opt2 *MediaOp

// TODO(port-refactor): the encode chain gained the always-on 48k resampler, refresh
// the expected string once the package builds and this can be run.
// w2 := m2.GetAudioWriter()
// w2 := m2.GetOutboundAudioWriter()
// require.Equal(t, "Switch(16000) -> LatencyEntry -> G722(encode) -> ByteEncoder(16000) -> StatsWriter(G722/8000) -> LatencyExit -> RTPWriteStream(1.1.1.1:10000)", w2.String())

return m1, m2
Expand Down Expand Up @@ -339,7 +339,7 @@ func TestMediaTimeout(t *testing.T) {
MediaTimeout: timeout,
}, nil, codec)

w2 := m2.GetAudioWriter()
w2 := m2.GetOutboundAudioWriter()
err := w2.WriteSample(msdk.PCM16Sample{0, 0})
require.NoError(t, err)

Expand All @@ -362,7 +362,7 @@ func TestMediaTimeout(t *testing.T) {
MediaTimeout: timeout,
}, nil, codec)

w2 := m2.GetAudioWriter()
w2 := m2.GetOutboundAudioWriter()

for i := 0; i < 10; i++ {
err := w2.WriteSample(msdk.PCM16Sample{0, 0})
Expand All @@ -382,7 +382,7 @@ func TestMediaTimeout(t *testing.T) {
MediaTimeout: timeout,
}, nil, codec)

w2 := m2.GetAudioWriter()
w2 := m2.GetOutboundAudioWriter()

for i := 0; i < 5; i++ {
err := w2.WriteSample(msdk.PCM16Sample{0, 0})
Expand Down Expand Up @@ -440,7 +440,7 @@ func TestMediaTimeout(t *testing.T) {
MediaTimeout: timeout,
}, nil, codec)

w2 := m2.GetAudioWriter()
w2 := m2.GetOutboundAudioWriter()

for i := 0; i < 5; i++ {
err := w2.WriteSample(msdk.PCM16Sample{0, 0})
Expand Down Expand Up @@ -480,7 +480,7 @@ func TestSymmetricRTP(t *testing.T) {
newAddr := netip.AddrPortFrom(newIP("9.9.9.9"), 9999)
c2.addr = newAddr

err := m2.GetAudioWriter().WriteSample(msdk.PCM16Sample{0, 0})
err := m2.GetOutboundAudioWriter().WriteSample(msdk.PCM16Sample{0, 0})
require.NoError(t, err)

select {
Expand All @@ -504,7 +504,7 @@ func TestSymmetricRTP(t *testing.T) {
newAddr := netip.AddrPortFrom(newIP("9.9.9.9"), 9999)
c2.addr = newAddr

err := m2.GetAudioWriter().WriteSample(msdk.PCM16Sample{0, 0})
err := m2.GetOutboundAudioWriter().WriteSample(msdk.PCM16Sample{0, 0})
require.NoError(t, err)

select {
Expand Down Expand Up @@ -534,7 +534,7 @@ func TestSymmetricRTP(t *testing.T) {
newAddr := netip.AddrPortFrom(newIP("3.3.3.3"), 9999)
c2.addr = newAddr

err := m2.GetAudioWriter().WriteSample(msdk.PCM16Sample{0, 0})
err := m2.GetOutboundAudioWriter().WriteSample(msdk.PCM16Sample{0, 0})
require.NoError(t, err)

select {
Expand Down
Loading