Updating DTMF detection to work based off of timestamp rather than first marker (#742)

Author: alexlivekit

pkg/sip/media_port.go

@@ -415,6 +415,7 @@ func NewMediaPortWith(tid traceid.ID, log logger.Logger, mon *stats.CallMonitor,
 		audioIn:          msdk.NewSwitchWriter(inSampleRate),
 		stats:            opts.Stats,
 	}
+	p.lastDTMFTimestamp.Store(math.MaxUint32)
 	if p.opts.IgnorePreanswerData {
 		p.port.startDiscarding()
 	}
@@ -457,11 +458,12 @@ type MediaPort struct {
 	dtmfOutRTP   *rtp.Stream
 	dtmfOutAudio msdk.PCM16Writer
 
-	audioOutRTP    *rtp.Stream
-	audioOut       *msdk.SwitchWriter // LK PCM -> SIP RTP
-	audioIn        *msdk.SwitchWriter // SIP RTP -> LK PCM
-	audioInHandler rtp.Handler        // for debug only
-	dtmfIn         atomic.Pointer[func(ev dtmf.Event)]
+	audioOutRTP       *rtp.Stream
+	audioOut          *msdk.SwitchWriter // LK PCM -> SIP RTP
+	audioIn           *msdk.SwitchWriter // SIP RTP -> LK PCM
+	audioInHandler    rtp.Handler        // for debug only
+	dtmfIn            atomic.Pointer[func(ev dtmf.Event)]
+	lastDTMFTimestamp atomic.Uint32 // rtp timestamp of last DTMF packet seen
 }
 
 func (p *MediaPort) DisableOut() {
@@ -1015,9 +1017,20 @@ func (p *MediaPort) dtmfHandler(h *rtp.Header, payload []byte) error {
 		return nil
 	}
 	fnc := *ptr
-	if ev, ok := dtmf.DecodeRTP(h, payload); ok && fnc != nil {
-		fnc(ev)
+	if fnc == nil {
+		return nil
+	}
+	// RFC 4733 requires all packets of a given digit to share identical timestamps.
+	// The marker bit could be used instead, but it is prone to occasional loss.
+	if h.Timestamp == p.lastDTMFTimestamp.Load() {
+		return nil
+	}
+	ev, err := dtmf.Decode(payload)
+	if err != nil {
+		return nil
 	}
+	p.lastDTMFTimestamp.Store(h.Timestamp)
+	fnc(ev)
 	return nil
 }
 

pkg/sip/media_port_test.go

@@ -15,6 +15,7 @@
 package sip
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"math"
@@ -32,6 +33,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	msdk "github.com/livekit/media-sdk"
+	"github.com/livekit/media-sdk/dtmf"
 	"github.com/livekit/media-sdk/rtp"
 	"github.com/livekit/media-sdk/sdp"
 	"github.com/livekit/mediatransportutil/pkg/rtcconfig"
@@ -802,3 +804,75 @@ func TestSymmetricRTP(t *testing.T) {
 		require.Equal(t, newAddr.String(), curDstPtr.String())
 	})
 }
+
+func generateDTMFPackets(t *testing.T, digits string) [][]*rtp.Packet {
+	t.Helper()
+	var buf rtp.Buffer
+	packets := make([][]*rtp.Packet, len(digits))
+	last := len(buf)
+	w := rtp.NewSeqWriter(&buf).NewStream(101, dtmf.SampleRate)
+	timestamp := uint32(1000)
+	for i := range digits {
+		err := dtmf.Write(context.Background(), nil, w, timestamp, digits[i:i+1])
+		require.NoError(t, err)
+		require.NotEmpty(t, buf)
+		timestamp += uint32(dtmf.SampleRate / 2)
+		packets[i] = slices.Clone(buf[last:])
+		last = len(buf)
+	}
+	return packets
+}
+
+func dropPackets(t *testing.T, dropType string, packets []*rtp.Packet) []*rtp.Packet {
+	t.Helper()
+	switch dropType {
+	case "none":
+		return packets
+	case "first":
+		require.Greater(t, len(packets), 3)
+		return packets[3:]
+	case "last":
+		require.Greater(t, len(packets), 3)
+		return packets[:len(packets)-3]
+	case "middle":
+		require.Greater(t, len(packets), 6)
+		ret := slices.Clone(packets[:3])
+		ret = append(ret, packets[len(packets)-3:]...)
+		return ret
+	default:
+		t.Fatal("unknown drop type: " + dropType)
+		return nil
+	}
+}
+
+func TestMediaPortDTMF(t *testing.T) {
+	digitCases := []string{"1", "12", "123"}
+	lossCases := []string{"none", "first", "last", "middle"}
+
+	for _, digits := range digitCases {
+		packets := generateDTMFPackets(t, digits)
+		for _, lossPackets := range lossCases {
+			t.Run(fmt.Sprintf("digits=%s/loss=%s", digits, lossPackets), func(t *testing.T) {
+				p := &MediaPort{}
+				p.lastDTMFTimestamp.Store(math.MaxUint32)
+				got := ""
+				p.HandleDTMF(func(ev dtmf.Event) {
+					t.Logf("received DTMF event: %+v", ev)
+					got = fmt.Sprintf("%s%s", got, strconv.Itoa(int(ev.Code)))
+				})
+				for _, digitPackets := range packets {
+					sendPackets := dropPackets(t, lossPackets, digitPackets)
+					t.Logf("sending %d/%d packets", len(sendPackets), len(digitPackets))
+					for _, pkt := range sendPackets {
+						h := pkt.Header
+						t.Logf("sending packet: seq=%d, ts=%d, marker=%t", h.SequenceNumber, h.Timestamp, h.Marker)
+						require.NoError(t, p.dtmfHandler(&h, pkt.Payload))
+					}
+				}
+				t.Logf("sent: %s", digits)
+				t.Logf("got: %s", got)
+				require.Equal(t, digits, got)
+			})
+		}
+	}
+}