Added support for compressed websocket messages.

This commit is contained in:
RogueException
2015-10-22 05:02:12 -03:00
parent 69de854db6
commit 0f31e18592
5 changed files with 57 additions and 15 deletions

View File

@@ -56,7 +56,11 @@ namespace Discord.API
public int Version = 3; public int Version = 3;
[JsonProperty("properties")] [JsonProperty("properties")]
public Dictionary<string, string> Properties = new Dictionary<string, string>(); public Dictionary<string, string> Properties = new Dictionary<string, string>();
} [JsonProperty("large_threshold", NullValueHandling = NullValueHandling.Ignore)]
public int? LargeThreshold;
[JsonProperty("compress", NullValueHandling = NullValueHandling.Ignore)]
public bool? Compress;
}
} }
internal sealed class ResumeCommand : WebSocketMessage<ResumeCommand.Data> internal sealed class ResumeCommand : WebSocketMessage<ResumeCommand.Data>
{ {

View File

@@ -26,6 +26,8 @@ namespace Discord.Net.WebSockets
LoginCommand msg = new LoginCommand(); LoginCommand msg = new LoginCommand();
msg.Payload.Token = token; msg.Payload.Token = token;
msg.Payload.Properties["$device"] = "Discord.Net"; msg.Payload.Properties["$device"] = "Discord.Net";
//msg.Payload.LargeThreshold = 50;
msg.Payload.Compress = true;
QueueMessage(msg); QueueMessage(msg);
} }
private async Task Redirect(string server) private async Task Redirect(string server)
@@ -67,6 +69,7 @@ namespace Discord.Net.WebSockets
protected override async Task ProcessMessage(string json) protected override async Task ProcessMessage(string json)
{ {
await base.ProcessMessage(json);
var msg = JsonConvert.DeserializeObject<WebSocketMessage>(json); var msg = JsonConvert.DeserializeObject<WebSocketMessage>(json);
if (msg.Sequence.HasValue) if (msg.Sequence.HasValue)
_lastSeq = msg.Sequence.Value; _lastSeq = msg.Sequence.Value;

View File

@@ -5,15 +5,21 @@ using System.Threading.Tasks;
namespace Discord.Net.WebSockets namespace Discord.Net.WebSockets
{ {
internal class WebSocketMessageEventArgs : EventArgs internal class WebSocketBinaryMessageEventArgs : EventArgs
{
public readonly byte[] Data;
public WebSocketBinaryMessageEventArgs(byte[] data) { Data = data; }
}
internal class WebSocketTextMessageEventArgs : EventArgs
{ {
public readonly string Message; public readonly string Message;
public WebSocketMessageEventArgs(string msg) { Message = msg; } public WebSocketTextMessageEventArgs(string msg) { Message = msg; }
} }
internal interface IWebSocketEngine internal interface IWebSocketEngine
{ {
event EventHandler<WebSocketMessageEventArgs> ProcessMessage; event EventHandler<WebSocketBinaryMessageEventArgs> BinaryMessage;
event EventHandler<WebSocketTextMessageEventArgs> TextMessage;
Task Connect(string host, CancellationToken cancelToken); Task Connect(string host, CancellationToken cancelToken);
Task Disconnect(); Task Disconnect();

View File

@@ -1,6 +1,8 @@
using Newtonsoft.Json; using Newtonsoft.Json;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.IO;
using System.IO.Compression;
using System.Linq; using System.Linq;
using System.Runtime.ExceptionServices; using System.Runtime.ExceptionServices;
using System.Threading; using System.Threading;
@@ -51,13 +53,23 @@ namespace Discord.Net.WebSockets
_connectedEvent = new ManualResetEventSlim(false); _connectedEvent = new ManualResetEventSlim(false);
_engine = new WebSocketSharpEngine(this, client.Config); _engine = new WebSocketSharpEngine(this, client.Config);
_engine.ProcessMessage += async (s, e) => _engine.BinaryMessage += async (s, e) =>
{
using (var compressed = new MemoryStream(e.Data, 2, e.Data.Length - 2))
using (var decompressed = new MemoryStream())
{
using (var zlib = new DeflateStream(compressed, CompressionMode.Decompress))
await zlib.CopyToAsync(decompressed);
decompressed.Position = 0;
using (var reader = new StreamReader(decompressed))
await ProcessMessage(await reader.ReadToEndAsync());
}
};
_engine.TextMessage += async (s, e) =>
{ {
if (_logLevel >= LogMessageSeverity.Debug)
RaiseOnLog(LogMessageSeverity.Debug, $"In: {e.Message}");
await ProcessMessage(e.Message); await ProcessMessage(e.Message);
}; };
} }
protected async Task BeginConnect() protected async Task BeginConnect()
{ {
@@ -185,7 +197,12 @@ namespace Discord.Net.WebSockets
RaiseDisconnected(wasDisconnectUnexpected, _disconnectReason?.SourceException); RaiseDisconnected(wasDisconnectUnexpected, _disconnectReason?.SourceException);
} }
protected abstract Task ProcessMessage(string json); protected virtual Task ProcessMessage(string json)
{
if (_logLevel >= LogMessageSeverity.Debug)
RaiseOnLog(LogMessageSeverity.Debug, $"In: {json}");
return TaskHelper.CompletedTask;
}
protected abstract object GetKeepAlive(); protected abstract object GetKeepAlive();
protected void QueueMessage(object message) protected void QueueMessage(object message)

View File

@@ -15,11 +15,17 @@ namespace Discord.Net.WebSockets
private readonly WebSocket _parent; private readonly WebSocket _parent;
private WSSharpNWebSocket _webSocket; private WSSharpNWebSocket _webSocket;
public event EventHandler<WebSocketMessageEventArgs> ProcessMessage; public event EventHandler<WebSocketBinaryMessageEventArgs> BinaryMessage;
private void RaiseProcessMessage(string msg) public event EventHandler<WebSocketTextMessageEventArgs> TextMessage;
private void RaiseBinaryMessage(byte[] data)
{ {
if (ProcessMessage != null) if (BinaryMessage != null)
ProcessMessage(this, new WebSocketMessageEventArgs(msg)); BinaryMessage(this, new WebSocketBinaryMessageEventArgs(data));
}
private void RaiseTextMessage(string msg)
{
if (TextMessage != null)
TextMessage(this, new WebSocketTextMessageEventArgs(msg));
} }
internal WebSocketSharpEngine(WebSocket parent, DiscordWSClientConfig config) internal WebSocketSharpEngine(WebSocket parent, DiscordWSClientConfig config)
@@ -34,9 +40,15 @@ namespace Discord.Net.WebSockets
_webSocket = new WSSharpNWebSocket(host); _webSocket = new WSSharpNWebSocket(host);
_webSocket.EmitOnPing = false; _webSocket.EmitOnPing = false;
_webSocket.EnableRedirection = true; _webSocket.EnableRedirection = true;
_webSocket.Compression = WebSocketSharp.CompressionMethod.None; _webSocket.Compression = WebSocketSharp.CompressionMethod.Deflate;
_webSocket.SetProxy(_config.ProxyUrl, _config.ProxyCredentials?.UserName, _config.ProxyCredentials?.Password); _webSocket.SetProxy(_config.ProxyUrl, _config.ProxyCredentials?.UserName, _config.ProxyCredentials?.Password);
_webSocket.OnMessage += (s, e) => RaiseProcessMessage(e.Data); _webSocket.OnMessage += (s, e) =>
{
if (e.Type == WebSocketSharp.Opcode.Binary)
RaiseBinaryMessage(e.RawData);
else if (e.Type == WebSocketSharp.Opcode.Text)
RaiseTextMessage(e.Data);
};
_webSocket.OnError += async (s, e) => _webSocket.OnError += async (s, e) =>
{ {
_parent.RaiseOnLog(LogMessageSeverity.Error, e.Exception.GetBaseException().Message); _parent.RaiseOnLog(LogMessageSeverity.Error, e.Exception.GetBaseException().Message);