282 lines
6.9 KiB
Go
282 lines
6.9 KiB
Go
package teamspeak
|
|
|
|
import (
|
|
"errors"
|
|
"log"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
receiveSampleRate = 48000
|
|
receiveChannels = 2
|
|
receiveFrameSamplesPerChannel = 960
|
|
receiveFrameInterleavedSamples = receiveFrameSamplesPerChannel * receiveChannels
|
|
receiveFrameBytes = receiveFrameInterleavedSamples * 2
|
|
receiveMaxDecodeSamplesPerChannel = 5760
|
|
receivePrebuffer = 3
|
|
receiveMaxPackets = 12
|
|
receiveMaxPLC = 3
|
|
)
|
|
|
|
// voiceDecoder decodes one remote Opus stream. It is intentionally private so
|
|
// gomobile does not try to expose the decoder interface in the public AAR API.
|
|
// Android production wiring will provide a libopus-backed implementation.
|
|
type voiceDecoder interface {
|
|
Decode(packet []byte) ([]int16, error)
|
|
DecodeLost() ([]int16, error)
|
|
Release()
|
|
}
|
|
|
|
type voiceDecoderFactory interface {
|
|
NewvoiceDecoder(clientID int, codec int) (voiceDecoder, error)
|
|
}
|
|
|
|
type voicePacket struct {
|
|
clientID int
|
|
codec int
|
|
sequence uint16
|
|
data []byte
|
|
at time.Time
|
|
}
|
|
|
|
type audioCommand struct {
|
|
kind string
|
|
clientID int
|
|
muted bool
|
|
done chan struct{}
|
|
}
|
|
|
|
// receiveAudio owns all receive-side decoder, jitter, mute, speaking, and
|
|
// mixing state on one goroutine. Its ingress method is safe for protocol
|
|
// callback goroutines and never waits for decoding or playback.
|
|
type receiveAudio struct {
|
|
factory voiceDecoderFactory
|
|
packets uint64
|
|
decoded uint64
|
|
mixed uint64
|
|
ingress chan voicePacket
|
|
commands chan audioCommand
|
|
stop chan struct{}
|
|
done chan struct{}
|
|
onPCM func([]byte)
|
|
onSpeaking func(int, bool)
|
|
onDrop func()
|
|
stopOnce sync.Once
|
|
startOnce sync.Once
|
|
stateMu sync.Mutex
|
|
started bool
|
|
}
|
|
|
|
func newReceiveAudio(factory voiceDecoderFactory, onPCM func([]byte), onSpeaking func(int, bool)) *receiveAudio {
|
|
return &receiveAudio{
|
|
factory: factory, ingress: make(chan voicePacket, 64), commands: make(chan audioCommand, 32),
|
|
stop: make(chan struct{}), done: make(chan struct{}), onPCM: onPCM, onSpeaking: onSpeaking,
|
|
}
|
|
}
|
|
|
|
func (r *receiveAudio) start() {
|
|
r.startOnce.Do(func() {
|
|
log.Printf("[TSVoice] receive worker started factory=%T", r.factory)
|
|
r.stateMu.Lock()
|
|
r.started = true
|
|
r.stateMu.Unlock()
|
|
go r.run()
|
|
})
|
|
}
|
|
|
|
func (r *receiveAudio) stopAndWait() {
|
|
r.stateMu.Lock()
|
|
started := r.started
|
|
r.stateMu.Unlock()
|
|
if !started {
|
|
return
|
|
}
|
|
r.stopOnce.Do(func() { close(r.stop) })
|
|
<-r.done
|
|
}
|
|
|
|
func (r *receiveAudio) enqueue(clientID, codec int, sequence uint16, data []byte) {
|
|
if clientID <= 0 || (codec != 4 && codec != 5) || len(data) == 0 {
|
|
return
|
|
}
|
|
packet := voicePacket{clientID: clientID, codec: codec, sequence: sequence, data: append([]byte(nil), data...), at: time.Now()}
|
|
count := atomic.AddUint64(&r.packets, 1)
|
|
if count <= 5 || count%250 == 0 {
|
|
log.Printf("[TSVoice] ingress count=%d client=%d codec=%d sequence=%d payload=%d", count, clientID, codec, sequence, len(data))
|
|
}
|
|
select {
|
|
case r.ingress <- packet:
|
|
default:
|
|
if r.onDrop != nil {
|
|
r.onDrop()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *receiveAudio) command(cmd audioCommand) {
|
|
select {
|
|
case r.commands <- cmd:
|
|
default:
|
|
if cmd.done != nil {
|
|
close(cmd.done)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *receiveAudio) run() {
|
|
defer close(r.done)
|
|
timelines := make(map[int]*receiveTimeline)
|
|
ticker := time.NewTicker(20 * time.Millisecond)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-r.stop:
|
|
for id, t := range timelines {
|
|
t.release()
|
|
if t.speaking && r.onSpeaking != nil {
|
|
r.onSpeaking(id, false)
|
|
}
|
|
}
|
|
return
|
|
case cmd := <-r.commands:
|
|
r.applyCommand(timelines, cmd)
|
|
case p := <-r.ingress:
|
|
r.accept(timelines, p)
|
|
case now := <-ticker.C:
|
|
// Drain a bounded amount before every playout deadline so ingress
|
|
// bursts cannot indefinitely postpone a 20 ms output frame.
|
|
for i := 0; i < 32; i++ {
|
|
select {
|
|
case cmd := <-r.commands:
|
|
r.applyCommand(timelines, cmd)
|
|
case p := <-r.ingress:
|
|
r.accept(timelines, p)
|
|
default:
|
|
i = 32
|
|
}
|
|
}
|
|
r.tick(timelines, now)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *receiveAudio) accept(timelines map[int]*receiveTimeline, p voicePacket) {
|
|
t := timelines[p.clientID]
|
|
if t == nil {
|
|
if r.factory == nil {
|
|
return
|
|
}
|
|
decoder, err := r.factory.NewvoiceDecoder(p.clientID, p.codec)
|
|
if err != nil || decoder == nil {
|
|
log.Printf("[TSVoice] decoder create failed client=%d codec=%d err=%v", p.clientID, p.codec, err)
|
|
return
|
|
}
|
|
log.Printf("[TSVoice] decoder created client=%d codec=%d", p.clientID, p.codec)
|
|
t = newReceiveTimeline(p.clientID, decoder)
|
|
timelines[p.clientID] = t
|
|
}
|
|
t.push(p)
|
|
}
|
|
|
|
func (r *receiveAudio) applyCommand(timelines map[int]*receiveTimeline, cmd audioCommand) {
|
|
switch cmd.kind {
|
|
case "mute":
|
|
if t := timelines[cmd.clientID]; t != nil {
|
|
t.muted = cmd.muted
|
|
}
|
|
case "remove":
|
|
if t := timelines[cmd.clientID]; t != nil {
|
|
t.release()
|
|
if t.speaking && r.onSpeaking != nil {
|
|
r.onSpeaking(cmd.clientID, false)
|
|
}
|
|
delete(timelines, cmd.clientID)
|
|
}
|
|
case "clear":
|
|
for id, t := range timelines {
|
|
t.release()
|
|
if t.speaking && r.onSpeaking != nil {
|
|
r.onSpeaking(id, false)
|
|
}
|
|
delete(timelines, id)
|
|
}
|
|
}
|
|
if cmd.done != nil {
|
|
close(cmd.done)
|
|
}
|
|
}
|
|
|
|
func (r *receiveAudio) tick(timelines map[int]*receiveTimeline, now time.Time) {
|
|
mix := make([]int32, receiveFrameInterleavedSamples)
|
|
for id, t := range timelines {
|
|
if now.Sub(t.lastPacket) > 2*time.Second {
|
|
t.release()
|
|
if t.speaking && r.onSpeaking != nil {
|
|
r.onSpeaking(id, false)
|
|
}
|
|
delete(timelines, id)
|
|
continue
|
|
}
|
|
pcm, active := t.next()
|
|
if active {
|
|
count := atomic.AddUint64(&r.decoded, 1)
|
|
if count <= 5 || count%250 == 0 {
|
|
log.Printf("[TSVoice] decoded count=%d client=%d", count, id)
|
|
}
|
|
if t.lastReal {
|
|
t.lastActive = now
|
|
if !t.speaking {
|
|
t.speaking = true
|
|
if r.onSpeaking != nil {
|
|
r.onSpeaking(id, true)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if t.speaking && now.Sub(t.lastActive) > 400*time.Millisecond {
|
|
t.speaking = false
|
|
if r.onSpeaking != nil {
|
|
r.onSpeaking(id, false)
|
|
}
|
|
}
|
|
if !t.muted && len(pcm) == receiveFrameInterleavedSamples {
|
|
for i, sample := range pcm {
|
|
mix[i] += int32(sample)
|
|
}
|
|
}
|
|
}
|
|
if r.onPCM != nil && len(timelines) > 0 {
|
|
count := atomic.AddUint64(&r.mixed, 1)
|
|
if count <= 5 || count%250 == 0 {
|
|
log.Printf("[TSVoice] mixed count=%d clients=%d", count, len(timelines))
|
|
}
|
|
r.onPCM(mixPCM16(mix))
|
|
}
|
|
}
|
|
|
|
var errBadFrame = errors.New("voice decoder returned a non-20ms frame")
|
|
|
|
func voiceCodecChannels(codec int) int {
|
|
if codec == 5 {
|
|
return receiveChannels
|
|
}
|
|
return 1
|
|
}
|
|
|
|
func normalizeDecodedPCM(pcm []int16, channels int) []int16 {
|
|
if channels == receiveChannels {
|
|
return pcm
|
|
}
|
|
return monoToStereo(pcm)
|
|
}
|
|
|
|
func monoToStereo(mono []int16) []int16 {
|
|
stereo := make([]int16, len(mono)*receiveChannels)
|
|
for i, sample := range mono {
|
|
stereo[i*receiveChannels], stereo[i*receiveChannels+1] = sample, sample
|
|
}
|
|
return stereo
|
|
}
|