Skip to content
Merged
Show file tree
Hide file tree
Changes from 9 commits
Commits
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
7 changes: 7 additions & 0 deletions dtmf/dtmf.go
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,13 @@ var eventToChar = [16]byte{
codeA: 'a', codeB: 'b', codeC: 'c', codeD: 'd',
}

func CodeToChar(code byte) byte {
Comment thread
alexlivekit marked this conversation as resolved.
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,
Expand Down
100 changes: 77 additions & 23 deletions media.go
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand All @@ -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 {
Expand All @@ -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 {
Expand All @@ -151,23 +147,81 @@ 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() {
func NewWriteCloserSwitch[T any](sampleRate int) *WriteCloserSwitch[T] {
Comment thread
alexlivekit marked this conversation as resolved.
s := &WriteCloserSwitch[T]{}
if sampleRate > 0 {
s.sampleRate.Store(int32(sampleRate))
}
return s
}

type WriteCloserSwitch[T any] struct { // msdk.WriteCloser[T]
Comment thread
alexlivekit marked this conversation as resolved.
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 "Switch(nil)"
Comment thread
alexlivekit marked this conversation as resolved.
Outdated
}
return fmt.Sprintf("Switch(%d) -> %v", s.SampleRate(), *w)
Comment thread
alexlivekit marked this conversation as resolved.
Outdated
}

func (s *WriteCloserSwitch[T]) SampleRate() int {
Comment thread
alexlivekit marked this conversation as resolved.
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
}

func (s *WriteCloserSwitch[T]) Swap(w WriteCloser[T]) WriteCloser[T] {
Comment thread
alexlivekit marked this conversation as resolved.
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: expected %d, actual %d", newRate, oldRate))
Comment thread
alexlivekit marked this conversation as resolved.
Outdated
}
old = s.w.Swap(&w)
} else {
old = s.w.Swap(nil)
}
if old == nil {
return nil
}
return *old
}

type MultiWriter[T any] []WriteCloser[T]
Expand Down
27 changes: 15 additions & 12 deletions sdp/offer.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
Loading