diff --git a/dtmf/dtmf.go b/dtmf/dtmf.go index 7190ca8..093ceb6 100644 --- a/dtmf/dtmf.go +++ b/dtmf/dtmf.go @@ -132,6 +132,15 @@ var eventToChar = [16]byte{ codeA: 'a', codeB: 'b', codeC: 'c', codeD: 'd', } +// CodeToChar converts a DTMF event code to its character representation. +// It returns 0 for unknown codes. +func CodeToChar(code byte) byte { + if int(code) < len(eventToChar) { + return eventToChar[code] + } + return 0 +} + var charToEvent = map[byte]byte{ '0': code0, '1': code1, '2': code2, '3': code3, '4': code4, '5': code5, '6': code6, '7': code7, '8': code8, '9': code9, diff --git a/media.go b/media.go index bf37073..9487b21 100644 --- a/media.go +++ b/media.go @@ -85,9 +85,8 @@ func NewSwitchWriter(sampleRate int) *SwitchWriter { } type SwitchWriter struct { - ptr atomic.Pointer[PCM16Writer] - sampleRate atomic.Int32 - disabled atomic.Bool + WriteCloserSwitch[PCM16Sample] + disabled atomic.Bool } func (s *SwitchWriter) Enable() { @@ -99,29 +98,26 @@ func (s *SwitchWriter) Disable() { } func (s *SwitchWriter) Get() PCM16Writer { - ptr := s.ptr.Load() + ptr := s.WriteCloserSwitch.Get() if ptr == nil { - return nil + return nil // Untyped nil } - return *ptr + return ptr } // Swap sets an underlying writer and returns the old one. // Caller is responsible for closing the old writer. func (s *SwitchWriter) Swap(w PCM16Writer) PCM16Writer { - var old *PCM16Writer - if w == nil { - old = s.ptr.Swap(nil) - } else { + if w != nil { if rate := s.SampleRate(); rate != w.SampleRate() { w = ResampleWriter(w, rate) } - old = s.ptr.Swap(&w) } + old := s.WriteCloserSwitch.Swap(w) if old == nil { - return nil + return nil // Untyped nil } - return *old + return old } func (s *SwitchWriter) String() string { @@ -135,14 +131,14 @@ func (s *SwitchWriter) SetSampleRate(rate int) { if rate <= 0 { panic("invalid sample rate") } - if !s.sampleRate.CompareAndSwap(-1, int32(rate)) { + if !s.WriteCloserSwitch.sampleRate.CompareAndSwap(-1, int32(rate)) { panic("sample rate can only be changed once") } } // SampleRate returns an expected sample rate for this writer. It panics if the sample rate is not specified. func (s *SwitchWriter) SampleRate() int { - rate := int(s.sampleRate.Load()) + rate := s.WriteCloserSwitch.SampleRate() if rate == 0 { panic("switch writer not initialized") } else if rate < 0 { @@ -151,23 +147,94 @@ func (s *SwitchWriter) SampleRate() int { return rate } -func (s *SwitchWriter) Close() error { - ptr := s.ptr.Swap(nil) - if ptr == nil { +func (s *SwitchWriter) WriteSample(sample PCM16Sample) error { + if s.disabled.Load() { return nil } - return (*ptr).Close() + return s.WriteCloserSwitch.WriteSample(sample) } -func (s *SwitchWriter) WriteSample(sample PCM16Sample) error { - if s.disabled.Load() { +// NewWriteCloserSwitch creates a switch that expects writers with the given sample rate. +// If a positive sample rate is provided, it is locked in at the start. +// If a zero or negative sample rate is provided, the real rate will be taken +// from the first downstream writer, and locked to that rate at that time. +func NewWriteCloserSwitch[T any](sampleRate int) *WriteCloserSwitch[T] { + s := &WriteCloserSwitch[T]{} + if sampleRate > 0 { + s.sampleRate.Store(int32(sampleRate)) + } + return s +} + +// WriteCloserSwitch is a WriteCloser that forwards samples to an underlying writer, +// which can be replaced atomically with Swap. Writes are dropped while no writer is attached. +// All writers must agree on the sample rate. +type WriteCloserSwitch[T any] struct { // msdk.WriteCloser[T] + sampleRate atomic.Int32 // Prevents changing sample rate after the switch is created + w atomic.Pointer[WriteCloser[T]] +} + +func (s *WriteCloserSwitch[T]) String() string { + w := s.w.Load() + if w == nil { + return "WriteCloserSwitch(nil)" + } + return fmt.Sprintf("WriteCloserSwitch(%d) -> %v", s.SampleRate(), *w) +} + +// SampleRate returns the sample rate expected from the underlying writers, +// or -1 if it is still unset. +func (s *WriteCloserSwitch[T]) SampleRate() int { + if rate := s.sampleRate.Load(); rate > 0 { + return int(rate) + } + return -1 +} + +func (s *WriteCloserSwitch[T]) WriteSample(sample T) error { + w := s.w.Load() + if w == nil { return nil } - w := s.Get() + return (*w).WriteSample(sample) +} + +func (s *WriteCloserSwitch[T]) Close() error { + w := s.w.Load() if w == nil { return nil } - return w.WriteSample(sample) + return (*w).Close() +} + +func (s *WriteCloserSwitch[T]) Get() WriteCloser[T] { + ptr := s.w.Load() + if ptr == nil { + return nil + } + return *ptr +} + +// Swap sets an underlying writer and returns the old one. +// Caller is responsible for closing the old writer. +// When switch sample rate is unset, it is set to the new writer's sample rate. +// If sample rate is already set, this method panics on sample rate mismatch. +func (s *WriteCloserSwitch[T]) Swap(w WriteCloser[T]) WriteCloser[T] { + var old *WriteCloser[T] + if w != nil { + newRate := int32(w.SampleRate()) + oldRate := s.sampleRate.Swap(newRate) + if oldRate > 0 && oldRate != newRate { + panic(fmt.Sprintf("sample rate mismatch: newRate %d, oldRate %d", newRate, oldRate)) + } + old = s.w.Swap(&w) + } else { + old = s.w.Swap(nil) + } + if old == nil { + return nil + } + return *old } type MultiWriter[T any] []WriteCloser[T] diff --git a/sdp/offer.go b/sdp/offer.go index ae0f94a..7ef10c3 100644 --- a/sdp/offer.go +++ b/sdp/offer.go @@ -356,10 +356,11 @@ func (d *Offer) Answer(publicIp netip.Addr, rtpListenerPort int, enc Encryption) DTMFType: audio.DTMFType, }, }, &MediaConfig{ - Local: src, - Remote: d.Addr, - Audio: *audio, - Crypto: sconf, + Local: src, + Remote: d.Addr, + Audio: *audio, + Crypto: sconf, + PeerDirection: d.Direction, }, nil } @@ -391,10 +392,11 @@ func (d *Answer) apply(offer *Offer, enc Encryption, generateLocalSDP bool) (*Me return nil, nil, ErrNoCommonCrypto } mc := &MediaConfig{ - Local: offer.Addr, - Remote: d.Addr, - Audio: *audio, - Crypto: sconf, + Local: offer.Addr, + Remote: d.Addr, + Audio: *audio, + Crypto: sconf, + PeerDirection: d.Direction, } if !generateLocalSDP { @@ -760,10 +762,11 @@ func ParseMedia(d *sdp.MediaDescription) (*MediaDesc, error) { // MediaConfig is the canonical representation of the negotiated session. type MediaConfig struct { - Local netip.AddrPort - Remote netip.AddrPort - Audio AudioConfig - Crypto *srtp.Config + Local netip.AddrPort + Remote netip.AddrPort + Audio AudioConfig + Crypto *srtp.Config + PeerDirection sdp.Direction // RFC 3264, offer direction for server, answer direction for client } type AudioConfig struct {