ts-mobile-go/go/teamspeak/receive_timeline.go

package teamspeak

import "time"

type receiveTimeline struct {
	clientID    int
	decoder     voiceDecoder
	packets     map[uint16]voicePacket
	expected    uint16
	initialized bool
	ready       bool
	plc         int
	lossWait    int
	muted       bool
	speaking    bool
	starved     bool
	hasPlayed   bool
	lastPacket  time.Time
	lastActive  time.Time
	pending     []int16
	lastReal    bool
}

const (
	receiveLossWaitTicks = 2
	receiveTalkSpurtGap  = 400 * time.Millisecond
	receiveMaxPending    = receiveFrameInterleavedSamples * 12
)

func newReceiveTimeline(clientID int, decoder voiceDecoder) *receiveTimeline {
	return &receiveTimeline{clientID: clientID, decoder: decoder, packets: make(map[uint16]voicePacket)}
}

func (t *receiveTimeline) release() {
	if t.decoder != nil {
		t.decoder.Release()
		t.decoder = nil
	}
}

func seqAhead(a, b uint16) bool { return int16(a-b) > 0 }

func (t *receiveTimeline) resetForTalkSpurt(sequence uint16) {
	clear(t.packets)
	t.pending = t.pending[:0]
	t.expected = sequence
	t.initialized = true
	t.ready = false
	t.plc = 0
	t.lossWait = 0
	t.starved = false
}

func (t *receiveTimeline) push(p voicePacket) {
	if _, exists := t.packets[p.sequence]; exists {
		return
	}

	// A quiet gap or an exhausted timeline starts a new packet epoch. This is
	// deliberately checked before the normal forward-window rejection so a
	// resumed talk spurt cannot be trapped behind an obsolete expected cursor.
	if t.initialized && !t.lastPacket.IsZero() &&
		(p.at.Sub(t.lastPacket) >= receiveTalkSpurtGap ||
			(t.hasPlayed && t.starved && seqAhead(p.sequence, t.expected) &&
				int(uint16(p.sequence-t.expected)) > receiveMaxPackets)) {
		t.resetForTalkSpurt(p.sequence)
	}
	if !t.initialized {
		t.expected = p.sequence
		t.initialized = true
	}
	if p.sequence != t.expected && !seqAhead(p.sequence, t.expected) {
		return
	}
	if seqAhead(p.sequence, t.expected) && int(uint16(p.sequence-t.expected)) > receiveMaxPackets {
		return
	}
	if len(t.packets) >= receiveMaxPackets {
		return
	}
	t.packets[p.sequence] = p
	t.lastPacket = p.at
	if len(t.packets) >= receivePrebuffer {
		t.ready = true
	}
}

func (t *receiveTimeline) takeFrame(pcm []int16) ([]int16, bool) {
	if len(pcm) == 0 {
		return nil, false
	}
	if len(t.pending)+len(pcm) > receiveMaxPending {
		// Keep the newest decoded audio when a decoder returns a long burst.
		over := len(t.pending) + len(pcm) - receiveMaxPending
		if over >= len(t.pending) {
			t.pending = t.pending[:0]
		} else {
			t.pending = t.pending[over:]
		}
	}
	t.pending = append(t.pending, pcm...)
	if len(t.pending) < receiveFrameInterleavedSamples {
		return nil, false
	}
	frame := append([]int16(nil), t.pending[:receiveFrameInterleavedSamples]...)
	t.pending = t.pending[receiveFrameInterleavedSamples:]
	return frame, true
}

func (t *receiveTimeline) popPending() ([]int16, bool) {
	if len(t.pending) < receiveFrameInterleavedSamples {
		return nil, false
	}
	frame := append([]int16(nil), t.pending[:receiveFrameInterleavedSamples]...)
	t.pending = t.pending[receiveFrameInterleavedSamples:]
	return frame, true
}

func (t *receiveTimeline) earliestPacket() (uint16, bool) {
	var earliest uint16
	found := false
	for sequence := range t.packets {
		if !found || seqAhead(earliest, sequence) {
			earliest, found = sequence, true
		}
	}
	return earliest, found
}

// next returns at most one fixed output frame. It never advances expected while
// the packet buffer is empty: an empty buffer is normal talk-spurt silence, not
// confirmed packet loss. lastReal distinguishes decoded audio from PLC for UI.
func (t *receiveTimeline) next() ([]int16, bool) {
	t.lastReal = false
	if frame, ok := t.popPending(); ok {
		t.starved = false
		t.lastReal = true
		return frame, true
	}
	if !t.ready || t.decoder == nil {
		t.starved = true
		return nil, false
	}
	if p, ok := t.packets[t.expected]; ok {
		delete(t.packets, t.expected)
		t.expected++
		t.plc = 0
		t.lossWait = 0
		pcm, err := t.decoder.Decode(p.data)
		if err != nil {
			t.starved = true
			return nil, false
		}
		frame, active := t.takeFrame(pcm)
		if active {
			t.starved = false
			t.hasPlayed = true
			t.lastReal = true
		}
		return frame, active
	}

	// Do not PLC ordinary silence. A later packet is evidence of a gap.
	if _, found := t.earliestPacket(); !found {
		t.starved = true
		return nil, false
	}
	if t.lossWait < receiveLossWaitTicks {
		t.lossWait++
		t.starved = true
		return nil, false
	}
	if t.plc < receiveMaxPLC {
		t.plc++
		t.lossWait = 0
		t.expected++
		pcm, err := t.decoder.DecodeLost()
		if err != nil {
			t.starved = true
			return nil, false
		}
		frame, active := t.takeFrame(pcm)
		if active {
			// PLC preserves an existing speaking state but never starts one.
			t.starved = false
		}
		return frame, active
	}

	if earliest, found := t.earliestPacket(); found {
		t.expected = earliest
		t.plc = 0
		t.lossWait = 0
		t.starved = false
		return t.next()
	}
	t.starved = true
	return nil, false
}

func mixPCM16(acc []int32) []byte {
	out := make([]byte, receiveFrameBytes)
	for i, sample := range acc {
		if sample > 32767 {
			sample = 32767
		} else if sample < -32768 {
			sample = -32768
		}
		out[2*i], out[2*i+1] = byte(int16(sample)), byte(int16(sample)>>8)
	}
	return out
}