YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-30 18:56:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor web socket stream. Add mqtt as default sub protocol.

Browse Source
This commit is contained in:
Christian
2018-03-25 14:42:12 +02:00
parent 7a9ddfc2b5
commit 3193381b69
2 changed files with 23 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetStandard/Client/MqttClientWebSocketOptions.cs
+1 -1
View File
@@ -9,7 +9,7 @@ namespace MQTTnet.Client
public IDictionary<string, string> RequestHeaders { get; set; } public IDictionary<string, string> RequestHeaders { get; set; }
public ICollection<string> SubProtocols { get; set; } public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" };
public CookieContainer CookieContainer { get; set; } public CookieContainer CookieContainer { get; set; }
Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs
+20 -12
View File
@@ -11,9 +11,9 @@ namespace MQTTnet.Implementations
{ {
public class WebSocketStream : Stream public class WebSocketStream : Stream
{ {
private readonly WebSocket _webSocket; private readonly byte[] _chunckBuffer = new byte[MqttWebSocketChannel.BufferSize];
private readonly byte[] _chunkBuffer = new byte[MqttWebSocketChannel.BufferSize];
private readonly Queue<byte> _buffer = new Queue<byte>(MqttWebSocketChannel.BufferSize); private readonly Queue<byte> _buffer = new Queue<byte>(MqttWebSocketChannel.BufferSize);
private readonly WebSocket _webSocket;
public WebSocketStream(WebSocket webSocket) public WebSocketStream(WebSocket webSocket)
{ {
@@ -55,9 +55,10 @@ namespace MQTTnet.Implementations
// Use existing date from buffer. // Use existing date from buffer.
while (count > 0 && _buffer.Any()) while (count > 0 && _buffer.Any())
{ {
buffer[offset++] = _buffer.Dequeue(); buffer[offset] = _buffer.Dequeue();
count--; count--;
bytesRead++; bytesRead++;
offset++;
} }
if (count == 0) if (count == 0)
@@ -65,15 +66,17 @@ namespace MQTTnet.Implementations
return bytesRead; return bytesRead;
} }
// Fetch new data if the buffer is not full.
while (_webSocket.State == WebSocketState.Open) while (_webSocket.State == WebSocketState.Open)
{ {
await FetchChunkAsync(cancellationToken); await FetchChunkAsync(cancellationToken).ConfigureAwait(false);
while (count > 0 && _buffer.Any()) while (count > 0 && _buffer.Any())
{ {
buffer[offset++] = _buffer.Dequeue(); buffer[offset] = _buffer.Dequeue();
count--; count--;
bytesRead++; bytesRead++;
offset++;
} }
if (count == 0) if (count == 0)
@@ -112,18 +115,23 @@ namespace MQTTnet.Implementations
private async Task FetchChunkAsync(CancellationToken cancellationToken) private async Task FetchChunkAsync(CancellationToken cancellationToken)
{ {
var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(_chunkBuffer, 0, _chunkBuffer.Length), cancellationToken).ConfigureAwait(false); var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(_chunckBuffer, 0, _chunckBuffer.Length), cancellationToken).ConfigureAwait(false);
for (var i = 0; i < response.Count; i++)
{
var @byte = _chunkBuffer[i];
_buffer.Enqueue(@byte);
}
if (response.MessageType == WebSocketMessageType.Close) if (response.MessageType == WebSocketMessageType.Close)
{ {
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false); await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
} }
else if (response.MessageType == WebSocketMessageType.Binary)
{
for (var i = 0; i < response.Count; i++)
{
_buffer.Enqueue(_chunckBuffer[i]);
}
}
else if (response.MessageType == WebSocketMessageType.Text)
{
throw new MqttProtocolViolationException("WebSocket channel received TEXT message.");
}
} }
} }
} }