diff --git a/pkg/sip/media_port.go b/pkg/sip/media_port.go index e6df1d85..b130f3eb 100644 --- a/pkg/sip/media_port.go +++ b/pkg/sip/media_port.go @@ -415,7 +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) + p.lastDTMFEvent.Store(math.MaxUint64) if p.opts.IgnorePreanswerData { p.port.startDiscarding() } @@ -463,7 +463,7 @@ type MediaPort struct { 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 + lastDTMFEvent atomic.Uint64 // (rtp timestamp, event code) of last DTMF packet seen } func (p *MediaPort) DisableOut() { @@ -1030,16 +1030,19 @@ func (p *MediaPort) dtmfHandler(h *rtp.Header, payload []byte) error { 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) + // RFC 4733 requires all packets of a given digit to share identical timestamps. + // Some SIP devices or carriers may reuse the timestamp of the previous digit + // for the next one, so we combine timestamp and event code for deduplication. + // The marker bit could be used instead, but it is prone to occasional loss. + eventID := uint64(h.Timestamp)<<8 | uint64(ev.Code) + if eventID == p.lastDTMFEvent.Load() { + return nil + } + p.lastDTMFEvent.Store(eventID) fnc(ev) return nil } diff --git a/pkg/sip/media_port_test.go b/pkg/sip/media_port_test.go index 64b693b4..fa2680f6 100644 --- a/pkg/sip/media_port_test.go +++ b/pkg/sip/media_port_test.go @@ -872,7 +872,7 @@ func TestMediaPortDTMF(t *testing.T) { 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) + p.lastDTMFEvent.Store(math.MaxUint64) got := "" p.HandleDTMF(func(ev dtmf.Event) { t.Logf("received DTMF event: %+v", ev) @@ -895,6 +895,44 @@ func TestMediaPortDTMF(t *testing.T) { } } +// TestMediaPortDTMFSameTimestamp verifies that DTMF digits sharing the same +// RTP timestamp (a non-RFC-compliant but observed behaviour from some SIP +// devices/carriers) are still reported individually after the (timestamp, +// event code) deduplication fix. +func TestMediaPortDTMFSameTimestamp(t *testing.T) { + // Generate packets for two individual digits, then force the second + // digit's packets to reuse the first digit's timestamp. + first := generateDTMFPackets(t, "0")[0] + second := generateDTMFPackets(t, "1")[0] + + sharedTS := first[0].Header.Timestamp + for i := range second { + second[i].Header.Timestamp = sharedTS + } + + p := &MediaPort{} + p.lastDTMFEvent.Store(math.MaxUint64) + 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 _, pkt := range first { + h := pkt.Header + t.Logf("sending first digit packet: seq=%d, ts=%d, marker=%t", h.SequenceNumber, h.Timestamp, h.Marker) + require.NoError(t, p.dtmfHandler(&h, pkt.Payload)) + } + for _, pkt := range second { + h := pkt.Header + t.Logf("sending second digit packet: seq=%d, ts=%d, marker=%t", h.SequenceNumber, h.Timestamp, h.Marker) + require.NoError(t, p.dtmfHandler(&h, pkt.Payload)) + } + + t.Logf("got: %s", got) + require.Equal(t, "01", got) +} + // Test util for incrementing prometheus counter metrics. func gatherCounter(t testing.TB, name string, labels map[string]string) float64 { t.Helper()