github/Discord.Net
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature: Implement Dispose for types which have disposable data (#1171)

Browse Source 
* Initial set of dispose implementations

Not handled yet:
- Discord.Net.Websocket/Entities/SocketGuild
- Discord.Net.Tests

* Refactor DiscordSocketClient init into ctor

This way we remove an IDisposableAnalyzer warning for not disposing
the client when we set the client variable.

* Dispose of clients when disposing sharded client

* Finish implementing IDisposable where appropriate

I opted to use NoWarn in the Tests project as it wasn't really necessary
considering that our tests only run once

* Tweak samples after feedback
This commit is contained in:
Monica S
2018-11-29 01:18:16 +00:00
committed by Christopher F
parent dca6c33da3
commit 7366cd4361
31 changed files with 406 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/Discord.Net.WebSocket/Audio/AudioClient.cs
View File

@@ -71,7 +71,7 @@ namespace Discord.Audio
ApiClient.ReceivedPacket += ProcessPacketAsync;
_stateLock = new SemaphoreSlim(1, 1);
_connection = new ConnectionManager(_stateLock, _audioLogger, 30000,
_connection = new ConnectionManager(_stateLock, _audioLogger, 30000,
OnConnectingAsync, OnDisconnectingAsync, x => ApiClient.Disconnected += x);
_connection.Connected += () => _connectedEvent.InvokeAsync();
_connection.Disconnected += (ex, recon) => _disconnectedEvent.InvokeAsync(ex);
@@ -79,7 +79,7 @@ namespace Discord.Audio
_keepaliveTimes = new ConcurrentQueue<KeyValuePair<ulong, int>>();
_ssrcMap = new ConcurrentDictionary<uint, ulong>();
_streams = new ConcurrentDictionary<ulong, StreamPair>();
_serializer = new JsonSerializer { ContractResolver = new DiscordContractResolver() };
_serializer.Error += (s, e) =>
{
@@ -91,7 +91,7 @@ namespace Discord.Audio
UdpLatencyUpdated += async (old, val) => await _audioLogger.DebugAsync($"UDP Latency = {val} ms").ConfigureAwait(false);
}
internal async Task StartAsync(string url, ulong userId, string sessionId, string token)
internal async Task StartAsync(string url, ulong userId, string sessionId, string token)
{
_url = url;
_userId = userId;
@@ -100,7 +100,7 @@ namespace Discord.Audio
await _connection.StartAsync().ConfigureAwait(false);
}
public async Task StopAsync()
{
{
await _connection.StopAsync().ConfigureAwait(false);
}
@@ -225,11 +225,11 @@ namespace Discord.Audio
if (!data.Modes.Contains(DiscordVoiceAPIClient.Mode))
throw new InvalidOperationException($"Discord does not support {DiscordVoiceAPIClient.Mode}");
ApiClient.SetUdpEndpoint(data.Ip, data.Port);
await ApiClient.SendDiscoveryAsync(_ssrc).ConfigureAwait(false);
_heartbeatTask = RunHeartbeatAsync(41250, _connection.CancelToken);
}
break;
@@ -305,9 +305,9 @@ namespace Discord.Audio
catch (Exception ex)
{
await _audioLogger.DebugAsync("Malformed Packet", ex).ConfigureAwait(false);
return;
return;
}
await _audioLogger.DebugAsync("Received Discovery").ConfigureAwait(false);
await ApiClient.SendSelectProtocol(ip, port).ConfigureAwait(false);
}
@@ -317,7 +317,7 @@ namespace Discord.Audio
{
await _audioLogger.DebugAsync("Received Keepalive").ConfigureAwait(false);
ulong value =
ulong value =
((ulong)packet[0] >> 0) |
((ulong)packet[1] >> 8) |
((ulong)packet[2] >> 16) |
@@ -341,7 +341,7 @@ namespace Discord.Audio
}
}
else
{
{
if (!RTPReadStream.TryReadSsrc(packet, 0, out var ssrc))
{
await _audioLogger.DebugAsync("Malformed Frame").ConfigureAwait(false);
@@ -388,7 +388,7 @@ namespace Discord.Audio
var now = Environment.TickCount;
//Did server respond to our last heartbeat?
if (_heartbeatTimes.Count != 0 && (now - _lastMessageTime) > intervalMillis &&
if (_heartbeatTimes.Count != 0 && (now - _lastMessageTime) > intervalMillis &&
ConnectionState == ConnectionState.Connected)
{
_connection.Error(new Exception("Server missed last heartbeat"));
@@ -437,7 +437,7 @@ namespace Discord.Audio
{
await _audioLogger.WarningAsync("Failed to send keepalive", ex).ConfigureAwait(false);
}
await Task.Delay(intervalMillis, cancelToken).ConfigureAwait(false);
}
await _audioLogger.DebugAsync("Keepalive Stopped").ConfigureAwait(false);
@@ -467,6 +467,7 @@ namespace Discord.Audio
{
StopAsync().GetAwaiter().GetResult();
ApiClient.Dispose();
_stateLock?.Dispose();
}
}
/// <inheritdoc />

26
src/Discord.Net.WebSocket/Audio/Streams/BufferedWriteStream.cs
View File

@@ -27,7 +27,7 @@ namespace Discord.Audio.Streams
private readonly AudioClient _client;
private readonly AudioStream _next;
private readonly CancellationTokenSource _cancelTokenSource;
private readonly CancellationTokenSource _disposeTokenSource, _cancelTokenSource;
private readonly CancellationToken _cancelToken;
private readonly Task _task;
private readonly ConcurrentQueue<Frame> _queuedFrames;
@@ -49,12 +49,13 @@ namespace Discord.Audio.Streams
_logger = logger;
_queueLength = (bufferMillis + (_ticksPerFrame - 1)) / _ticksPerFrame; //Round up
_cancelTokenSource = new CancellationTokenSource();
_cancelToken = CancellationTokenSource.CreateLinkedTokenSource(_cancelTokenSource.Token, cancelToken).Token;
_disposeTokenSource = new CancellationTokenSource();
_cancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_disposeTokenSource.Token, cancelToken);
_cancelToken = _cancelTokenSource.Token;
_queuedFrames = new ConcurrentQueue<Frame>();
_bufferPool = new ConcurrentQueue<byte[]>();
for (int i = 0; i < _queueLength; i++)
_bufferPool.Enqueue(new byte[maxFrameSize]);
_bufferPool.Enqueue(new byte[maxFrameSize]);
_queueLock = new SemaphoreSlim(_queueLength, _queueLength);
_silenceFrames = MaxSilenceFrames;
@@ -63,7 +64,12 @@ namespace Discord.Audio.Streams
protected override void Dispose(bool disposing)
{
if (disposing)
_cancelTokenSource.Cancel();
{
_disposeTokenSource?.Cancel();
_disposeTokenSource?.Dispose();
_cancelTokenSource?.Dispose();
_queueLock?.Dispose();
}
base.Dispose(disposing);
}
@@ -131,8 +137,12 @@ namespace Discord.Audio.Streams
public override void WriteHeader(ushort seq, uint timestamp, bool missed) { } //Ignore, we use our own timing
public override async Task WriteAsync(byte[] data, int offset, int count, CancellationToken cancelToken)
{
CancellationTokenSource writeCancelToken = null;
if (cancelToken.CanBeCanceled)
cancelToken = CancellationTokenSource.CreateLinkedTokenSource(cancelToken, _cancelToken).Token;
{
writeCancelToken = CancellationTokenSource.CreateLinkedTokenSource(cancelToken, _cancelToken);
cancelToken = writeCancelToken.Token;
}
else
cancelToken = _cancelToken;
@@ -142,6 +152,9 @@ namespace Discord.Audio.Streams
#if DEBUG
var _ = _logger?.DebugAsync("Buffer overflow"); //Should never happen because of the queueLock
#endif
#pragma warning disable IDISP016
writeCancelToken?.Dispose();
#pragma warning restore IDISP016
return;
}
Buffer.BlockCopy(data, offset, buffer, 0, count);
@@ -153,6 +166,7 @@ namespace Discord.Audio.Streams
#endif
_isPreloaded = true;
}
writeCancelToken?.Dispose();
}
public override async Task FlushAsync(CancellationToken cancelToken)

12
src/Discord.Net.WebSocket/Audio/Streams/InputStream.cs
View File

@@ -96,7 +96,17 @@ namespace Discord.Audio.Streams
protected override void Dispose(bool isDisposing)
{
_isDisposed = true;
if (!_isDisposed)
{
if (isDisposing)
{
_signal?.Dispose();
}
_isDisposed = true;
}
base.Dispose(isDisposing);
}
}
}

43
src/Discord.Net.WebSocket/ConnectionManager.cs
View File

@@ -6,7 +6,7 @@ using Discord.Net;
namespace Discord
{
internal class ConnectionManager
internal class ConnectionManager : IDisposable
{
public event Func<Task> Connected { add { _connectedEvent.Add(value); } remove { _connectedEvent.Remove(value); } }
private readonly AsyncEvent<Func<Task>> _connectedEvent = new AsyncEvent<Func<Task>>();
@@ -23,10 +23,12 @@ namespace Discord
private CancellationTokenSource _combinedCancelToken, _reconnectCancelToken, _connectionCancelToken;
private Task _task;
private bool _isDisposed;
public ConnectionState State { get; private set; }
public CancellationToken CancelToken { get; private set; }
internal ConnectionManager(SemaphoreSlim stateLock, Logger logger, int connectionTimeout,
internal ConnectionManager(SemaphoreSlim stateLock, Logger logger, int connectionTimeout,
Func<Task> onConnecting, Func<Exception, Task> onDisconnecting, Action<Func<Exception, Task>> clientDisconnectHandler)
{
_stateLock = stateLock;
@@ -55,6 +57,7 @@ namespace Discord
{
await AcquireConnectionLock().ConfigureAwait(false);
var reconnectCancelToken = new CancellationTokenSource();
_reconnectCancelToken?.Dispose();
_reconnectCancelToken = reconnectCancelToken;
_task = Task.Run(async () =>
{
@@ -67,16 +70,16 @@ namespace Discord
try
{
await ConnectAsync(reconnectCancelToken).ConfigureAwait(false);
nextReconnectDelay = 1000; //Reset delay
nextReconnectDelay = 1000; //Reset delay
await _connectionPromise.Task.ConfigureAwait(false);
}
catch (OperationCanceledException ex)
{
catch (OperationCanceledException ex)
{
Cancel(); //In case this exception didn't come from another Error call
await DisconnectAsync(ex, !reconnectCancelToken.IsCancellationRequested).ConfigureAwait(false);
}
catch (Exception ex)
{
catch (Exception ex)
{
Error(ex); //In case this exception didn't come from another Error call
if (!reconnectCancelToken.IsCancellationRequested)
{
@@ -113,6 +116,8 @@ namespace Discord
private async Task ConnectAsync(CancellationTokenSource reconnectCancelToken)
{
_connectionCancelToken?.Dispose();
_combinedCancelToken?.Dispose();
_connectionCancelToken = new CancellationTokenSource();
_combinedCancelToken = CancellationTokenSource.CreateLinkedTokenSource(_connectionCancelToken.Token, reconnectCancelToken.Token);
CancelToken = _combinedCancelToken.Token;
@@ -120,7 +125,7 @@ namespace Discord
_connectionPromise = new TaskCompletionSource<bool>();
State = ConnectionState.Connecting;
await _logger.InfoAsync("Connecting").ConfigureAwait(false);
try
{
var readyPromise = new TaskCompletionSource<bool>();
@@ -206,5 +211,25 @@ namespace Discord
break;
}
}
protected virtual void Dispose(bool disposing)
{
if (!_isDisposed)
{
if (disposing)
{
_combinedCancelToken?.Dispose();
_reconnectCancelToken?.Dispose();
_connectionCancelToken?.Dispose();
}
_isDisposed = true;
}
}
public void Dispose()
{
Dispose(true);
}
}
}
}

35
src/Discord.Net.WebSocket/DiscordShardedClient.cs
View File

@@ -18,7 +18,9 @@ namespace Discord.WebSocket
private int[] _shardIds;
private DiscordSocketClient[] _shards;
private int _totalShards;
private bool _isDisposed;
/// <inheritdoc />
public override int Latency { get => GetLatency(); protected set { } }
/// <inheritdoc />
@@ -38,11 +40,15 @@ namespace Discord.WebSocket
/// <summary> Creates a new REST/WebSocket Discord client. </summary>
public DiscordShardedClient() : this(null, new DiscordSocketConfig()) { }
/// <summary> Creates a new REST/WebSocket Discord client. </summary>
#pragma warning disable IDISP004
public DiscordShardedClient(DiscordSocketConfig config) : this(null, config, CreateApiClient(config)) { }
#pragma warning restore IDISP004
/// <summary> Creates a new REST/WebSocket Discord client. </summary>
public DiscordShardedClient(int[] ids) : this(ids, new DiscordSocketConfig()) { }
/// <summary> Creates a new REST/WebSocket Discord client. </summary>
#pragma warning disable IDISP004
public DiscordShardedClient(int[] ids, DiscordSocketConfig config) : this(ids, config, CreateApiClient(config)) { }
#pragma warning restore IDISP004
private DiscordShardedClient(int[] ids, DiscordSocketConfig config, API.DiscordSocketApiClient client)
: base(config, client)
{
@@ -119,10 +125,10 @@ namespace Discord.WebSocket
}
/// <inheritdoc />
public override async Task StartAsync()
public override async Task StartAsync()
=> await Task.WhenAll(_shards.Select(x => x.StartAsync())).ConfigureAwait(false);
/// <inheritdoc />
public override async Task StopAsync()
public override async Task StopAsync()
=> await Task.WhenAll(_shards.Select(x => x.StopAsync())).ConfigureAwait(false);
public DiscordSocketClient GetShard(int id)
@@ -145,7 +151,7 @@ namespace Discord.WebSocket
=> await _shards[0].GetApplicationInfoAsync(options).ConfigureAwait(false);
/// <inheritdoc />
public override SocketGuild GetGuild(ulong id)
public override SocketGuild GetGuild(ulong id)
=> GetShardFor(id).GetGuild(id);
/// <inheritdoc />
@@ -173,7 +179,7 @@ namespace Discord.WebSocket
for (int i = 0; i < _shards.Length; i++)
result += _shards[i].PrivateChannels.Count;
return result;
}
}
private IEnumerable<SocketGuild> GetGuilds()
{
@@ -189,7 +195,7 @@ namespace Discord.WebSocket
for (int i = 0; i < _shards.Length; i++)
result += _shards[i].Guilds.Count;
return result;
}
}
/// <inheritdoc />
public override SocketUser GetUser(ulong id)
@@ -369,5 +375,22 @@ namespace Discord.WebSocket
/// <inheritdoc />
Task<IVoiceRegion> IDiscordClient.GetVoiceRegionAsync(string id, RequestOptions options)
=> Task.FromResult<IVoiceRegion>(GetVoiceRegion(id));
internal override void Dispose(bool disposing)
{
if (!_isDisposed)
{
if (disposing)
{
foreach (var client in _shards)
client?.Dispose();
_connectionGroupLock?.Dispose();
}
_isDisposed = true;
}
base.Dispose(disposing);
}
}
}

5
src/Discord.Net.WebSocket/DiscordSocketApiClient.cs
View File

@@ -108,6 +108,8 @@ namespace Discord.API
}
_isDisposed = true;
}
base.Dispose(disposing);
}
public async Task ConnectAsync()
@@ -137,6 +139,7 @@ namespace Discord.API
ConnectionState = ConnectionState.Connecting;
try
{
_connectCancelToken?.Dispose();
_connectCancelToken = new CancellationTokenSource();
if (WebSocketClient != null)
WebSocketClient.SetCancelToken(_connectCancelToken.Token);
@@ -209,7 +212,7 @@ namespace Discord.API
await RequestQueue.SendAsync(new WebSocketRequest(WebSocketClient, null, bytes, true, options)).ConfigureAwait(false);
await _sentGatewayMessageEvent.InvokeAsync(opCode).ConfigureAwait(false);
}
public async Task SendIdentifyAsync(int largeThreshold = 100, int shardID = 0, int totalShards = 1, RequestOptions options = null)
{
options = RequestOptions.CreateOrClone(options);

24
src/Discord.Net.WebSocket/DiscordSocketClient.cs
View File

@@ -43,6 +43,8 @@ namespace Discord.WebSocket
private DateTimeOffset? _statusSince;
private RestApplication _applicationInfo;
private bool _isDisposed;
/// <summary> Gets the shard of of this client. </summary>
public int ShardId { get; }
/// <summary> Gets the current connection state of this client. </summary>
@@ -63,7 +65,7 @@ namespace Discord.WebSocket
internal WebSocketProvider WebSocketProvider { get; private set; }
internal bool AlwaysDownloadUsers { get; private set; }
internal int? HandlerTimeout { get; private set; }
internal new DiscordSocketApiClient ApiClient => base.ApiClient as DiscordSocketApiClient;
/// <inheritdoc />
public override IReadOnlyCollection<SocketGuild> Guilds => State.Guilds;
@@ -110,8 +112,10 @@ namespace Discord.WebSocket
/// Initializes a new REST/WebSocket-based Discord client with the provided configuration.
/// </summary>
/// <param name="config">The configuration to be used with the client.</param>
#pragma warning disable IDISP004
public DiscordSocketClient(DiscordSocketConfig config) : this(config, CreateApiClient(config), null, null) { }
internal DiscordSocketClient(DiscordSocketConfig config, SemaphoreSlim groupLock, DiscordSocketClient parentClient) : this(config, CreateApiClient(config), groupLock, parentClient) { }
#pragma warning restore IDISP004
private DiscordSocketClient(DiscordSocketConfig config, API.DiscordSocketApiClient client, SemaphoreSlim groupLock, DiscordSocketClient parentClient)
: base(config, client)
{
@@ -170,11 +174,18 @@ namespace Discord.WebSocket
/// <inheritdoc />
internal override void Dispose(bool disposing)
{
if (disposing)
if (!_isDisposed)
{
StopAsync().GetAwaiter().GetResult();
ApiClient.Dispose();
if (disposing)
{
StopAsync().GetAwaiter().GetResult();
ApiClient?.Dispose();
_stateLock?.Dispose();
}
_isDisposed = true;
}
base.Dispose(disposing);
}
/// <inheritdoc />
@@ -197,10 +208,10 @@ namespace Discord.WebSocket
}
/// <inheritdoc />
public override async Task StartAsync()
public override async Task StartAsync()
=> await _connection.StartAsync().ConfigureAwait(false);
/// <inheritdoc />
public override async Task StopAsync()
public override async Task StopAsync()
=> await _connection.StopAsync().ConfigureAwait(false);
private async Task OnConnectingAsync()
@@ -704,6 +715,7 @@ namespace Discord.WebSocket
{
await GuildUnavailableAsync(guild).ConfigureAwait(false);
await TimedInvokeAsync(_leftGuildEvent, nameof(LeftGuild), guild).ConfigureAwait(false);
(guild as IDisposable).Dispose();
}
else
{

14
src/Discord.Net.WebSocket/DiscordVoiceApiClient.cs
View File

@@ -16,7 +16,7 @@ using System.Threading.Tasks;
namespace Discord.Audio
{
internal class DiscordVoiceAPIClient
internal class DiscordVoiceAPIClient : IDisposable
{
public const int MaxBitrate = 128 * 1024;
public const string Mode = "xsalsa20_poly1305";
@@ -36,7 +36,7 @@ namespace Discord.Audio
private readonly AsyncEvent<Func<byte[], Task>> _receivedPacketEvent = new AsyncEvent<Func<byte[], Task>>();
public event Func<Exception, Task> Disconnected { add { _disconnectedEvent.Add(value); } remove { _disconnectedEvent.Remove(value); } }
private readonly AsyncEvent<Func<Exception, Task>> _disconnectedEvent = new AsyncEvent<Func<Exception, Task>>();
private readonly JsonSerializer _serializer;
private readonly SemaphoreSlim _connectionLock;
private readonly IUdpSocket _udp;
@@ -103,8 +103,9 @@ namespace Discord.Audio
if (disposing)
{
_connectCancelToken?.Dispose();
(_udp as IDisposable)?.Dispose();
(WebSocketClient as IDisposable)?.Dispose();
_udp?.Dispose();
WebSocketClient?.Dispose();
_connectionLock?.Dispose();
}
_isDisposed = true;
}
@@ -122,7 +123,7 @@ namespace Discord.Audio
}
public async Task SendAsync(byte[] data, int offset, int bytes)
{
await _udp.SendAsync(data, offset, bytes).ConfigureAwait(false);
await _udp.SendAsync(data, offset, bytes).ConfigureAwait(false);
await _sentDataEvent.InvokeAsync(bytes).ConfigureAwait(false);
}
@@ -177,6 +178,7 @@ namespace Discord.Audio
ConnectionState = ConnectionState.Connecting;
try
{
_connectCancelToken?.Dispose();
_connectCancelToken = new CancellationTokenSource();
var cancelToken = _connectCancelToken.Token;
@@ -208,7 +210,7 @@ namespace Discord.Audio
{
if (ConnectionState == ConnectionState.Disconnected) return;
ConnectionState = ConnectionState.Disconnecting;
try { _connectCancelToken?.Cancel(false); }
catch { }

20
src/Discord.Net.WebSocket/Entities/Guilds/SocketGuild.cs
View File

@@ -25,8 +25,9 @@ namespace Discord.WebSocket
/// Represents a WebSocket-based guild object.
/// </summary>
[DebuggerDisplay(@"{DebuggerDisplay,nq}")]
public class SocketGuild : SocketEntity<ulong>, IGuild
public class SocketGuild : SocketEntity<ulong>, IGuild, IDisposable
{
#pragma warning disable IDISP002, IDISP006
private readonly SemaphoreSlim _audioLock;
private TaskCompletionSource<bool> _syncPromise, _downloaderPromise;
private TaskCompletionSource<AudioClient> _audioConnectPromise;
@@ -37,6 +38,7 @@ namespace Discord.WebSocket
private ImmutableArray<GuildEmote> _emotes;
private ImmutableArray<string> _features;
private AudioClient _audioClient;
#pragma warning restore IDISP002, IDISP006
/// <inheritdoc />
public string Name { get; private set; }
@@ -63,7 +65,7 @@ namespace Discord.WebSocket
/// number here is the most accurate in terms of counting the number of users within this guild.
/// </para>
/// <para>
/// Use this instead of enumerating the count of the
/// Use this instead of enumerating the count of the
/// <see cref="Discord.WebSocket.SocketGuild.Users" /> collection, as you may see discrepancy
/// between that and this property.
/// </para>
@@ -872,9 +874,11 @@ namespace Discord.WebSocket
if (external)
{
#pragma warning disable IDISP001
var _ = promise.TrySetResultAsync(null);
await Discord.ApiClient.SendVoiceStateUpdateAsync(Id, channelId, selfDeaf, selfMute).ConfigureAwait(false);
return null;
#pragma warning restore IDISP001
}
if (_audioClient == null)
@@ -897,10 +901,14 @@ namespace Discord.WebSocket
};
audioClient.Connected += () =>
{
#pragma warning disable IDISP001
var _ = promise.TrySetResultAsync(_audioClient);
#pragma warning restore IDISP001
return Task.Delay(0);
};
#pragma warning disable IDISP003
_audioClient = audioClient;
#pragma warning restore IDISP003
}
await Discord.ApiClient.SendVoiceStateUpdateAsync(Id, channelId, selfDeaf, selfMute).ConfigureAwait(false);
@@ -948,6 +956,7 @@ namespace Discord.WebSocket
if (_audioClient != null)
await _audioClient.StopAsync().ConfigureAwait(false);
await Discord.ApiClient.SendVoiceStateUpdateAsync(Id, null, false, false).ConfigureAwait(false);
_audioClient?.Dispose();
_audioClient = null;
}
internal async Task FinishConnectAudio(string url, string token)
@@ -1130,5 +1139,12 @@ namespace Discord.WebSocket
/// <inheritdoc />
async Task<IReadOnlyCollection<IWebhook>> IGuild.GetWebhooksAsync(RequestOptions options)
=> await GetWebhooksAsync(options).ConfigureAwait(false);
void IDisposable.Dispose()
{
DisconnectAudioAsync().GetAwaiter().GetResult();
_audioLock?.Dispose();
_audioClient?.Dispose();
}
}
}

27
src/Discord.Net.WebSocket/Net/DefaultUdpSocket.cs
View File

@@ -13,24 +13,29 @@ namespace Discord.Net.Udp
private readonly SemaphoreSlim _lock;
private UdpClient _udp;
private IPEndPoint _destination;
private CancellationTokenSource _cancelTokenSource;
private CancellationTokenSource _stopCancelTokenSource, _cancelTokenSource;
private CancellationToken _cancelToken, _parentToken;
private Task _task;
private bool _isDisposed;
public ushort Port => (ushort)((_udp?.Client.LocalEndPoint as IPEndPoint)?.Port ?? 0);
public DefaultUdpSocket()
{
_lock = new SemaphoreSlim(1, 1);
_cancelTokenSource = new CancellationTokenSource();
_stopCancelTokenSource = new CancellationTokenSource();
}
private void Dispose(bool disposing)
{
if (!_isDisposed)
{
if (disposing)
{
StopInternalAsync(true).GetAwaiter().GetResult();
_stopCancelTokenSource?.Dispose();
_cancelTokenSource?.Dispose();
_lock?.Dispose();
}
_isDisposed = true;
}
}
@@ -56,9 +61,14 @@ namespace Discord.Net.Udp
{
await StopInternalAsync().ConfigureAwait(false);
_cancelTokenSource = new CancellationTokenSource();
_cancelToken = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _cancelTokenSource.Token).Token;
_stopCancelTokenSource?.Dispose();
_cancelTokenSource?.Dispose();
_stopCancelTokenSource = new CancellationTokenSource();
_cancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _stopCancelTokenSource.Token);
_cancelToken = _cancelTokenSource.Token;
_udp?.Dispose();
_udp = new UdpClient(0);
_task = RunAsync(_cancelToken);
@@ -77,7 +87,7 @@ namespace Discord.Net.Udp
}
public async Task StopInternalAsync(bool isDisposing = false)
{
try { _cancelTokenSource.Cancel(false); } catch { }
try { _stopCancelTokenSource.Cancel(false); } catch { }
if (!isDisposing)
await (_task ?? Task.Delay(0)).ConfigureAwait(false);
@@ -96,8 +106,11 @@ namespace Discord.Net.Udp
}
public void SetCancelToken(CancellationToken cancelToken)
{
_cancelTokenSource?.Dispose();
_parentToken = cancelToken;
_cancelToken = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _cancelTokenSource.Token).Token;
_cancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _stopCancelTokenSource.Token);
_cancelToken = _cancelTokenSource.Token;
}
public async Task SendAsync(byte[] data, int index, int count)

35
src/Discord.Net.WebSocket/Net/DefaultWebSocketClient.cs
View File

@@ -25,14 +25,14 @@ namespace Discord.Net.WebSockets
private readonly IWebProxy _proxy;
private ClientWebSocket _client;
private Task _task;
private CancellationTokenSource _cancelTokenSource;
private CancellationTokenSource _disconnectTokenSource, _cancelTokenSource;
private CancellationToken _cancelToken, _parentToken;
private bool _isDisposed, _isDisconnecting;
public DefaultWebSocketClient(IWebProxy proxy = null)
{
_lock = new SemaphoreSlim(1, 1);
_cancelTokenSource = new CancellationTokenSource();
_disconnectTokenSource = new CancellationTokenSource();
_cancelToken = CancellationToken.None;
_parentToken = CancellationToken.None;
_headers = new Dictionary<string, string>();
@@ -43,7 +43,12 @@ namespace Discord.Net.WebSockets
if (!_isDisposed)
{
if (disposing)
{
DisconnectInternalAsync(true).GetAwaiter().GetResult();
_disconnectTokenSource?.Dispose();
_cancelTokenSource?.Dispose();
_lock?.Dispose();
}
_isDisposed = true;
}
}
@@ -68,9 +73,14 @@ namespace Discord.Net.WebSockets
{
await DisconnectInternalAsync().ConfigureAwait(false);
_cancelTokenSource = new CancellationTokenSource();
_cancelToken = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _cancelTokenSource.Token).Token;
_disconnectTokenSource?.Dispose();
_cancelTokenSource?.Dispose();
_disconnectTokenSource = new CancellationTokenSource();
_cancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _disconnectTokenSource.Token);
_cancelToken = _cancelTokenSource.Token;
_client?.Dispose();
_client = new ClientWebSocket();
_client.Options.Proxy = _proxy;
_client.Options.KeepAliveInterval = TimeSpan.Zero;
@@ -98,7 +108,7 @@ namespace Discord.Net.WebSockets
}
private async Task DisconnectInternalAsync(bool isDisposing = false)
{
try { _cancelTokenSource.Cancel(false); } catch { }
try { _disconnectTokenSource.Cancel(false); } catch { }
_isDisconnecting = true;
try
@@ -117,7 +127,7 @@ namespace Discord.Net.WebSockets
}
try { _client.Dispose(); }
catch { }
_client = null;
}
}
@@ -144,8 +154,11 @@ namespace Discord.Net.WebSockets
}
public void SetCancelToken(CancellationToken cancelToken)
{
_cancelTokenSource?.Dispose();
_parentToken = cancelToken;
_cancelToken = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _cancelTokenSource.Token).Token;
_cancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_parentToken, _disconnectTokenSource.Token);
_cancelToken = _cancelTokenSource.Token;
}
public async Task SendAsync(byte[] data, int index, int count, bool isText)
@@ -166,7 +179,7 @@ namespace Discord.Net.WebSockets
frameSize = count - (i * SendChunkSize);
else
frameSize = SendChunkSize;
var type = isText ? WebSocketMessageType.Text : WebSocketMessageType.Binary;
await _client.SendAsync(new ArraySegment<byte>(data, index, count), type, isLast, _cancelToken).ConfigureAwait(false);
}
@@ -176,7 +189,7 @@ namespace Discord.Net.WebSockets
_lock.Release();
}
}
private async Task RunAsync(CancellationToken cancelToken)
{
var buffer = new ArraySegment<byte>(new byte[ReceiveChunkSize]);
@@ -188,7 +201,7 @@ namespace Discord.Net.WebSockets
WebSocketReceiveResult socketResult = await _client.ReceiveAsync(buffer, cancelToken).ConfigureAwait(false);
byte[] result;
int resultCount;
if (socketResult.MessageType == WebSocketMessageType.Close)
throw new WebSocketClosedException((int)socketResult.CloseStatus, socketResult.CloseStatusDescription);
@@ -219,7 +232,7 @@ namespace Discord.Net.WebSockets
resultCount = socketResult.Count;
result = buffer.Array;
}
if (socketResult.MessageType == WebSocketMessageType.Text)
{
string text = Encoding.UTF8.GetString(result, 0, resultCount);