Attempt to improve over all voice code.

gumble/gumble/audio.go

@@ -85,6 +85,11 @@ type AudioPacket struct {
 	Client *Client
 	Sender *User
 	Target *VoiceTarget
+	Final  bool
+	// LocallyMuted and LocalMuteGeneration snapshot the receiver's local mute
+	// state when the packet entered the playback queue.
+	LocallyMuted        bool
+	LocalMuteGeneration uint64
 
 	AudioBuffer
 

gumble/gumble/audio_buffer_test.go

@@ -0,0 +1,178 @@
+package gumble
+
+import (
+	"sync"
+	"testing"
+
+	"git.stormux.org/storm/barnard/gumble/go-openal/openal"
+)
+
+type testAudioListener struct {
+	events chan *AudioStreamEvent
+}
+
+func (l testAudioListener) OnAudioStream(event *AudioStreamEvent) {
+	if l.events != nil {
+		l.events <- event
+	}
+}
+
+func TestAudioBufferCount(t *testing.T) {
+	tests := []struct {
+		configured int
+		want       int
+	}{
+		{configured: -1, want: AudioMinimumBufferCount},
+		{configured: 0, want: AudioMinimumBufferCount},
+		{configured: 2, want: AudioMinimumBufferCount},
+		{configured: 16, want: 16},
+	}
+
+	for _, tt := range tests {
+		config := &Config{Buffers: tt.configured}
+		if got := config.AudioBufferCount(); got != tt.want {
+			t.Errorf("AudioBufferCount() with %d = %d, want %d", tt.configured, got, tt.want)
+		}
+	}
+}
+
+func TestAudioListenerStreamBufferingAndCleanup(t *testing.T) {
+	var listeners AudioListeners
+	events := make(chan *AudioStreamEvent, 1)
+	detacher := listeners.Attach(testAudioListener{events: events})
+	user := &User{Session: 7}
+	packet := &AudioPacket{AudioBuffer: AudioBuffer{1}}
+
+	listeners.dispatchAudio(nil, user, packet, 16)
+	event := <-events
+	stream := event.C
+	if got := cap(stream); got != 16 {
+		t.Fatalf("stream capacity = %d, want 16", got)
+	}
+	if got := <-stream; got != packet {
+		t.Fatal("expected the packet to be delivered to the stream")
+	}
+	listeners.dispatchAudio(nil, user, packet, 16)
+	select {
+	case <-events:
+		t.Fatal("expected an existing stream to be reused without another callback")
+	default:
+	}
+	<-stream
+
+	listeners.closeUserAudio(user)
+	if _, ok := <-stream; ok {
+		t.Fatal("expected user audio stream to be closed")
+	}
+	item := detacher.(*audioEventItem)
+	if _, ok := item.streams[user]; ok {
+		t.Fatal("expected user audio stream to be removed")
+	}
+}
+
+func TestAudioListenerDetachClosesStreams(t *testing.T) {
+	var listeners AudioListeners
+	events := make(chan *AudioStreamEvent, 1)
+	detacher := listeners.Attach(testAudioListener{events: events})
+	listeners.dispatchAudio(nil, &User{Session: 9}, &AudioPacket{}, 8)
+	stream := (<-events).C
+	<-stream
+
+	detacher.Detach()
+	if _, ok := <-stream; ok {
+		t.Fatal("expected detach to close its audio streams")
+	}
+	if listeners.head != nil || listeners.tail != nil {
+		t.Fatal("expected detached listener to be unlinked")
+	}
+}
+
+func TestAudioListenerDropsOldestPacketWhenFull(t *testing.T) {
+	var listeners AudioListeners
+	events := make(chan *AudioStreamEvent, 1)
+	listeners.Attach(testAudioListener{events: events})
+	user := &User{Session: 11}
+
+	for sample := int16(1); sample <= 4; sample++ {
+		listeners.dispatchAudio(nil, user, &AudioPacket{AudioBuffer: AudioBuffer{sample}}, 3)
+	}
+	stream := (<-events).C
+	for _, want := range []int16{2, 3, 4} {
+		packet := <-stream
+		if got := packet.AudioBuffer[0]; got != want {
+			t.Fatalf("queued sample = %d, want %d", got, want)
+		}
+	}
+}
+
+func TestAudioListenerDetachDuringDelivery(t *testing.T) {
+	var listeners AudioListeners
+	detacher := listeners.Attach(testAudioListener{})
+	user := &User{Session: 13}
+	start := make(chan struct{})
+	var workers sync.WaitGroup
+
+	workers.Add(1)
+	go func() {
+		defer workers.Done()
+		<-start
+		for i := 0; i < 1000; i++ {
+			listeners.dispatchAudio(nil, user, &AudioPacket{}, 3)
+		}
+	}()
+	close(start)
+	detacher.Detach()
+	workers.Wait()
+}
+
+func TestAudioPacketLength(t *testing.T) {
+	audioLength, final := parseAudioPacketLength(42 | audioTerminatorFlag)
+	if audioLength != 42 {
+		t.Fatalf("audio length = %d, want 42", audioLength)
+	}
+	if !final {
+		t.Fatal("expected terminator flag to mark packet final")
+	}
+
+	audioLength, final = parseAudioPacketLength(42)
+	if audioLength != 42 || final {
+		t.Fatalf("non-final packet parsed as length %d, final %v", audioLength, final)
+	}
+}
+
+func TestUserAudioSourceLifecycle(t *testing.T) {
+	user := &User{}
+	source := openal.Source(1)
+	otherSource := openal.Source(2)
+
+	user.SetAudioSource(&source)
+	if !user.HasAudioSource() {
+		t.Fatal("expected the audio source to be published")
+	}
+	user.ClearAudioSource(otherSource)
+	if !user.HasAudioSource() {
+		t.Fatal("expected a different source not to clear the active source")
+	}
+	user.ClearAudioSource(source)
+	if user.HasAudioSource() {
+		t.Fatal("expected the active audio source to be cleared")
+	}
+}
+
+func TestLocalMuteGenerationChangesOnlyOnTransitions(t *testing.T) {
+	user := &User{}
+	muted, generation := user.LocalMuteState()
+	if muted || generation != 0 {
+		t.Fatalf("initial local mute state = (%v, %d), want (false, 0)", muted, generation)
+	}
+
+	if got := user.SetLocallyMuted(true); got != 1 {
+		t.Fatalf("mute generation = %d, want 1", got)
+	}
+	if got := user.SetLocallyMuted(true); got != 1 {
+		t.Fatalf("unchanged mute generation = %d, want 1", got)
+	}
+	if got := user.SetLocallyMuted(false); got != 2 {
+		t.Fatalf("unmute generation = %d, want 2", got)
+	}
+}

gumble/gumble/audiolisteners.go

@@ -1,38 +1,83 @@
 package gumble
 
+import "sync"
+
 type audioEventItem struct {
 	parent     *AudioListeners
 	prev, next *audioEventItem
 	listener   AudioListener
 	streams    map[*User]chan *AudioPacket
+	attached   bool
 }
 
 func (e *audioEventItem) Detach() {
+	parent := e.parent
+	parent.mu.Lock()
+	defer parent.mu.Unlock()
+
+	if !e.attached {
+		return
+	}
+	e.closeAudioStreamsLocked()
 	if e.prev == nil {
-		e.parent.head = e.next
+		parent.head = e.next
 	} else {
 		e.prev.next = e.next
 	}
 	if e.next == nil {
-		e.parent.tail = e.prev
+		parent.tail = e.prev
 	} else {
 		e.next.prev = e.prev
 	}
+	e.attached = false
+	e.prev = nil
+	e.next = nil
+}
+
+func (e *audioEventItem) closeUserAudioLocked(user *User) {
+	if stream := e.streams[user]; stream != nil {
+		for {
+			select {
+			case <-stream:
+				continue
+			default:
+				close(stream)
+				delete(e.streams, user)
+				return
+			}
+		}
+	}
+}
+
+func (e *audioEventItem) closeAudioStreamsLocked() {
+	for user := range e.streams {
+		e.closeUserAudioLocked(user)
+	}
+}
+
+type audioStreamCallback struct {
+	listener AudioListener
+	event    AudioStreamEvent
 }
 
 // AudioListeners is a list of audio listeners. Each attached listener is
 // called in sequence when a new user audio stream begins.
 type AudioListeners struct {
+	mu         sync.Mutex
 	head, tail *audioEventItem
 }
 
 // Attach adds a new audio listener to the end of the current list of listeners.
 func (e *AudioListeners) Attach(listener AudioListener) Detacher {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
 	item := &audioEventItem{
 		parent:   e,
 		prev:     e.tail,
 		listener: listener,
 		streams:  make(map[*User]chan *AudioPacket),
+		attached: true,
 	}
 	if e.head == nil {
 		e.head = item
@@ -41,6 +86,73 @@ func (e *AudioListeners) Attach(listener AudioListener) Detacher {
 		e.tail = item
 	} else {
 		e.tail.next = item
+		e.tail = item
 	}
 	return item
 }
+
+func (e *AudioListeners) dispatchAudio(client *Client, user *User, packet *AudioPacket, bufferCount int) {
+	if bufferCount < 1 {
+		bufferCount = 1
+	}
+	e.mu.Lock()
+	callbacks := make([]audioStreamCallback, 0)
+	for item := e.head; item != nil; item = item.next {
+		stream := item.streams[user]
+		if stream == nil {
+			stream = make(chan *AudioPacket, bufferCount)
+			item.streams[user] = stream
+			callbacks = append(callbacks, audioStreamCallback{
+				listener: item.listener,
+				event: AudioStreamEvent{
+					Client: client,
+					User:   user,
+					C:      stream,
+				},
+			})
+		}
+		enqueueLatestAudio(stream, packet)
+	}
+	e.mu.Unlock()
+
+	for i := range callbacks {
+		callbacks[i].listener.OnAudioStream(&callbacks[i].event)
+	}
+}
+
+// enqueueLatestAudio keeps packet delivery bounded. If playback falls behind,
+// discard the oldest queued packet so the network reader never waits on audio.
+func enqueueLatestAudio(stream chan *AudioPacket, packet *AudioPacket) {
+	select {
+	case stream <- packet:
+		return
+	default:
+	}
+
+	select {
+	case <-stream:
+	default:
+	}
+	select {
+	case stream <- packet:
+	default:
+	}
+}
+
+func (e *AudioListeners) closeUserAudio(user *User) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	for item := e.head; item != nil; item = item.next {
+		item.closeUserAudioLocked(user)
+	}
+}
+
+func (e *AudioListeners) closeAllAudio() {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	for item := e.head; item != nil; item = item.next {
+		item.closeAudioStreamsLocked()
+	}
+}

gumble/gumble/client.go

@@ -245,6 +245,7 @@ func (c *Client) readRoutine() {
 	wasSynced := c.State() == StateSynced
 	atomic.StoreUint32(&c.state, uint32(StateDisconnected))
 	close(c.end)
+	c.Config.AudioListeners.closeAllAudio()
 	if wasSynced {
 		c.Config.Listeners.onDisconnect(&c.disconnectEvent)
 	}

gumble/gumble/config.go

@@ -28,9 +28,15 @@ type Config struct {
 	// The event listeners used when client events are triggered.
 	Listeners      Listeners
 	AudioListeners AudioListeners
-	Buffers        int
+	// Buffers is the per-user capacity used between network decoding and audio
+	// playback, and for the OpenAL playback buffer pool.
+	Buffers int
 }
 
+// AudioMinimumBufferCount is the smallest pool that can provide a short
+// playback prebuffer without dropping the current packet.
+const AudioMinimumBufferCount = 3
+
 // NewConfig returns a new Config struct with default values set.
 func NewConfig() *Config {
 	return &Config{
@@ -40,6 +46,14 @@ func NewConfig() *Config {
 	}
 }
 
+// AudioBufferCount returns a safe per-user audio buffer count.
+func (c *Config) AudioBufferCount() int {
+	if c == nil || c.Buffers < AudioMinimumBufferCount {
+		return AudioMinimumBufferCount
+	}
+	return c.Buffers
+}
+
 // Attach is an alias of c.Listeners.Attach.
 func (c *Config) Attach(l EventListener) Detacher {
 	return c.Listeners.Attach(l)

gumble/gumble/handlers.go

@@ -22,6 +22,12 @@ var (
 	errNoCodec              = errors.New("gumble: no audio codec")
 )
 
+const audioTerminatorFlag int64 = 0x2000
+
+func parseAudioPacketLength(length int64) (int, bool) {
+	return int(length &^ audioTerminatorFlag), length&audioTerminatorFlag != 0
+}
+
 var handlers = [...]func(*Client, []byte) error{
 	(*Client).handleVersion,
 	(*Client).handleUDPTunnel,
@@ -127,19 +133,27 @@ func (c *Client) handleUDPTunnel(buffer []byte) error {
 	}
 	buffer = buffer[n:]
 	// Opus audio packets set the 13th bit in the size field as the terminator.
-	audioLength := int(length) &^ 0x2000
+	audioLength, final := parseAudioPacketLength(length)
 	if audioLength > len(buffer) {
 		return errInvalidProtobuf
 	}
 
-	pcm, err := decoder.Decode(buffer[:audioLength], AudioMaximumFrameSize)
-	if err != nil {
-		return err
+	var pcm []int16
+	if audioLength > 0 {
+		var err error
+		pcm, err = decoder.Decode(buffer[:audioLength], AudioMaximumFrameSize)
+		if err != nil {
+			return err
+		}
 	}
 
+	locallyMuted, localMuteGeneration := user.LocalMuteState()
 	event := AudioPacket{
-		Client: c,
-		Sender: user,
+		Client:              c,
+		Sender:              user,
+		Final:               final,
+		LocallyMuted:        locallyMuted,
+		LocalMuteGeneration: localMuteGeneration,
 		Target: &VoiceTarget{
 			ID: uint32(audioTarget),
 		},
@@ -156,24 +170,7 @@ func (c *Client) handleUDPTunnel(buffer []byte) error {
 		event.HasPosition = true
 	}
 
-	c.volatile.Lock()
-	for item := c.Config.AudioListeners.head; item != nil; item = item.next {
-		c.volatile.Unlock()
-		ch := item.streams[user]
-		if ch == nil {
-			ch = make(chan *AudioPacket)
-			item.streams[user] = ch
-			event := AudioStreamEvent{
-				Client: c,
-				User:   user,
-				C:      ch,
-			}
-			item.listener.OnAudioStream(&event)
-		}
-		ch <- &event
-		c.volatile.Lock()
-	}
-	c.volatile.Unlock()
+	c.Config.AudioListeners.dispatchAudio(c, user, &event, c.Config.AudioBufferCount())
 
 	return nil
 }
@@ -477,6 +474,7 @@ func (c *Client) handleUserRemove(buffer []byte) error {
 
 		c.volatile.Unlock()
 	}
+	c.Config.AudioListeners.closeUserAudio(event.User)
 
 	if c.State() == StateSynced {
 		c.Config.Listeners.onUserChange(&event)

gumble/gumble/user.go

@@ -1,6 +1,8 @@
 package gumble
 
 import (
+	"sync"
+
 	"git.stormux.org/storm/barnard/gumble/go-openal/openal"
 	"git.stormux.org/storm/barnard/gumble/gumble/MumbleProto"
 	"google.golang.org/protobuf/proto"
@@ -53,20 +55,85 @@ type User struct {
 	client  *Client
 	decoder AudioDecoder
 
-	AudioSource *openal.Source
-	Boost       uint16
-	Volume      float32
+	localMuteMu         sync.RWMutex
+	localMuteGeneration uint64
+	audioSourceMu       sync.Mutex
+	AudioSource         *openal.Source
+	Boost               uint16
+	Volume              float32
 }
 
 // IsMuted returns true if the user is muted either server-side or locally
 func (u *User) IsMuted() bool {
-	return u.Muted || u.LocallyMuted
+	return u.Muted || u.IsLocallyMuted()
 }
 
 func (u *User) GetClient() *Client {
 	return u.client
 }
 
+// IsLocallyMuted reports whether this client has locally muted the user.
+func (u *User) IsLocallyMuted() bool {
+	u.localMuteMu.RLock()
+	defer u.localMuteMu.RUnlock()
+	return u.LocallyMuted
+}
+
+// SetLocallyMuted updates the local mute state and advances its generation
+// when the state changes. The generation prevents queued audio from crossing
+// a mute or unmute boundary.
+func (u *User) SetLocallyMuted(muted bool) uint64 {
+	u.localMuteMu.Lock()
+	defer u.localMuteMu.Unlock()
+	if u.LocallyMuted != muted {
+		u.LocallyMuted = muted
+		u.localMuteGeneration++
+	}
+	return u.localMuteGeneration
+}
+
+// LocalMuteState returns the local mute state and its current generation.
+func (u *User) LocalMuteState() (bool, uint64) {
+	u.localMuteMu.RLock()
+	defer u.localMuteMu.RUnlock()
+	return u.LocallyMuted, u.localMuteGeneration
+}
+
+// SetAudioSource publishes the OpenAL source used for this user's playback.
+func (u *User) SetAudioSource(source *openal.Source) {
+	u.audioSourceMu.Lock()
+	defer u.audioSourceMu.Unlock()
+	u.AudioSource = source
+}
+
+// ClearAudioSource removes source if it is still this user's active source.
+func (u *User) ClearAudioSource(source openal.Source) {
+	u.audioSourceMu.Lock()
+	defer u.audioSourceMu.Unlock()
+	if u.AudioSource != nil && *u.AudioSource == source {
+		u.AudioSource = nil
+	}
+}
+
+// SetAudioGain changes the active playback source gain. It returns false when
+// the user does not currently have an active source.
+func (u *User) SetAudioGain(gain float32) bool {
+	u.audioSourceMu.Lock()
+	defer u.audioSourceMu.Unlock()
+	if u.AudioSource == nil {
+		return false
+	}
+	u.AudioSource.SetGain(gain)
+	return true
+}
+
+// HasAudioSource reports whether the user currently has an active source.
+func (u *User) HasAudioSource() bool {
+	u.audioSourceMu.Lock()
+	defer u.audioSourceMu.Unlock()
+	return u.AudioSource != nil
+}
+
 // SetTexture sets the user's texture.
 func (u *User) SetTexture(texture []byte) {
 	packet := MumbleProto.UserState{

gumble/gumbleopenal/stream.go

@@ -32,9 +32,49 @@ type Recorder interface {
 const recorderOutgoingSource uint32 = ^uint32(0)
 
 const (
-	maxBufferSize = 11520 // Max frame size (2880) * bytes per stereo sample (4)
+	maxBufferSize            = 11520 // Max frame size (2880) * bytes per stereo sample (4)
+	playbackPrebufferBuffers = gumble.AudioMinimumBufferCount
+	playbackBufferWait       = 20 * time.Millisecond
 )
 
+func shouldStartPlayback(state openal.State, queued int32, final bool) bool {
+	return state != openal.Playing && queued > 0 && (queued >= playbackPrebufferBuffers || final)
+}
+
+type playbackCommand struct {
+	gain       float32
+	muted      bool
+	generation uint64
+	reset      bool
+	applied    chan struct{}
+}
+
+type playbackControl struct {
+	commands chan playbackCommand
+	cancel   chan struct{}
+	done     chan struct{}
+}
+
+type playbackState struct {
+	gain       float32
+	muted      bool
+	generation uint64
+}
+
+func (s *playbackState) apply(command playbackCommand) bool {
+	s.gain = command.gain
+	if !command.reset {
+		return false
+	}
+	s.muted = command.muted
+	s.generation = command.generation
+	return true
+}
+
+func (s playbackState) accepts(packet *gumble.AudioPacket) bool {
+	return packet != nil && !s.muted && !packet.LocallyMuted && packet.LocalMuteGeneration == s.generation
+}
+
 var (
 	ErrState        = errors.New("gumbleopenal: invalid state")
 	ErrMic          = errors.New("gumbleopenal: microphone disconnected or misconfigured")
@@ -61,7 +101,10 @@ type Stream struct {
 	sourceChannels  int
 	sourceFrameSize int
 	micVolume       float32
-	sourceStop      chan bool
+	sourceStop      chan struct{}
+	sourceMu        sync.Mutex
+	sourceWG        sync.WaitGroup
+	destroyed       bool
 
 	deviceSink  *openal.Device
 	contextSink *openal.Context
@@ -73,6 +116,11 @@ type Stream struct {
 	filePlayer          FilePlayer
 	recorderMu          sync.RWMutex
 	recorder            Recorder
+	playbackMu          sync.Mutex
+	playbackWG          sync.WaitGroup
+	playbackStopped     bool
+	playbackControls    map[*gumble.User]*playbackControl
+	destroyOnce         sync.Once
 }
 
 func New(client *gumble.Client, inputDevice *string, outputDevice *string, test bool) (*Stream, error) {
@@ -165,49 +213,182 @@ func (s *Stream) getRecorder() Recorder {
 	return s.recorder
 }
 
-func (s *Stream) Destroy() {
-	if s.link != nil {
-		s.link.Detach()
+// HasUserAudio reports whether user currently has a playback stream.
+func (s *Stream) HasUserAudio(user *gumble.User) bool {
+	s.playbackMu.Lock()
+	defer s.playbackMu.Unlock()
+	return !s.playbackStopped && s.playbackControls[user] != nil
+}
+
+// SetUserGain applies gain on the playback goroutine that owns the OpenAL
+// source.
+func (s *Stream) SetUserGain(user *gumble.User, gain float32) bool {
+	muted, generation := user.LocalMuteState()
+	return s.sendPlaybackCommand(user, playbackCommand{
+		gain:       gain,
+		muted:      muted,
+		generation: generation,
+	})
+}
+
+// SetUserMuteState flushes queued audio and synchronizes the user's current
+// local mute state with the playback goroutine.
+func (s *Stream) SetUserMuteState(user *gumble.User) bool {
+	muted, generation := user.LocalMuteState()
+	return s.sendPlaybackCommand(user, playbackCommand{
+		gain:       user.Volume,
+		muted:      muted,
+		generation: generation,
+		reset:      true,
+	})
+}
+
+func (s *Stream) sendPlaybackCommand(user *gumble.User, command playbackCommand) bool {
+	s.playbackMu.Lock()
+	control := s.playbackControls[user]
+	stopped := s.playbackStopped
+	s.playbackMu.Unlock()
+	if stopped || control == nil {
+		return false
 	}
-	if s.deviceSource != nil {
-		s.StopSource()
-		s.deviceSource.CaptureCloseDevice()
-		s.deviceSource = nil
+
+	command.applied = make(chan struct{})
+	select {
+	case control.commands <- command:
+	case <-control.cancel:
+		return false
+	case <-control.done:
+		return false
 	}
-	if s.deviceSink != nil {
-		s.contextSink.Destroy()
-		s.deviceSink.CloseDevice()
-		s.contextSink = nil
-		s.deviceSink = nil
+
+	select {
+	case <-command.applied:
+		return true
+	case <-control.cancel:
+		select {
+		case <-command.applied:
+			return true
+		default:
+			return false
+		}
+	case <-control.done:
+		select {
+		case <-command.applied:
+			return true
+		default:
+			return false
+		}
 	}
 }
 
+func (s *Stream) Destroy() {
+	s.destroyOnce.Do(func() {
+		s.playbackMu.Lock()
+		s.playbackStopped = true
+		for _, control := range s.playbackControls {
+			close(control.cancel)
+		}
+		s.playbackMu.Unlock()
+		if s.link != nil {
+			s.link.Detach()
+			s.link = nil
+		}
+		s.playbackWG.Wait()
+
+		s.sourceMu.Lock()
+		s.destroyed = true
+		if s.sourceStop != nil {
+			_ = s.stopSourceLocked()
+		}
+		if s.deviceSource != nil {
+			s.deviceSource.CaptureCloseDevice()
+			s.deviceSource = nil
+		}
+		s.sourceMu.Unlock()
+
+		if s.contextSink != nil {
+			s.contextSink.Destroy()
+			s.contextSink = nil
+		}
+		if s.deviceSink != nil {
+			s.deviceSink.CloseDevice()
+			s.deviceSink = nil
+		}
+	})
+}
+
 func (s *Stream) StartSource(inputDevice *string) error {
+	s.sourceMu.Lock()
+	defer s.sourceMu.Unlock()
+
+	if s.destroyed {
+		return ErrState
+	}
 	if s.sourceStop != nil {
 		return ErrState
 	}
 	if s.deviceSource == nil {
 		return ErrMic
 	}
-	s.deviceSource.CaptureStart()
-	s.sourceStop = make(chan bool)
-	go s.sourceRoutine(inputDevice)
+	if err := s.configureSourceLocked(inputDevice); err != nil {
+		return err
+	}
+
+	stop := make(chan struct{})
+	device := s.deviceSource
+	frameSize := s.sourceFrameSize
+	sourceChannels := s.sourceChannels
+	s.sourceStop = stop
+	device.CaptureStart()
+	s.sourceWG.Add(1)
+	go func() {
+		defer s.sourceWG.Done()
+		s.sourceRoutine(device, stop, frameSize, sourceChannels)
+	}()
 	return nil
 }
 
 func (s *Stream) StopSource() error {
+	s.sourceMu.Lock()
+	defer s.sourceMu.Unlock()
+
+	return s.stopSourceLocked()
+}
+
+func (s *Stream) stopSourceLocked() error {
 	if s.deviceSource == nil {
 		return ErrMic
 	}
-	s.deviceSource.CaptureStop()
 	if s.sourceStop == nil {
 		return ErrState
 	}
 	close(s.sourceStop)
+	s.deviceSource.CaptureStop()
+	s.sourceWG.Wait()
 	s.sourceStop = nil
 	return nil
 }
 
+func (s *Stream) configureSourceLocked(inputDevice *string) error {
+	frameSize := s.client.Config.AudioFrameSize()
+	if frameSize == s.sourceFrameSize {
+		return nil
+	}
+
+	s.deviceSource.CaptureCloseDevice()
+	s.deviceSource = openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, s.sourceFormat, uint32(frameSize))
+	if s.deviceSource == nil && s.sourceFormat == openal.FormatStereo16 {
+		s.sourceFormat = openal.FormatMono16
+		s.sourceChannels = 1
+		s.deviceSource = openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, s.sourceFormat, uint32(frameSize))
+	}
+	if s.deviceSource == nil {
+		return ErrMic
+	}
+	s.sourceFrameSize = frameSize
+	return nil
+}
+
 func (s *Stream) GetMicVolume() float32 {
 	return s.micVolume
 }
@@ -229,22 +410,53 @@ func (s *Stream) SetMicVolume(change float32, relative bool) {
 }
 
 func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
+	control := &playbackControl{
+		commands: make(chan playbackCommand),
+		cancel:   make(chan struct{}),
+		done:     make(chan struct{}),
+	}
+	s.playbackMu.Lock()
+	if s.playbackStopped {
+		s.playbackMu.Unlock()
+		return
+	}
+	if s.playbackControls == nil {
+		s.playbackControls = make(map[*gumble.User]*playbackControl)
+	}
+	s.playbackControls[e.User] = control
+	s.playbackWG.Add(1)
+	s.playbackMu.Unlock()
+
 	go func(e *gumble.AudioStreamEvent) {
+		defer func() {
+			s.playbackMu.Lock()
+			if s.playbackControls[e.User] == control {
+				delete(s.playbackControls, e.User)
+			}
+			s.playbackMu.Unlock()
+			close(control.done)
+			s.playbackWG.Done()
+		}()
+
 		var source = openal.NewSource()
-		e.User.AudioSource = &source
 
 		// Set initial gain based on volume and mute state
-		if e.User.LocallyMuted {
-			e.User.AudioSource.SetGain(0)
+		locallyMuted, localMuteGeneration := e.User.LocalMuteState()
+		state := playbackState{
+			gain:       e.User.Volume,
+			muted:      locallyMuted,
+			generation: localMuteGeneration,
+		}
+		if state.muted {
+			source.SetGain(0)
 		} else {
-			e.User.AudioSource.SetGain(e.User.Volume)
+			source.SetGain(state.gain)
 		}
+		e.User.SetAudioSource(&source)
 
-		bufferCount := e.Client.Config.Buffers
-		if bufferCount < 64 {
-			bufferCount = 64
-		}
-		emptyBufs := openal.NewBuffers(bufferCount)
+		bufferCount := e.Client.Config.AudioBufferCount()
+		allBufs := openal.NewBuffers(bufferCount)
+		emptyBufs := append(openal.Buffers(nil), allBufs...)
 
 		reclaim := func() {
 			if n := source.BuffersProcessed(); n > 0 {
@@ -254,16 +466,97 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 			}
 		}
 
+		flushPlayback := func() {
+			source.Stop()
+			if n := source.BuffersQueued(); n > 0 {
+				flushedBufs := make(openal.Buffers, n)
+				source.UnqueueBuffers(flushedBufs)
+				emptyBufs = append(emptyBufs, flushedBufs...)
+			}
+		}
+		defer func() {
+			flushPlayback()
+			e.User.ClearAudioSource(source)
+			source.Delete()
+			allBufs.Delete()
+		}()
+
+		applyCommand := func(command playbackCommand) bool {
+			reset := state.apply(command)
+			if reset {
+				flushPlayback()
+			}
+			if state.muted {
+				source.SetGain(0)
+			} else {
+				source.SetGain(state.gain)
+			}
+			close(command.applied)
+			return reset
+		}
+
+		acquireBuffer := func() (openal.Buffer, bool, bool) {
+			deadline := time.NewTimer(playbackBufferWait)
+			defer deadline.Stop()
+			retry := time.NewTicker(time.Millisecond)
+			defer retry.Stop()
+
+			for {
+				reclaim()
+				if len(emptyBufs) > 0 {
+					last := len(emptyBufs) - 1
+					buffer := emptyBufs[last]
+					emptyBufs = emptyBufs[:last]
+					return buffer, true, false
+				}
+				if source.State() != openal.Playing && source.BuffersQueued() > 0 {
+					source.Play()
+				}
+
+				select {
+				case command := <-control.commands:
+					if applyCommand(command) {
+						return 0, false, false
+					}
+				case <-control.cancel:
+					return 0, false, true
+				case <-deadline.C:
+					return 0, false, false
+				case <-retry.C:
+					continue
+				}
+			}
+		}
+
 		var raw [maxBufferSize]byte
 
-		for packet := range e.C {
-			// Skip processing if user is locally muted
-			if e.User.LocallyMuted {
+		for {
+			var packet *gumble.AudioPacket
+			select {
+			case command := <-control.commands:
+				applyCommand(command)
+				continue
+			case <-control.cancel:
+				return
+			case incoming, ok := <-e.C:
+				if !ok {
+					return
+				}
+				packet = incoming
+			}
+
+			if !state.accepts(packet) {
 				continue
 			}
 
 			var boost uint16 = uint16(1)
 			samples := len(packet.AudioBuffer)
+			if samples == 0 {
+				if shouldStartPlayback(source.State(), source.BuffersQueued(), packet.Final) {
+					source.Play()
+				}
+				continue
+			}
 			if samples > cap(raw)/2 {
 				continue
 			}
@@ -354,51 +647,30 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 				recorder.RecordAudioFrame(e.User.Session, recordBuffer[:recordPtr])
 			}
 
-			reclaim()
-			if len(emptyBufs) == 0 {
+			buffer, ok, canceled := acquireBuffer()
+			if canceled {
+				return
+			}
+			if !ok {
 				continue
 			}
 
-			last := len(emptyBufs) - 1
-			buffer := emptyBufs[last]
-			emptyBufs = emptyBufs[:last]
-
 			buffer.SetData(format, raw[:rawPtr], gumble.AudioSampleRate)
 			source.QueueBuffer(buffer)
 
-			if source.State() != openal.Playing {
+			if shouldStartPlayback(source.State(), source.BuffersQueued(), packet.Final) {
 				source.Play()
 			}
 		}
-		reclaim()
-		emptyBufs.Delete()
-		source.Delete()
 	}(e)
 }
 
-func (s *Stream) sourceRoutine(inputDevice *string) {
+func (s *Stream) sourceRoutine(device *openal.CaptureDevice, stop <-chan struct{}, frameSize int, sourceChannels int) {
 	interval := s.client.Config.AudioInterval
-	frameSize := s.client.Config.AudioFrameSize()
-
-	if frameSize != s.sourceFrameSize {
-		s.deviceSource.CaptureCloseDevice()
-		s.sourceFrameSize = frameSize
-		s.deviceSource = openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, s.sourceFormat, uint32(s.sourceFrameSize))
-		if s.deviceSource == nil && s.sourceFormat == openal.FormatStereo16 {
-			s.sourceFormat = openal.FormatMono16
-			s.sourceChannels = 1
-			s.deviceSource = openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, s.sourceFormat, uint32(s.sourceFrameSize))
-		}
-	}
-	if s.deviceSource == nil {
-		return
-	}
 
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
-	stop := s.sourceStop
-
 	outgoing := s.client.AudioOutgoing()
 	defer close(outgoing)
 
@@ -407,12 +679,12 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 		case <-stop:
 			return
 		case <-ticker.C:
-			sampleCount := frameSize * s.sourceChannels
+			sampleCount := frameSize * sourceChannels
 			int16Buffer := make([]int16, sampleCount)
 
 			// Capture microphone if available
 			hasMicInput := false
-			buff := s.deviceSource.CaptureSamples(uint32(frameSize))
+			buff := device.CaptureSamples(uint32(frameSize))
 			if len(buff) == sampleCount*2 {
 				hasMicInput = true
 				for i := 0; i < sampleCount; i++ {
@@ -423,7 +695,7 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 					int16Buffer[i] = sample
 				}
 
-				if s.sourceChannels == 1 {
+				if sourceChannels == 1 {
 					s.processMonoSamples(int16Buffer)
 				} else {
 					s.processStereoSamples(int16Buffer, frameSize)
@@ -443,7 +715,7 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 					outputBuffer = make([]int16, frameSize*2)
 
 					if hasMicInput {
-						if s.sourceChannels == 2 {
+						if sourceChannels == 2 {
 							// Mix stereo mic with stereo file
 							for i := 0; i < frameSize; i++ {
 								idx := i * 2
@@ -498,13 +770,17 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 			// Determine what to send
 			if hasFileAudio {
 				// Send stereo buffer when file is playing
-				outgoing <- gumble.AudioBuffer(outputBuffer)
+				if !sendOutgoingAudio(stop, outgoing, gumble.AudioBuffer(outputBuffer)) {
+					return
+				}
 				if recorder := s.getRecorder(); recorder != nil {
 					recorder.RecordAudioFrame(recorderOutgoingSource, outputBuffer)
 				}
 			} else if hasMicInput {
 				// Send mic when no file is playing
-				outgoing <- gumble.AudioBuffer(int16Buffer)
+				if !sendOutgoingAudio(stop, outgoing, gumble.AudioBuffer(int16Buffer)) {
+					return
+				}
 				if recorder := s.getRecorder(); recorder != nil {
 					recorder.RecordAudioFrame(recorderOutgoingSource, int16Buffer)
 				}
@@ -513,6 +789,21 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 	}
 }
 
+func sendOutgoingAudio(stop <-chan struct{}, outgoing chan<- gumble.AudioBuffer, buffer gumble.AudioBuffer) bool {
+	select {
+	case <-stop:
+		return false
+	default:
+	}
+
+	select {
+	case <-stop:
+		return false
+	case outgoing <- buffer:
+		return true
+	}
+}
+
 func scaleForRecording(sample int16, volume float32) int16 {
 	scaled := int32(float32(sample) * volume)
 	if scaled > 32767 {

gumble/gumbleopenal/stream_test.go

@@ -0,0 +1,133 @@
+package gumbleopenal
+
+import (
+	"testing"
+
+	"git.stormux.org/storm/barnard/gumble/go-openal/openal"
+	"git.stormux.org/storm/barnard/gumble/gumble"
+)
+
+func TestShouldStartPlayback(t *testing.T) {
+	tests := []struct {
+		name   string
+		state  openal.State
+		queued int32
+		final  bool
+		want   bool
+	}{
+		{name: "prebuffer not full", state: openal.Initial, queued: playbackPrebufferBuffers - 1},
+		{name: "prebuffer full", state: openal.Initial, queued: playbackPrebufferBuffers, want: true},
+		{name: "recover after underrun", state: openal.Stopped, queued: playbackPrebufferBuffers, want: true},
+		{name: "short final utterance", state: openal.Initial, queued: 1, final: true, want: true},
+		{name: "empty final packet", state: openal.Initial, final: true},
+		{name: "already playing", state: openal.Playing, queued: playbackPrebufferBuffers, final: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := shouldStartPlayback(tt.state, tt.queued, tt.final); got != tt.want {
+				t.Fatalf("shouldStartPlayback() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSendOutgoingAudioStopsWhileSendIsBlocked(t *testing.T) {
+	stop := make(chan struct{})
+	outgoing := make(chan gumble.AudioBuffer)
+	close(stop)
+
+	if sendOutgoingAudio(stop, outgoing, gumble.AudioBuffer{1}) {
+		t.Fatal("expected a stopped source not to block on outgoing audio")
+	}
+}
+
+func TestSendOutgoingAudioDeliversWhenRunning(t *testing.T) {
+	stop := make(chan struct{})
+	outgoing := make(chan gumble.AudioBuffer, 1)
+	want := gumble.AudioBuffer{1, 2}
+
+	if !sendOutgoingAudio(stop, outgoing, want) {
+		t.Fatal("expected a running source to deliver outgoing audio")
+	}
+	got := <-outgoing
+	if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
+		t.Fatalf("outgoing audio = %v, want %v", got, want)
+	}
+}
+
+func TestPlaybackStateRejectsAudioAcrossMuteBoundary(t *testing.T) {
+	state := playbackState{gain: 1, generation: 0}
+	beforeMute := &gumble.AudioPacket{LocalMuteGeneration: 0}
+	if !state.accepts(beforeMute) {
+		t.Fatal("expected current-generation audio before mute to be accepted")
+	}
+
+	mute := playbackCommand{gain: 1, muted: true, generation: 1, reset: true}
+	if !state.apply(mute) {
+		t.Fatal("expected mute transition to request a playback flush")
+	}
+	duringMute := &gumble.AudioPacket{LocallyMuted: true, LocalMuteGeneration: 1}
+	if state.accepts(duringMute) {
+		t.Fatal("expected audio received while muted to be rejected")
+	}
+
+	unmute := playbackCommand{gain: 1, generation: 2, reset: true}
+	if !state.apply(unmute) {
+		t.Fatal("expected unmute transition to request a playback flush")
+	}
+	if state.accepts(beforeMute) {
+		t.Fatal("expected partial prebuffer audio from before mute to stay rejected after unmute")
+	}
+	if state.accepts(duringMute) {
+		t.Fatal("expected muted-period audio to stay rejected after unmute")
+	}
+	afterUnmute := &gumble.AudioPacket{LocalMuteGeneration: 2}
+	if !state.accepts(afterUnmute) {
+		t.Fatal("expected current-generation audio after unmute to be accepted")
+	}
+}
+
+func TestSetUserMuteStateWaitsForPlaybackFlush(t *testing.T) {
+	user := &gumble.User{Volume: 0.75}
+	user.SetLocallyMuted(true)
+	control := &playbackControl{
+		commands: make(chan playbackCommand),
+		cancel:   make(chan struct{}),
+		done:     make(chan struct{}),
+	}
+	stream := &Stream{
+		playbackControls: map[*gumble.User]*playbackControl{user: control},
+	}
+	commands := make(chan playbackCommand, 1)
+	go func() {
+		command := <-control.commands
+		commands <- command
+		close(command.applied)
+	}()
+
+	if !stream.SetUserMuteState(user) {
+		t.Fatal("expected mute state command to be applied")
+	}
+	command := <-commands
+	if !command.reset || !command.muted || command.generation != 1 || command.gain != user.Volume {
+		t.Fatalf("unexpected mute command: %#v", command)
+	}
+}
+
+func TestSetUserMuteStateStopsOnPlaybackCancellation(t *testing.T) {
+	user := &gumble.User{}
+	control := &playbackControl{
+		commands: make(chan playbackCommand),
+		cancel:   make(chan struct{}),
+		done:     make(chan struct{}),
+	}
+	stream := &Stream{
+		playbackControls: map[*gumble.User]*playbackControl{user: control},
+	}
+	close(control.cancel)
+
+	if stream.SetUserMuteState(user) {
+		t.Fatal("expected a canceled playback stream to reject the mute command")
+	}
+}