@@ -1,21 +1,16 @@
using Discord.API.Gateway;
using Discord.API.Rest;
using Discord.API.Rpc;
using Discord.Net;
using Discord.API.Rpc;
using Discord.Logging;
using Discord.Net.Converters;
using Discord.Net.Queue;
using Discord.Net.Rest;
using Discord.Net.WebSockets;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.IO.Compression;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
@@ -24,16 +19,46 @@ namespace Discord.API
{
public class DiscordRpcApiClient : IDisposable
{
private abstract class RpcRequest
{
public abstract Task SetResultAsync(JToken data, JsonSerializer serializer);
public abstract Task SetExceptionAsync(JToken data, JsonSerializer serializer);
}
private class RpcRequest<T> : RpcRequest
{
public TaskCompletionSource<T> Promise { get; set; }
public RpcRequest(RequestOptions options)
{
Promise = new TaskCompletionSource<T>();
Task.Run(async () =>
{
await Task.Delay(options?.Timeout ?? 15000).ConfigureAwait(false);
Promise.TrySetCanceled(); //Doesn't need to be async, we're already in a separate task
});
}
public override Task SetResultAsync(JToken data, JsonSerializer serializer)
{
return Promise.TrySetResultAsync(data.ToObject<T>(serializer));
}
public override Task SetExceptionAsync(JToken data, JsonSerializer serializer)
{
var error = data.ToObject<ErrorEvent>(serializer);
return Promise.TrySetExceptionAsync(new RpcException(error.Code, error.Message));
}
}
private object _eventLock = new object();
public event Func<string, Task> SentRpcMessage { add { _sentRpcMessageEvent.Add(value); } remove { _sentRpcMessageEvent.Remove(value); } }
private readonly AsyncEvent<Func<string, Task>> _sentRpcMessageEvent = new AsyncEvent<Func<string, Task>>();
public event Func<string, string, object, string, Task> ReceivedRpcEvent { add { _receivedRpcEvent.Add(value); } remove { _receivedRpcEvent.Remove(value); } }
private readonly AsyncEvent<Func<string, string, object, string, Task>> _receivedRpcEvent = new AsyncEvent<Func<string, string, object, string, Task>>();
public event Func<string, Optional<string>, Optional<object> , Task> ReceivedRpcEvent { add { _receivedRpcEvent.Add(value); } remove { _receivedRpcEvent.Remove(value); } }
private readonly AsyncEvent<Func<string, Optional<string>, Optional<object>, Task>> _receivedRpcEvent = new AsyncEvent<Func<string, Optional<string>, Optional<object> , 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 ConcurrentDictionary<Guid, RpcRequest> _requests;
private readonly RequestQueue _requestQueue;
private readonly JsonSerializer _serializer;
private readonly IWebSocketClient _webSocketClient;
@@ -41,22 +66,26 @@ namespace Discord.API
private readonly string _clientId;
private CancellationTokenSource _loginCancelToken, _connectCancelToken;
private string _authToken;
private string _origin;
private bool _isDisposed;
public LoginState LoginState { get; private set; }
public ConnectionState ConnectionState { get; private set; }
public DiscordRpcApiClient(string clientId, WebSocketProvider webSocketProvider, JsonSerializer serializer = null, RequestQueue requestQueue = null)
public DiscordRpcApiClient(string clientId, string origin, WebSocketProvider webSocketProvider, JsonSerializer serializer = null, RequestQueue requestQueue = null)
{
_connectionLock = new SemaphoreSlim(1, 1);
_clientId = clientId;
_origin = origin;
_requestQueue = requestQueue ?? new RequestQueue();
_requests = new ConcurrentDictionary<Guid, RpcRequest>();
if (webSocketProvider != null)
{
_webSocketClient = webSocketProvider();
//_gatewayClient.SetHeader("user-agent", DiscordConfig.UserAgent); (Causes issues in .Net 4.6+)
//_webSocketClient.SetHeader("user-agent", DiscordConfig.UserAgent); (Causes issues in .Net 4.6+)
_webSocketClient.SetHeader("origin", _origin);
_webSocketClient.BinaryMessage += async (data, index, count) =>
{
using (var compressed = new MemoryStream(data, index + 2, count - 2))
@@ -66,16 +95,25 @@ namespace Discord.API
zlib.CopyTo(decompressed);
decompressed.Position = 0;
using (var reader = new StreamReader(decompressed))
using (var jsonReader = new JsonTextReader(reader))
{
var msg = JsonConvert.DeserializeObject<RpcMessage>(reader.ReadToEnd());
await _receivedRpcEvent.InvokeAsync(msg.Cmd, msg.Event, msg.Data, msg.Nonce).ConfigureAwait(false);
var msg = _serializer.Deserialize<RpcMessage>(jsonReader);
await _receivedRpcEvent.InvokeAsync(msg.Cmd, msg.Event, msg.Data).ConfigureAwait(false);
if (msg.Nonce.IsSpecified && msg.Nonce.Value.HasValue)
ProcessMessage(msg);
}
}
};
_webSocketClient.TextMessage += async text =>
{
var msg = JsonConvert.DeserializeObject<RpcMessage>(text);
await _receivedRpcEvent.InvokeAsync(msg.Cmd, msg.Event, msg.Data, msg.Nonce).ConfigureAwait(false);
using (var reader = new StringReader(text))
using (var jsonReader = new JsonTextReader(reader))
{
var msg = _serializer.Deserialize<RpcMessage>(jsonReader);
await _receivedRpcEvent.InvokeAsync(msg.Cmd, msg.Event, msg.Data).ConfigureAwait(false);
if (msg.Nonce.IsSpecified && msg.Nonce.Value.HasValue)
ProcessMessage(msg);
}
};
_webSocketClient.Closed += async ex =>
{
@@ -99,19 +137,19 @@ namespace Discord.API
}
}
public void Dispose() => Dispose(true);
public async Task LoginAsync(TokenType tokenType, string token, RequestOptions options = null)
public async Task LoginAsync(TokenType tokenType, string token, bool upgrade = false, RequestOptions options = null)
{
await _connectionLock.WaitAsync().ConfigureAwait(false);
try
{
await LoginInternalAsync(tokenType, token, options).ConfigureAwait(false);
await LoginInternalAsync(tokenType, token, upgrade, options).ConfigureAwait(false);
}
finally { _connectionLock.Release(); }
}
private async Task LoginInternalAsync(TokenType tokenType, string token, RequestOptions options = null)
private async Task LoginInternalAsync(TokenType tokenType, string token, bool upgrade = false, RequestOptions options = null)
{
if (LoginState != LoginState.LoggedOut)
if (!upgrade && LoginState != LoginState.LoggedOut)
await LogoutInternalAsync().ConfigureAwait(false);
if (tokenType != TokenType.Bearer)
@@ -233,39 +271,155 @@ namespace Discord.API
}
//Core
public Task SendRpcAsync(string cmd, object payload, GlobalBucket bucket = GlobalBucket.GeneralRpc, RequestOptions options = null)
=> SendRpcAsyncInternal(cmd, payload, BucketGroup.Global, (int)bucket, 0, options);
public Task SendRpcAsync(string cmd, object payload, GuildBucket bucket, ulong guildId, RequestOptions options = null)
=> SendRpcAsyncInternal(cmd, payload, BucketGroup.Guild, (int)bucket, guildId, options);
private async Task SendRpcAsyncInternal(string cmd, object payload,
BucketGroup group, int bucketId, ulong guildId, RequestOptions options)
public Task<TResponse> SendRpcAsync<TResponse>(string cmd, object payload, GlobalBucket bucket = GlobalBucket.GeneralRpc,
Optional<string> evt = default(Optional<string>), RequestOptions options = null)
where TResponse : class
=> SendRpcAsyncInternal<TResponse>(cmd, payload, BucketGroup.Global, (int)bucket, 0, evt, options);
public Task<TResponse> SendRpcAsync<TResponse>(string cmd, object payload, GuildBucket bucket, ulong guildId,
Optional<string> evt = default(Optional<string>), RequestOptions options = null)
where TResponse : class
=> SendRpcAsyncInternal<TResponse>(cmd, payload, BucketGroup.Guild, (int)bucket, guildId, evt, options);
private async Task<TResponse> SendRpcAsyncInternal<TResponse>(string cmd, object payload, BucketGroup group, int bucketId, ulong guildId,
Optional<string> evt, RequestOptions options)
where TResponse : class
{
//TODO: Add Nonce to pair sent requests with responses
byte[] bytes = null;
payload = new RpcMessage { Cmd = cmd, Args = payload, Nonce = Guid.NewGuid().ToString() };
var guid = Guid.NewGuid();
payload = new RpcMessage { Cmd = cmd, Event = evt, Args = payload, Nonce = guid };
if (payload != null)
bytes = Encoding.UTF8.GetBytes(SerializeJson(payload));
{
var json = SerializeJson(payload);
bytes = Encoding.UTF8.GetBytes(json);
}
var requestTracker = new RpcRequest<TResponse>(options);
_requests[guid] = requestTracker;
await _requestQueue.SendAsync(new WebSocketRequest(_webSocketClient, bytes, true, options), group, bucketId, guildId).ConfigureAwait(false);
await _sentRpcMessageEvent.InvokeAsync(cmd).ConfigureAwait(false);
return await requestTracker.Promise.Task.ConfigureAwait(false);
}
//Rpc
public async Task SendAuthenticateAsync(RequestOptions options = null)
public async Task<AuthenticateResponse> SendAuthenticateAsync(RequestOptions options = null)
{
var msg = new AuthenticateParams()
{
AccessToken = _authToken
};
await SendRpcAsync("AUTHENTICATE", msg, options: options).ConfigureAwait(false);
return await SendRpcAsync<AuthenticateResponse> ("AUTHENTICATE", msg, options: options).ConfigureAwait(false);
}
public async Task SendAuthorizeAsync(string[] scopes, RequestOptions options = null)
public async Task<AuthorizeResponse> SendAuthorizeAsync(string[] scopes, RequestOptions options = null)
{
var msg = new AuthorizeParams()
{
ClientId = _clientId,
Scopes = scopes
};
await SendRpcAsync("AUTHORIZE", msg, options: options).ConfigureAwait(false);
if (options == null)
options = new RequestOptions();
if (options.Timeout == null)
options.Timeout = 60000; //This requires manual input on the user's end, lets give them more time
return await SendRpcAsync<AuthorizeResponse>("AUTHORIZE", msg, options: options).ConfigureAwait(false);
}
public async Task<GetGuildsResponse> SendGetGuildsAsync(RequestOptions options = null)
{
return await SendRpcAsync<GetGuildsResponse>("GET_GUILDS", null, options: options).ConfigureAwait(false);
}
public async Task<RpcGuild> SendGetGuildAsync(ulong guildId, RequestOptions options = null)
{
var msg = new GetGuildParams
{
GuildId = guildId
};
return await SendRpcAsync<RpcGuild>("GET_GUILD", msg, options: options).ConfigureAwait(false);
}
public async Task<GetChannelsResponse> SendGetChannelsAsync(ulong guildId, RequestOptions options = null)
{
var msg = new GetChannelsParams
{
GuildId = guildId
};
return await SendRpcAsync<GetChannelsResponse>("GET_CHANNELS", msg, options: options).ConfigureAwait(false);
}
public async Task<RpcChannel> SendGetChannelAsync(ulong channelId, RequestOptions options = null)
{
var msg = new GetChannelParams
{
ChannelId = channelId
};
return await SendRpcAsync<RpcChannel>("GET_CHANNEL", msg, options: options).ConfigureAwait(false);
}
public async Task<SetLocalVolumeResponse> SendSetLocalVolumeAsync(int volume, RequestOptions options = null)
{
var msg = new SetLocalVolumeParams
{
Volume = volume
};
return await SendRpcAsync<SetLocalVolumeResponse>("SET_LOCAL_VOLUME", msg, options: options).ConfigureAwait(false);
}
public async Task<RpcChannel> SendSelectVoiceChannelAsync(ulong channelId, RequestOptions options = null)
{
var msg = new SelectVoiceChannelParams
{
ChannelId = channelId
};
return await SendRpcAsync<RpcChannel>("SELECT_VOICE_CHANNEL", msg, options: options).ConfigureAwait(false);
}
public async Task<SubscriptionResponse> SendChannelSubscribeAsync(string evt, ulong channelId, RequestOptions options = null)
{
var msg = new ChannelSubscriptionParams
{
ChannelId = channelId
};
return await SendRpcAsync<SubscriptionResponse>("SUBSCRIBE", msg, evt: evt, options: options).ConfigureAwait(false);
}
public async Task<SubscriptionResponse> SendChannelUnsubscribeAsync(string evt, ulong channelId, RequestOptions options = null)
{
var msg = new ChannelSubscriptionParams
{
ChannelId = channelId
};
return await SendRpcAsync<SubscriptionResponse>("UNSUBSCRIBE", msg, evt: evt, options: options).ConfigureAwait(false);
}
public async Task<SubscriptionResponse> SendGuildSubscribeAsync(string evt, ulong guildId, RequestOptions options = null)
{
var msg = new GuildSubscriptionParams
{
GuildId = guildId
};
return await SendRpcAsync<SubscriptionResponse>("SUBSCRIBE", msg, evt: evt, options: options).ConfigureAwait(false);
}
public async Task<SubscriptionResponse> SendGuildUnsubscribeAsync(string evt, ulong guildId, RequestOptions options = null)
{
var msg = new GuildSubscriptionParams
{
GuildId = guildId
};
return await SendRpcAsync<SubscriptionResponse>("UNSUBSCRIBE", msg, evt: evt, options: options).ConfigureAwait(false);
}
private bool ProcessMessage(RpcMessage msg)
{
RpcRequest requestTracker;
if (_requests.TryGetValue(msg.Nonce.Value.Value, out requestTracker))
{
if (msg.Event.GetValueOrDefault("") == "ERROR")
{
var _ = requestTracker.SetExceptionAsync(msg.Data.GetValueOrDefault() as JToken, _serializer);
}
else
{
var _ = requestTracker.SetResultAsync(msg.Data.GetValueOrDefault() as JToken, _serializer);
}
return true;
}
else
return false;
}
//Helpers