fix: ws receive buffer size.

oneRain 2021-02-01 16:56:38 +08:00
parent 1de886ea1e
commit 8645caf4a3
3 changed files with 25 additions and 11 deletions

View File

@ -136,9 +136,9 @@ namespace LeanCloud.LiveQuery.Internal {
await client.Close(); await client.Close();
} }
private void OnClientMessage(byte[] bytes) { private void OnClientMessage(byte[] bytes, int length) {
try { try {
string json = Encoding.UTF8.GetString(bytes); string json = Encoding.UTF8.GetString(bytes, 0, length);
Dictionary<string, object> msg = JsonConvert.DeserializeObject<Dictionary<string, object>>(json, Dictionary<string, object> msg = JsonConvert.DeserializeObject<Dictionary<string, object>>(json,
LCJsonConverter.Default); LCJsonConverter.Default);
LCLogger.Debug($"{id} <= {json}"); LCLogger.Debug($"{id} <= {json}");

View File

@ -160,9 +160,9 @@ namespace LeanCloud.Realtime.Internal.Connection {
} }
} }
private void OnMessage(byte[] bytes) { private void OnMessage(byte[] bytes, int length) {
try { try {
GenericCommand command = GenericCommand.Parser.ParseFrom(bytes); GenericCommand command = GenericCommand.Parser.ParseFrom(bytes, 0, length);
LCLogger.Debug($"{id} <= {FormatCommand(command)}"); LCLogger.Debug($"{id} <= {FormatCommand(command)}");
if (command.HasI) { if (command.HasI) {
// 应答 // 应答

View File

@ -6,13 +6,15 @@ using System.Text;
namespace LeanCloud.Realtime.Internal.WebSocket { namespace LeanCloud.Realtime.Internal.WebSocket {
public class LCWebSocketClient { public class LCWebSocketClient {
// .net standard 2.0 好像在拼合 Frame 时有 bug所以将这个值调整大一些 // .net standard 2.0 好像在拼合 Frame 时有 bug所以将这个值调整大一些
private const int RECV_BUFFER_SIZE = 1024 * 5; private const int SEND_BUFFER_SIZE = 1024 * 5;
private const int RECV_BUFFER_SIZE = 1024 * 8;
private const int MSG_BUFFER_SIZE = 1024 * 10;
private const int CLOSE_TIMEOUT = 5000; private const int CLOSE_TIMEOUT = 5000;
private const int CONNECT_TIMEOUT = 10000; private const int CONNECT_TIMEOUT = 10000;
public Action<byte[]> OnMessage; public Action<byte[], int> OnMessage;
public Action OnClose; public Action OnClose;
@ -23,6 +25,7 @@ namespace LeanCloud.Realtime.Internal.WebSocket {
LCLogger.Debug($"Connecting WebSocket: {server}"); LCLogger.Debug($"Connecting WebSocket: {server}");
Task timeoutTask = Task.Delay(CONNECT_TIMEOUT); Task timeoutTask = Task.Delay(CONNECT_TIMEOUT);
ws = new ClientWebSocket(); ws = new ClientWebSocket();
ws.Options.SetBuffer(RECV_BUFFER_SIZE, SEND_BUFFER_SIZE);
if (!string.IsNullOrEmpty(subProtocol)) { if (!string.IsNullOrEmpty(subProtocol)) {
ws.Options.AddSubProtocol(subProtocol); ws.Options.AddSubProtocol(subProtocol);
} }
@ -78,10 +81,12 @@ namespace LeanCloud.Realtime.Internal.WebSocket {
} }
private async Task StartReceive() { private async Task StartReceive() {
byte[] buffer = new byte[RECV_BUFFER_SIZE]; byte[] recvBuffer = new byte[RECV_BUFFER_SIZE];
byte[] msgBuffer = new byte[MSG_BUFFER_SIZE];
int offset = 0;
try { try {
while (ws.State == WebSocketState.Open) { while (ws.State == WebSocketState.Open) {
WebSocketReceiveResult result = await ws.ReceiveAsync(new ArraySegment<byte>(buffer), default); WebSocketReceiveResult result = await ws.ReceiveAsync(new ArraySegment<byte>(recvBuffer), default);
if (result.MessageType == WebSocketMessageType.Close) { if (result.MessageType == WebSocketMessageType.Close) {
LCLogger.Debug($"Receive Closed: {result.CloseStatus}"); LCLogger.Debug($"Receive Closed: {result.CloseStatus}");
if (ws.State == WebSocketState.CloseReceived) { if (ws.State == WebSocketState.CloseReceived) {
@ -98,9 +103,18 @@ namespace LeanCloud.Realtime.Internal.WebSocket {
} else { } else {
// 拼合 WebSocket Message // 拼合 WebSocket Message
int length = result.Count; int length = result.Count;
byte[] data = new byte[length]; if (offset + length > msgBuffer.Length) {
Array.Copy(buffer, data, length); // 反序列化数组大小不够,则以 2x 扩充
OnMessage?.Invoke(data); byte[] newBuffer = new byte[msgBuffer.Length * 2];
Array.Copy(msgBuffer, newBuffer, msgBuffer.Length);
msgBuffer = newBuffer;
}
Array.Copy(recvBuffer, 0, msgBuffer, offset, length);
offset += length;
if (result.EndOfMessage) {
OnMessage?.Invoke(msgBuffer, offset);
offset = 0;
}
} }
} }
} catch (Exception e) { } catch (Exception e) {