Started adding IAudioClient incoming stream creation/destruction events

src/Discord.Net.Core/Audio/IAudioClient.cs

@@ -8,6 +8,8 @@ namespace Discord.Audio
         event Func<Task> Connected;
         event Func<Exception, Task> Disconnected;
         event Func<int, int, Task> LatencyUpdated;
+        event Func<ulong, AudioInStream, Task> StreamCreated;
+        event Func<ulong, Task> StreamDestroyed;
 
         /// <summary> Gets the current connection state of this client. </summary>
         ConnectionState ConnectionState { get; }

src/Discord.Net.Core/Entities/Channels/IAudioChannel.cs

@@ -1,6 +1,12 @@
-namespace Discord
+using Discord.Audio;
+using System;
+using System.Threading.Tasks;
+
+namespace Discord
 {
     public interface IAudioChannel : IChannel
     {
+        /// <summary> Connects to this audio channel. </summary>
+        Task<IAudioClient> ConnectAsync(Action<IAudioClient> configAction = null);
     }
 }

src/Discord.Net.Core/Entities/Channels/IVoiceChannel.cs

@@ -1,5 +1,4 @@
-using Discord.Audio;
+using System;
-using System;
 using System.Threading.Tasks;
 
 namespace Discord
@@ -13,7 +12,5 @@ namespace Discord
 
         /// <summary> Modifies this voice channel. </summary>
         Task ModifyAsync(Action<VoiceChannelProperties> func, RequestOptions options = null);
-        /// <summary> Connects to this voice channel. </summary>
-        Task<IAudioClient> ConnectAsync();
     }
 }

src/Discord.Net.Rest/Entities/Channels/RestGroupChannel.cs

@@ -1,4 +1,5 @@
-using System;
+using Discord.Audio;
+using System;
 using System.Collections.Generic;
 using System.Collections.Immutable;
 using System.Diagnostics;
@@ -145,6 +146,9 @@ namespace Discord.Rest
         IDisposable IMessageChannel.EnterTypingState(RequestOptions options)
             => EnterTypingState(options);
 
+        //IAudioChannel
+        Task<IAudioClient> IAudioChannel.ConnectAsync(Action<IAudioClient> configAction) { throw new NotSupportedException(); }
+
         //IChannel        
         Task<IUser> IChannel.GetUserAsync(ulong id, CacheMode mode, RequestOptions options)
             => Task.FromResult<IUser>(GetUser(id));

src/Discord.Net.Rest/Entities/Channels/RestVoiceChannel.cs

@@ -40,8 +40,8 @@ namespace Discord.Rest
 
         private string DebuggerDisplay => $"{Name} ({Id}, Voice)";
 
-        //IVoiceChannel
+        //IAudioChannel
-        Task<IAudioClient> IVoiceChannel.ConnectAsync() { throw new NotSupportedException(); }
+        Task<IAudioClient> IAudioChannel.ConnectAsync(Action<IAudioClient> configAction) { throw new NotSupportedException(); }
 
         //IGuildChannel
         Task<IGuildUser> IGuildChannel.GetUserAsync(ulong id, CacheMode mode, RequestOptions options)

src/Discord.Net.Rpc/Entities/Channels/RpcGroupChannel.cs

@@ -1,4 +1,5 @@
-using Discord.Rest;
+using Discord.Audio;
+using Discord.Rest;
 using System;
 using System.Collections.Generic;
 using System.Collections.Immutable;
@@ -112,6 +113,9 @@ namespace Discord.Rpc
         IDisposable IMessageChannel.EnterTypingState(RequestOptions options)
             => EnterTypingState(options);
 
+        //IAudioChannel
+        Task<IAudioClient> IAudioChannel.ConnectAsync(Action<IAudioClient> configAction) { throw new NotSupportedException(); }
+
         //IChannel
         string IChannel.Name { get { throw new NotSupportedException(); } }
 

src/Discord.Net.Rpc/Entities/Channels/RpcVoiceChannel.cs

@@ -42,7 +42,7 @@ namespace Discord.Rpc
 
         private string DebuggerDisplay => $"{Name} ({Id}, Voice)";
 
-        //IVoiceChannel
+        //IAudioChannel
-        Task<IAudioClient> IVoiceChannel.ConnectAsync() { throw new NotSupportedException(); }
+        Task<IAudioClient> IAudioChannel.ConnectAsync(Action<IAudioClient> configAction) { throw new NotSupportedException(); }
     }
 }

src/Discord.Net.WebSocket/Audio/AudioClient.cs

@@ -47,6 +47,18 @@ namespace Discord.Audio
             remove { _latencyUpdatedEvent.Remove(value); }
         }
         private readonly AsyncEvent<Func<int, int, Task>> _latencyUpdatedEvent = new AsyncEvent<Func<int, int, Task>>();
+        public event Func<ulong, AudioInStream, Task> StreamCreated
+        {
+            add { _streamCreated.Add(value); }
+            remove { _streamCreated.Remove(value); }
+        }
+        private readonly AsyncEvent<Func<ulong, AudioInStream, Task>> _streamCreated = new AsyncEvent<Func<ulong, AudioInStream, Task>>();
+        public event Func<ulong, Task> StreamDestroyed
+        {
+            add { _streamDestroyed.Add(value); }
+            remove { _streamDestroyed.Remove(value); }
+        }
+        private readonly AsyncEvent<Func<ulong, Task>> _streamDestroyed = new AsyncEvent<Func<ulong, Task>>();
 
         private readonly Logger _audioLogger;
         private readonly JsonSerializer _serializer;
@@ -182,7 +194,7 @@ namespace Discord.Audio
                 throw new ArgumentException("Value must be 120, 240, 480, 960, 1920 or 2880", nameof(samplesPerFrame));
         }
 
-        internal void CreateInputStream(ulong userId)
+        internal async Task CreateInputStreamAsync(ulong userId)
         {
             //Assume Thread-safe
             if (!_streams.ContainsKey(userId))
@@ -190,6 +202,7 @@ namespace Discord.Audio
                 var readerStream = new InputStream();
                 var writerStream = new OpusDecodeStream(new RTPReadStream(readerStream, _secretKey));
                 _streams.TryAdd(userId, new StreamPair(readerStream, writerStream));
+                await _streamCreated.InvokeAsync(userId, readerStream);
             }
         }
         internal AudioInStream GetInputStream(ulong id)
@@ -199,14 +212,18 @@ namespace Discord.Audio
                 return streamPair.Reader;
             return null;
         }
-        internal void RemoveInputStream(ulong userId)
+        internal async Task RemoveInputStreamAsync(ulong userId)
         {
-            _streams.TryRemove(userId, out var ignored);
+            if (_streams.TryRemove(userId, out var ignored))
+                await _streamDestroyed.InvokeAsync(userId).ConfigureAwait(false);
         }
-        internal void ClearInputStreams()
+        internal async Task ClearInputStreamsAsync()
         {
-            foreach (var pair in _streams.Values)
+            foreach (var pair in _streams)
-                pair.Reader.Dispose();
+            {
+                pair.Value.Reader.Dispose();
+                await _streamDestroyed.InvokeAsync(pair.Key).ConfigureAwait(false);
+            }
             _ssrcMap.Clear();
             _streams.Clear();
         }

src/Discord.Net.WebSocket/DiscordSocketClient.cs

@@ -1414,7 +1414,7 @@ namespace Discord.WebSocket
                                             if (data.ChannelId != null)
                                             {
                                                 before = guild.GetVoiceState(data.UserId)?.Clone() ?? SocketVoiceState.Default;
-                                                after = guild.AddOrUpdateVoiceState(State, data);
+                                                after = await guild.AddOrUpdateVoiceStateAsync(State, data).ConfigureAwait(false);
                                                 /*if (data.UserId == CurrentUser.Id)
                                                 {
                                                     var _ = guild.FinishJoinAudioChannel().ConfigureAwait(false);
@@ -1471,7 +1471,7 @@ namespace Discord.WebSocket
                                     if (guild != null)
                                     {
                                         string endpoint = data.Endpoint.Substring(0, data.Endpoint.LastIndexOf(':'));
-                                        var _ = guild.FinishConnectAudio(_nextAudioId++, endpoint, data.Token).ConfigureAwait(false);
+                                        var _ = guild.FinishConnectAudio(endpoint, data.Token).ConfigureAwait(false);
                                     }
                                     else
                                     {
@@ -1725,6 +1725,8 @@ namespace Discord.WebSocket
             }
         }
 
+        internal int GetAudioId() => _nextAudioId++;
+
         //IDiscordClient
         async Task<IApplication> IDiscordClient.GetApplicationInfoAsync(RequestOptions options)
             => await GetApplicationInfoAsync().ConfigureAwait(false);

src/Discord.Net.WebSocket/Entities/Channels/ISocketAudioChannel.cs

@@ -5,6 +5,5 @@ namespace Discord.WebSocket
 {
     public interface ISocketAudioChannel : IAudioChannel
     {
-        Task<IAudioClient> ConnectAsync();
     }
 }

src/Discord.Net.WebSocket/Entities/Channels/SocketGroupChannel.cs

@@ -212,6 +212,9 @@ namespace Discord.WebSocket
         IDisposable IMessageChannel.EnterTypingState(RequestOptions options)
             => EnterTypingState(options);
 
+        //IAudioChannel
+        Task<IAudioClient> IAudioChannel.ConnectAsync(Action<IAudioClient> configAction) { throw new NotSupportedException(); }
+
         //IChannel        
         Task<IUser> IChannel.GetUserAsync(ulong id, CacheMode mode, RequestOptions options)
             => Task.FromResult<IUser>(GetUser(id));

src/Discord.Net.WebSocket/Entities/Channels/SocketVoiceChannel.cs

@@ -40,9 +40,9 @@ namespace Discord.WebSocket
         public Task ModifyAsync(Action<VoiceChannelProperties> func, RequestOptions options = null)
             => ChannelHelper.ModifyAsync(this, Discord, func, options);
 
-        public async Task<IAudioClient> ConnectAsync()
+        public async Task<IAudioClient> ConnectAsync(Action<IAudioClient> configAction = null)
         {
-            return await Guild.ConnectAudioAsync(Id, false, false).ConfigureAwait(false);
+            return await Guild.ConnectAudioAsync(Id, false, false, configAction).ConfigureAwait(false);
         }
 
         public override SocketGuildUser GetUser(ulong id)

src/Discord.Net.WebSocket/Entities/Guilds/SocketGuild.cs

@@ -423,7 +423,7 @@ namespace Discord.WebSocket
         }
 
         //Voice States
-        internal SocketVoiceState AddOrUpdateVoiceState(ClientState state, VoiceStateModel model)
+        internal async Task<SocketVoiceState> AddOrUpdateVoiceStateAsync(ClientState state, VoiceStateModel model)
         {
             var voiceChannel = state.GetChannel(model.ChannelId.Value) as SocketVoiceChannel;
             var before = GetVoiceState(model.UserId) ?? SocketVoiceState.Default;
@@ -433,12 +433,12 @@ namespace Discord.WebSocket
             if (_audioClient != null && before.VoiceChannel?.Id != after.VoiceChannel?.Id)
             {
                 if (model.UserId == CurrentUser.Id)
-                    RepopulateAudioStreams();
+                    await RepopulateAudioStreamsAsync().ConfigureAwait(false);
                 else
                 {
-                    _audioClient.RemoveInputStream(model.UserId); //User changed channels, end their stream
+                    await _audioClient.RemoveInputStreamAsync(model.UserId).ConfigureAwait(false); //User changed channels, end their stream
                     if (CurrentUser.VoiceChannel != null && after.VoiceChannel?.Id == CurrentUser.VoiceChannel?.Id)
-                        _audioClient.CreateInputStream(model.UserId);
+                        await _audioClient.CreateInputStreamAsync(model.UserId).ConfigureAwait(false);
                 }
             }
 
@@ -464,7 +464,7 @@ namespace Discord.WebSocket
         {
             return _audioClient?.GetInputStream(userId);
         }
-        internal async Task<IAudioClient> ConnectAudioAsync(ulong channelId, bool selfDeaf, bool selfMute)
+        internal async Task<IAudioClient> ConnectAudioAsync(ulong channelId, bool selfDeaf, bool selfMute, Action<IAudioClient> configAction)
         {
             selfDeaf = false;
             selfMute = false;
@@ -477,6 +477,32 @@ namespace Discord.WebSocket
                 await DisconnectAudioInternalAsync().ConfigureAwait(false);
                 promise = new TaskCompletionSource<AudioClient>();
                 _audioConnectPromise = promise;
+
+                if (_audioClient == null)
+                {
+                    var audioClient = new AudioClient(this, Discord.GetAudioId());
+                    audioClient.Disconnected += async ex =>
+                    {
+                        if (!promise.Task.IsCompleted)
+                        {
+                            try { audioClient.Dispose(); } catch { }
+                            _audioClient = null;
+                            if (ex != null)
+                                await promise.TrySetExceptionAsync(ex);
+                            else
+                                await promise.TrySetCanceledAsync();
+                            return;
+                        }
+                    };
+                    audioClient.Connected += () =>
+                    {
+                        var _ = promise.TrySetResultAsync(_audioClient);
+                        return Task.Delay(0);
+                    };
+                    configAction?.Invoke(audioClient);
+                    _audioClient = audioClient;
+                }
+
                 await Discord.ApiClient.SendVoiceStateUpdateAsync(Id, channelId, selfDeaf, selfMute).ConfigureAwait(false);
             }
             catch (Exception)
@@ -523,7 +549,7 @@ namespace Discord.WebSocket
                 await _audioClient.StopAsync().ConfigureAwait(false);
             _audioClient = null;
         }
-        internal async Task FinishConnectAudio(int id, string url, string token)
+        internal async Task FinishConnectAudio(string url, string token)
         {
             //TODO: Mem Leak: Disconnected/Connected handlers arent cleaned up
             var voiceState = GetVoiceState(Discord.CurrentUser.Id).Value;
@@ -531,31 +557,7 @@ namespace Discord.WebSocket
             await _audioLock.WaitAsync().ConfigureAwait(false);
             try
             {
-                var promise = _audioConnectPromise;
+                await RepopulateAudioStreamsAsync().ConfigureAwait(false);
-                if (_audioClient == null)
-                {
-                    var audioClient = new AudioClient(this, id);
-                    audioClient.Disconnected += async ex =>
-                    {
-                        if (!promise.Task.IsCompleted)
-                        {
-                            try { audioClient.Dispose(); } catch { }
-                            _audioClient = null;
-                            if (ex != null)
-                                await promise.TrySetExceptionAsync(ex);
-                            else
-                                await promise.TrySetCanceledAsync();
-                            return;
-                        }
-                    };
-                    _audioClient = audioClient;
-                    RepopulateAudioStreams();
-                }
-                _audioClient.Connected += () => 
-                { 
-                    var _ = promise.TrySetResultAsync(_audioClient); 
-                    return Task.Delay(0);
-                };
                 await _audioClient.StartAsync(url, Discord.CurrentUser.Id, voiceState.VoiceSessionId, token).ConfigureAwait(false);                
             }
             catch (OperationCanceledException)
@@ -573,15 +575,15 @@ namespace Discord.WebSocket
             }
         }
 
-        internal void RepopulateAudioStreams()
+        internal async Task RepopulateAudioStreamsAsync()
         {
-            _audioClient.ClearInputStreams(); //We changed channels, end all current streams
+            await _audioClient.ClearInputStreamsAsync().ConfigureAwait(false); //We changed channels, end all current streams
             if (CurrentUser.VoiceChannel != null)
             {
                 foreach (var pair in _voiceStates)
                 {
-                    if (pair.Value.VoiceChannel?.Id == CurrentUser.VoiceChannel?.Id)
+                    if (pair.Value.VoiceChannel?.Id == CurrentUser.VoiceChannel?.Id && pair.Key != CurrentUser.Id)
-                        _audioClient.CreateInputStream(pair.Key);
+                        await _audioClient.CreateInputStreamAsync(pair.Key).ConfigureAwait(false);
                 }
             }
         }