diff --git a/README.md b/README.md index d786538..bcc0981 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,10 @@ Each event has the following parameters: * event: the name of the event - join: user has joined the channel you are in - leave: user has left the channel you are in + - mute: a user in your channel has muted themselves + - unmute: a user in your channel has unmuted themselves + - deafen: a user in your channel has deafened themselves + - undeafen: a user in your channel has undeafened themselves - micup: you have begun transmitting - micdown: you have stopped transmitting - connect: you have connected to a server diff --git a/barnard.go b/barnard.go index 9c4ded3..343628b 100644 --- a/barnard.go +++ b/barnard.go @@ -52,8 +52,9 @@ type Barnard struct { exitMessage string // Added for channel muting - MutedChannels map[uint32]bool - userChannels map[uint32]*gumble.Channel + MutedChannels map[uint32]bool + userChannels map[uint32]*gumble.Channel + userAudioStates map[uint32]userAudioState // Added for noise suppression NoiseSuppressor *noise.Suppressor @@ -115,18 +116,14 @@ func (b *Barnard) TreeItemKeyPress(ui *uiterm.Ui, tree *uiterm.Tree, item uiterm users := makeUsersArray(treeItem.Channel.Users) for _, u := range users { // Explicitly set user mute state to match channel state - if channelWillBeMuted && !u.LocallyMuted { + if channelWillBeMuted && !u.IsLocallyMuted() { b.UserConfig.ToggleMute(u) - } else if !channelWillBeMuted && u.LocallyMuted { + } else if !channelWillBeMuted && u.IsLocallyMuted() { b.UserConfig.ToggleMute(u) } - if u.AudioSource != nil { - if u.LocallyMuted { - u.AudioSource.SetGain(0) - } else { - u.AudioSource.SetGain(u.Volume) - } + if b.Stream != nil { + b.Stream.SetUserMuteState(u) } } @@ -159,12 +156,8 @@ func (b *Barnard) TreeItemKeyPress(ui *uiterm.Ui, tree *uiterm.Tree, item uiterm if key == *b.Hotkeys.MuteToggle { // Toggle mute for single user b.UserConfig.ToggleMute(treeItem.User) - if treeItem.User.AudioSource != nil { - if treeItem.User.LocallyMuted { - treeItem.User.AudioSource.SetGain(0) - } else { - treeItem.User.AudioSource.SetGain(treeItem.User.Volume) - } + if b.Stream != nil { + b.Stream.SetUserMuteState(treeItem.User) } b.RebuildUserChannelTreePreservingSelection() b.Ui.Refresh() diff --git a/client.go b/client.go index 1d5324d..3a47b26 100644 --- a/client.go +++ b/client.go @@ -78,6 +78,7 @@ func (b *Barnard) OnConnect(e *gumble.ConnectEvent) { // Reset muted channels state on connect b.MutedChannels = make(map[uint32]bool) b.userChannels = make(map[uint32]*gumble.Channel) + b.userAudioStates = make(map[uint32]userAudioState) b.RecordingMutex.Lock() b.recordingAllowed = nil b.recordingStarting = false @@ -90,6 +91,7 @@ func (b *Barnard) OnConnect(e *gumble.ConnectEvent) { for _, u := range b.Client.Users { b.UserConfig.UpdateUser(u) b.rememberUserChannel(u) + b.rememberUserAudioState(u) } b.UpdateInputStatus(fmt.Sprintf("[%s]", e.Client.Self.Channel.Name)) @@ -119,6 +121,17 @@ func (b *Barnard) OnDisconnect(e *gumble.DisconnectEvent) { reason = e.String } b.stopRecordingForDisconnect() + b.FileStreamMutex.Lock() + fileStream := b.FileStream + b.FileStream = nil + b.FileStreamMutex.Unlock() + if fileStream != nil && fileStream.IsPlaying() { + _ = fileStream.Stop() + } + if b.Stream != nil { + b.Stream.Destroy() + b.Stream = nil + } b.Notify("disconnect", "me", reason) if reason == "" { b.AddOutputLine("Disconnected") @@ -171,19 +184,30 @@ func (b *Barnard) OnTextMessage(e *gumble.TextMessageEvent) { func (b *Barnard) OnUserChange(e *gumble.UserChangeEvent) { notification, hasNotification := b.userChangeNotification(e) + audioNotification, hasAudioNotification := b.userAudioChangeNotification(e) if e.User != nil { + previousLocalMute, previousLocalMuteGeneration := e.User.LocalMuteState() + previousVolume := e.User.Volume b.UserConfig.UpdateUser(e.User) + localMute, localMuteGeneration := e.User.LocalMuteState() + if b.Stream != nil { + if localMute != previousLocalMute || localMuteGeneration != previousLocalMuteGeneration { + b.Stream.SetUserMuteState(e.User) + } else if e.User.Volume != previousVolume { + b.Stream.SetUserGain(e.User, e.User.Volume) + } + } // Check if user is joining a muted channel if e.Type.Has(gumble.UserChangeConnected) || e.Type.Has(gumble.UserChangeChannel) { // If the channel is muted, ensure the user is muted if b.MutedChannels[e.User.Channel.ID] { // Only mute if not already muted - if !e.User.LocallyMuted { + if !e.User.IsLocallyMuted() { b.UserConfig.ToggleMute(e.User) } - if e.User.AudioSource != nil { - e.User.AudioSource.SetGain(0) + if b.Stream != nil { + b.Stream.SetUserMuteState(e.User) } } } @@ -198,6 +222,10 @@ func (b *Barnard) OnUserChange(e *gumble.UserChangeEvent) { b.Notify(notification.event, notification.who, notification.what) b.AddOutputLine(notification.line) } + if hasAudioNotification { + b.Notify(audioNotification.event, audioNotification.who, audioNotification.what) + b.AddOutputLine(audioNotification.line) + } if e.Type.Has(gumble.UserChangeChannel) && e.User == b.Client.Self { b.UpdateInputStatus(fmt.Sprintf("[%s]", e.User.Channel.Name)) } @@ -215,6 +243,7 @@ func (b *Barnard) OnUserChange(e *gumble.UserChangeEvent) { b.AddOutputLine(formatUserStats(e.User)) } b.updateUserChannel(e) + b.updateUserAudioState(e) b.RebuildUserChannelTreePreservingSelection() b.Ui.Refresh() } @@ -226,6 +255,11 @@ type userChangeNotification struct { line string } +type userAudioState struct { + selfMuted bool + selfDeafened bool +} + func (b *Barnard) userChangeNotification(e *gumble.UserChangeEvent) (userChangeNotification, bool) { if e == nil || e.User == nil || b.Client == nil || b.Client.Self == nil || b.Client.Self.Channel == nil { return userChangeNotification{}, false @@ -256,6 +290,52 @@ func (b *Barnard) userChangeNotification(e *gumble.UserChangeEvent) (userChangeN return userChangeNotification{}, false } +func (b *Barnard) userAudioChangeNotification(e *gumble.UserChangeEvent) (userChangeNotification, bool) { + if e == nil || e.User == nil || !e.Type.Has(gumble.UserChangeAudio) || b.Client == nil || b.Client.Self == nil || b.Client.Self.Channel == nil { + return userChangeNotification{}, false + } + previous, known := b.userAudioStates[e.User.Session] + if !known || !sameChannel(e.User.Channel, b.Client.Self.Channel) { + return userChangeNotification{}, false + } + + event := "" + verb := "" + if previous.selfDeafened != e.User.SelfDeafened { + if e.User.SelfDeafened { + event = "deafen" + verb = "deafened" + } else { + event = "undeafen" + verb = "undeafened" + } + } else if previous.selfMuted != e.User.SelfMuted { + if e.User.SelfMuted { + event = "mute" + verb = "muted" + } else { + event = "unmute" + verb = "unmuted" + } + } + if event == "" { + return userChangeNotification{}, false + } + + who := e.User.Name + line := fmt.Sprintf("%s %s themselves", e.User.Name, verb) + if e.User.Session == b.Client.Self.Session { + who = "me" + line = fmt.Sprintf("You %s yourself", verb) + } + return userChangeNotification{ + event: event, + who: who, + what: e.User.Channel.Name, + line: line, + }, true +} + func buildUserChangeNotification(event string, verb string, user *gumble.User, eventChannel *gumble.Channel, currentChannel *gumble.Channel) (userChangeNotification, bool) { if !sameChannel(eventChannel, currentChannel) { return userChangeNotification{}, false @@ -302,6 +382,30 @@ func (b *Barnard) updateUserChannel(e *gumble.UserChangeEvent) { b.rememberUserChannel(e.User) } +func (b *Barnard) rememberUserAudioState(user *gumble.User) { + if user == nil { + return + } + if b.userAudioStates == nil { + b.userAudioStates = make(map[uint32]userAudioState) + } + b.userAudioStates[user.Session] = userAudioState{ + selfMuted: user.SelfMuted, + selfDeafened: user.SelfDeafened, + } +} + +func (b *Barnard) updateUserAudioState(e *gumble.UserChangeEvent) { + if e == nil || e.User == nil { + return + } + if e.Type.Has(gumble.UserChangeDisconnected) { + delete(b.userAudioStates, e.User.Session) + return + } + b.rememberUserAudioState(e.User) +} + func (b *Barnard) OnChannelChange(e *gumble.ChannelChangeEvent) { b.UpdateInputStatus(fmt.Sprintf("[%s]", e.Channel.Name)) if e.Type.Has(gumble.ChannelChangeDescription) { diff --git a/client_notification_test.go b/client_notification_test.go index 9eb0f74..4d87608 100644 --- a/client_notification_test.go +++ b/client_notification_test.go @@ -144,3 +144,145 @@ func TestUpdateUserChannel(t *testing.T) { t.Fatalf("expected disconnected user channel to be removed, got %#v", got) } } + +func TestUserAudioChangeNotification(t *testing.T) { + current := &gumble.Channel{ID: 1, Name: "Current"} + other := &gumble.Channel{ID: 2, Name: "Other"} + self := &gumble.User{Session: 1, Name: "Username", Channel: current} + + tests := []struct { + name string + user *gumble.User + previous userAudioState + remember bool + change gumble.UserChangeType + want userChangeNotification + wantOK bool + }{ + { + name: "user mutes themselves", + user: &gumble.User{Session: 2, Name: "Guest", Channel: current, SelfMuted: true}, + remember: true, + change: gumble.UserChangeAudio, + want: userChangeNotification{ + event: "mute", + who: "Guest", + what: "Current", + line: "Guest muted themselves", + }, + wantOK: true, + }, + { + name: "user unmutes themselves", + user: &gumble.User{Session: 2, Name: "Guest", Channel: current}, + previous: userAudioState{selfMuted: true}, + remember: true, + change: gumble.UserChangeAudio, + want: userChangeNotification{ + event: "unmute", + who: "Guest", + what: "Current", + line: "Guest unmuted themselves", + }, + wantOK: true, + }, + { + name: "deafen takes precedence over implied mute", + user: &gumble.User{Session: 2, Name: "Guest", Channel: current, SelfMuted: true, SelfDeafened: true}, + remember: true, + change: gumble.UserChangeAudio, + want: userChangeNotification{ + event: "deafen", + who: "Guest", + what: "Current", + line: "Guest deafened themselves", + }, + wantOK: true, + }, + { + name: "undeafen takes precedence over implied unmute", + user: &gumble.User{Session: 2, Name: "Guest", Channel: current}, + previous: userAudioState{selfMuted: true, selfDeafened: true}, + remember: true, + change: gumble.UserChangeAudio, + want: userChangeNotification{ + event: "undeafen", + who: "Guest", + what: "Current", + line: "Guest undeafened themselves", + }, + wantOK: true, + }, + { + name: "self mute uses self wording", + user: &gumble.User{Session: 1, Name: "Username", Channel: current, SelfMuted: true}, + remember: true, + change: gumble.UserChangeAudio, + want: userChangeNotification{ + event: "mute", + who: "me", + what: "Current", + line: "You muted yourself", + }, + wantOK: true, + }, + { + name: "other channel is ignored", + user: &gumble.User{Session: 2, Name: "Guest", Channel: other, SelfMuted: true}, + remember: true, + change: gumble.UserChangeAudio, + }, + { + name: "initial state is remembered without notification", + user: &gumble.User{Session: 2, Name: "Guest", Channel: current, SelfMuted: true}, + change: gumble.UserChangeConnected | gumble.UserChangeAudio, + }, + { + name: "unrelated user change is ignored", + user: &gumble.User{Session: 2, Name: "Guest", Channel: current, SelfMuted: true}, + remember: true, + change: gumble.UserChangeComment, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := &Barnard{ + Client: &gumble.Client{Self: self}, + } + if tt.remember { + b.userAudioStates = map[uint32]userAudioState{tt.user.Session: tt.previous} + } + + got, ok := b.userAudioChangeNotification(&gumble.UserChangeEvent{ + Client: b.Client, + Type: tt.change, + User: tt.user, + }) + if ok != tt.wantOK { + t.Fatalf("expected ok %v, got %v", tt.wantOK, ok) + } + if got != tt.want { + t.Fatalf("expected %#v, got %#v", tt.want, got) + } + }) + } +} + +func TestUpdateUserAudioState(t *testing.T) { + user := &gumble.User{Session: 2, SelfMuted: true, SelfDeafened: true} + b := &Barnard{} + + b.updateUserAudioState(&gumble.UserChangeEvent{User: user}) + if got := b.userAudioStates[user.Session]; got != (userAudioState{selfMuted: true, selfDeafened: true}) { + t.Fatalf("unexpected remembered audio state: %#v", got) + } + + b.updateUserAudioState(&gumble.UserChangeEvent{ + Type: gumble.UserChangeDisconnected, + User: user, + }) + if _, ok := b.userAudioStates[user.Session]; ok { + t.Fatal("expected disconnected user audio state to be removed") + } +} diff --git a/config/user_config.go b/config/user_config.go index ff026ae..5f2cc8e 100644 --- a/config/user_config.go +++ b/config/user_config.go @@ -253,7 +253,7 @@ func (c *Config) findUser(address string, username string) *eUser { func (c *Config) ToggleMute(u *gumble.User) { j := c.findUser(u.GetClient().Config.Address, u.Name) j.LocallyMuted = !j.LocallyMuted - u.LocallyMuted = j.LocallyMuted + u.SetLocallyMuted(j.LocallyMuted) c.SaveConfig() } @@ -331,7 +331,7 @@ func (c *Config) UpdateUser(u *gumble.User) { j = c.findUser(uc.Config.Address, u.Name) u.Boost = j.Boost u.Volume = j.Volume - u.LocallyMuted = j.LocallyMuted // Update LocallyMuted state from config + u.SetLocallyMuted(j.LocallyMuted) if u.Boost < 1 { u.Boost = 1 } @@ -343,7 +343,7 @@ func (c *Config) UpdateConfig(u *gumble.User) { j = c.findUser(u.GetClient().Config.Address, u.Name) j.Boost = u.Boost j.Volume = u.Volume - j.LocallyMuted = u.LocallyMuted // Save LocallyMuted state to config + j.LocallyMuted = u.IsLocallyMuted() } func NewConfig(fn *string) *Config { diff --git a/extras/barnard-sound.sh b/extras/barnard-sound.sh index 044167c..88f37f8 100755 --- a/extras/barnard-sound.sh +++ b/extras/barnard-sound.sh @@ -54,6 +54,38 @@ leave() { [[ $notify ]] && notify "$2 left the channel." } +mute() { + if [[ "$2" == "me" ]]; then + [[ $notify ]] && notify "You muted yourself." + else + [[ $notify ]] && notify "$2 muted themselves." + fi +} + +unmute() { + if [[ "$2" == "me" ]]; then + [[ $notify ]] && notify "You unmuted yourself." + else + [[ $notify ]] && notify "$2 unmuted themselves." + fi +} + +deafen() { + if [[ "$2" == "me" ]]; then + [[ $notify ]] && notify "You deafened yourself." + else + [[ $notify ]] && notify "$2 deafened themselves." + fi +} + +undeafen() { + if [[ "$2" == "me" ]]; then + [[ $notify ]] && notify "You undeafened yourself." + else + [[ $notify ]] && notify "$2 undeafened themselves." + fi +} + micdown() { [[ $sound ]] && play -qnV0 synth .25 sin G6:E5 norm -8 [[ $notify ]] && notify "You have stopped transmitting." diff --git a/gumble/gumble/audio.go b/gumble/gumble/audio.go index ceb10b2..59b090f 100644 --- a/gumble/gumble/audio.go +++ b/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 diff --git a/gumble/gumble/audio_buffer_test.go b/gumble/gumble/audio_buffer_test.go new file mode 100644 index 0000000..68cb826 --- /dev/null +++ b/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) + } +} diff --git a/gumble/gumble/audiolisteners.go b/gumble/gumble/audiolisteners.go index 7bf0e80..c9c2b10 100644 --- a/gumble/gumble/audiolisteners.go +++ b/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() + } +} diff --git a/gumble/gumble/client.go b/gumble/gumble/client.go index e77395b..3093032 100644 --- a/gumble/gumble/client.go +++ b/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) } diff --git a/gumble/gumble/config.go b/gumble/gumble/config.go index 7b3699f..780b7f4 100644 --- a/gumble/gumble/config.go +++ b/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) diff --git a/gumble/gumble/handlers.go b/gumble/gumble/handlers.go index a0278da..d2e91da 100644 --- a/gumble/gumble/handlers.go +++ b/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) diff --git a/gumble/gumble/user.go b/gumble/gumble/user.go index d23ffe2..0b120d3 100644 --- a/gumble/gumble/user.go +++ b/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{ diff --git a/gumble/gumbleopenal/stream.go b/gumble/gumbleopenal/stream.go index 819f35e..d0ab3d1 100644 --- a/gumble/gumbleopenal/stream.go +++ b/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 { diff --git a/gumble/gumbleopenal/stream_test.go b/gumble/gumbleopenal/stream_test.go new file mode 100644 index 0000000..7651aa8 --- /dev/null +++ b/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") + } +} diff --git a/main.go b/main.go index cd9637c..a6094fa 100644 --- a/main.go +++ b/main.go @@ -111,7 +111,7 @@ func main() { fifo := flag.String("fifo", "", "path of a FIFO from which to read commands") serverSet := false usernameSet := false - buffers := flag.Int("buffers", 16, "number of audio buffers to use") + buffers := flag.Int("buffers", 16, "per-user audio buffer count (minimum 3)") profile := flag.Bool("profile", false, "add http server to serve profiles") noiseSuppressionEnabled := flag.Bool("noise-suppression", false, "enable noise suppression for microphone input") diff --git a/ui_tree.go b/ui_tree.go index 6e60b3c..07647e0 100644 --- a/ui_tree.go +++ b/ui_tree.go @@ -9,7 +9,7 @@ import ( func (ti TreeItem) String() string { if ti.User != nil { - if ti.User.LocallyMuted { + if ti.User.IsLocallyMuted() { return "[MUTED] " + ti.User.Name } // Calculate total volume as percentage @@ -35,8 +35,7 @@ func (ti TreeItem) TreeItemStyle(fg, bg uiterm.Attribute, active bool) (uiterm.A func (b *Barnard) changeVolume(users []*gumble.User, change float32) { for _, u := range users { - au := u.AudioSource - if au == nil { + if b.Stream == nil || !b.Stream.HasUserAudio(u) { continue } var boost uint16 @@ -44,7 +43,7 @@ func (b *Barnard) changeVolume(users []*gumble.User, change float32) { var ng float32 var curboost float32 curboost = float32((u.Boost - 1)) / 10 - cv = au.GetGain() + curboost + cv = u.Volume + curboost ng = cv + change boost = uint16(1) if ng > 1.0 { @@ -58,9 +57,7 @@ func (b *Barnard) changeVolume(users []*gumble.User, change float32) { } u.Boost = boost u.Volume = ng - if !u.LocallyMuted { - au.SetGain(ng) - } + b.Stream.SetUserGain(u, ng) b.UserConfig.UpdateConfig(u) } b.UserConfig.SaveConfig() @@ -68,16 +65,13 @@ func (b *Barnard) changeVolume(users []*gumble.User, change float32) { func (b *Barnard) resetVolume(users []*gumble.User) { for _, u := range users { - au := u.AudioSource - if au == nil { + if b.Stream == nil || !b.Stream.HasUserAudio(u) { continue } // Reset to original volume (1.0) and boost (1) u.Boost = uint16(1) u.Volume = 1.0 - if !u.LocallyMuted { - au.SetGain(1.0) - } + b.Stream.SetUserGain(u, 1.0) b.UserConfig.UpdateConfig(u) } b.UserConfig.SaveConfig()