diff --git a/src/Microsoft.AspNetCore.Sockets.Client/DefaultTransportFactory.cs b/src/Microsoft.AspNetCore.Sockets.Client/DefaultTransportFactory.cs index f450fe8965..baf02a5b67 100644 --- a/src/Microsoft.AspNetCore.Sockets.Client/DefaultTransportFactory.cs +++ b/src/Microsoft.AspNetCore.Sockets.Client/DefaultTransportFactory.cs @@ -47,7 +47,7 @@ namespace Microsoft.AspNetCore.Sockets.Client if ((availableServerTransports & TransportType.ServerSentEvents & _requestedTransportType) == TransportType.ServerSentEvents) { - throw new NotImplementedException(); + return new ServerSentEventsTransport(_httpClient, _loggerFactory); } if ((availableServerTransports & TransportType.LongPolling & _requestedTransportType) == TransportType.LongPolling) diff --git a/src/Microsoft.AspNetCore.Sockets.Client/ServerSentEventsTransport.cs b/src/Microsoft.AspNetCore.Sockets.Client/ServerSentEventsTransport.cs new file mode 100644 index 0000000000..a675ee7e95 --- /dev/null +++ b/src/Microsoft.AspNetCore.Sockets.Client/ServerSentEventsTransport.cs @@ -0,0 +1,231 @@ +// Copyright (c) .NET Foundation. All rights reserved. +// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information. + +using System; +using System.Collections.Generic; +using System.IO; +using System.IO.Pipelines; +using System.IO.Pipelines.Text.Primitives; +using System.Net.Http; +using System.Net.Http.Headers; +using System.Text; +using System.Text.Formatting; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Sockets.Internal.Formatters; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Internal; + +namespace Microsoft.AspNetCore.Sockets.Client +{ + public class ServerSentEventsTransport : ITransport + { + private static readonly string DefaultUserAgent = "Microsoft.AspNetCore.SignalR.Client/0.0.0"; + private static readonly ProductInfoHeaderValue DefaultUserAgentHeader = ProductInfoHeaderValue.Parse(DefaultUserAgent); + + private readonly HttpClient _httpClient; + private readonly ILogger _logger; + private readonly CancellationTokenSource _transportCts = new CancellationTokenSource(); + private readonly ServerSentEventsMessageParser _parser = new ServerSentEventsMessageParser(); + + private IChannelConnection _application; + + public Task Running { get; private set; } = Task.CompletedTask; + + public ServerSentEventsTransport(HttpClient httpClient) + : this(httpClient, null) + { } + + public ServerSentEventsTransport(HttpClient httpClient, ILoggerFactory loggerFactory) + { + if (httpClient == null) + { + throw new ArgumentNullException(nameof(_httpClient)); + } + + _httpClient = httpClient; + _logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger(); + } + + public Task StartAsync(Uri url, IChannelConnection application) + { + _logger.LogInformation("Starting {transportName}", nameof(ServerSentEventsTransport)); + + _application = application; + var sseUrl = Utils.AppendPath(url, "sse"); + var sendUrl = Utils.AppendPath(url, "send"); + var sendTask = SendMessages(sendUrl, _transportCts.Token); + var receiveTask = OpenConnection(_application, sseUrl, _transportCts.Token); + + Running = Task.WhenAll(sendTask, receiveTask).ContinueWith(t => + { + if (t.Exception != null) { _logger.LogError(t.Exception, "Transport stopped"); } + + _application.Output.TryComplete(t.IsFaulted ? t.Exception.InnerException : null); + return t; + }).Unwrap(); + + return TaskCache.CompletedTask; + } + + private async Task OpenConnection(IChannelConnection application, Uri url, CancellationToken cancellationToken) + { + _logger.LogInformation("Starting receive loop"); + + var request = new HttpRequestMessage(HttpMethod.Get, url); + request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream")); + var response = await _httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken); + + var stream = await response.Content.ReadAsStreamAsync(); + + var pipelineReader = stream.AsPipelineReader(); + try + { + while (true) + { + var result = await pipelineReader.ReadAsync(); + var input = result.Buffer; + var consumed = input.Start; + var examined = input.End; + + try + { + if (input.IsEmpty && result.IsCompleted) + { + _logger.LogDebug("Server-Sent Event Stream ended"); + break; + } + + var parseResult = _parser.ParseMessage(input, out consumed, out examined, out var message); + + switch (parseResult) + { + case ServerSentEventsMessageParser.ParseResult.Completed: + _application.Output.TryWrite(message); + _parser.Reset(); + break; + case ServerSentEventsMessageParser.ParseResult.Incomplete: + if (result.IsCompleted) + { + throw new FormatException("Incomplete message"); + } + break; + } + } + finally + { + pipelineReader.Advance(consumed, examined); + } + } + } + finally + { + _transportCts.Cancel(); + } + } + + private async Task SendMessages(Uri sendUrl, CancellationToken cancellationToken) + { + _logger.LogInformation("Starting the send loop"); + + List messages = null; + try + { + while (await _application.Input.WaitToReadAsync(cancellationToken)) + { + messages = new List(); + while (!cancellationToken.IsCancellationRequested && _application.Input.TryRead(out SendMessage message)) + { + messages.Add(message); + } + + if (messages.Count > 0) + { + _logger.LogDebug("Sending {messageCount} message(s) to the server using url: {url}", messages.Count, sendUrl); + + var request = new HttpRequestMessage(HttpMethod.Post, sendUrl); + request.Headers.UserAgent.Add(DefaultUserAgentHeader); + + var memoryStream = new MemoryStream(); + + var pipe = memoryStream.AsPipelineWriter(); + var output = new PipelineTextOutput(pipe, TextEncoder.Utf8); + await WriteMessagesAsync(messages, output, MessageFormat.Binary); + + memoryStream.Seek(0, SeekOrigin.Begin); + + request.Content = new StreamContent(memoryStream); + request.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(MessageFormatter.GetContentType(MessageFormat.Binary)); + + var response = await _httpClient.SendAsync(request); + response.EnsureSuccessStatusCode(); + + _logger.LogDebug("Message(s) sent successfully"); + foreach (var message in messages) + { + message.SendResult?.TrySetResult(null); + } + } + } + } + catch (OperationCanceledException) + { + _logger.LogError("Send cancelled"); + + if (messages != null) + { + foreach (var message in messages) + { + message.SendResult?.TrySetCanceled(); + } + } + } + catch (Exception ex) + { + _logger.LogDebug("Error while sending to '{url}' : '{exception}'", sendUrl, ex); + if (messages != null) + { + foreach (var message in messages) + { + message.SendResult?.TrySetException(ex); + } + } + throw; + } + finally + { + // Make sure the poll loop is terminated + _transportCts.Cancel(); + } + + _logger.LogInformation("Send loop stopped"); + } + + private async Task WriteMessagesAsync(List messages, PipelineTextOutput output, MessageFormat format) + { + output.Append(MessageFormatter.GetFormatIndicator(format), TextEncoder.Utf8); + + foreach (var message in messages) + { + _logger.LogDebug("Writing '{messageType}' message to the server", message.Type); + + var payload = message.Payload ?? Array.Empty(); + if (!MessageFormatter.TryWriteMessage(new Message(payload, message.Type, endOfMessage: true), output, format)) + { + // We didn't get any more memory! + throw new InvalidOperationException("Unable to write message to pipeline"); + } + await output.FlushAsync(); + } + } + + public async Task StopAsync() + { + _logger.LogInformation("Transport {transportName} is stopping", nameof(ServerSentEventsTransport)); + _transportCts.Cancel(); + _application.Output.TryComplete(); + await Running; + } + } +} diff --git a/src/Microsoft.AspNetCore.Sockets.Common/Internal/Formatters/ServerSentEventsMessageParser.cs b/src/Microsoft.AspNetCore.Sockets.Common/Internal/Formatters/ServerSentEventsMessageParser.cs new file mode 100644 index 0000000000..e8e014812a --- /dev/null +++ b/src/Microsoft.AspNetCore.Sockets.Common/Internal/Formatters/ServerSentEventsMessageParser.cs @@ -0,0 +1,215 @@ +// Copyright (c) .NET Foundation. All rights reserved. +// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information. + +using System; +using System.Collections.Generic; +using System.IO.Pipelines; +using System.Runtime.CompilerServices; +using System.Text; + +namespace Microsoft.AspNetCore.Sockets.Internal.Formatters +{ + public class ServerSentEventsMessageParser + { + private const byte ByteCR = (byte)'\r'; + private const byte ByteLF = (byte)'\n'; + private const byte ByteT = (byte)'T'; + private const byte ByteB = (byte)'B'; + private const byte ByteC = (byte)'C'; + private const byte ByteE = (byte)'E'; + + private static byte[] _dataPrefix = Encoding.UTF8.GetBytes("data: "); + private static byte[] _sseLineEnding = Encoding.UTF8.GetBytes("\r\n"); + private static byte[] _newLine = Encoding.UTF8.GetBytes(Environment.NewLine); + + private InternalParseState _internalParserState = InternalParseState.ReadMessageType; + private List _data = new List(); + private MessageType _messageType = MessageType.Text; + + public ParseResult ParseMessage(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined, out Message message) + { + consumed = buffer.Start; + examined = buffer.End; + message = new Message(); + var reader = new ReadableBufferReader(buffer); + _messageType = MessageType.Text; + + var start = consumed; + var end = examined; + + while (!reader.End) + { + + if (ReadCursorOperations.Seek(start, end, out var lineEnd, ByteLF) == -1) + { + // For the case of data: Foo\r\n\r\ + if (_internalParserState == InternalParseState.ReadEndOfMessage) + { + if(ConvertBufferToSpan(buffer.Slice(start, buffer.End)).Length > 1) + { + throw new FormatException("Expected a \\r\\n frame ending"); + } + } + + // Partial message. We need to read more. + return ParseResult.Incomplete; + } + + lineEnd = buffer.Move(lineEnd, 1); + var line = ConvertBufferToSpan(buffer.Slice(start, lineEnd)); + reader.Skip(line.Length); + + if (line.Length <= 1) + { + throw new FormatException("There was an error in the frame format"); + } + + if (IsMessageEnd(line)) + { + _internalParserState = InternalParseState.ReadEndOfMessage; + } + + // To ensure that the \n was preceded by a \r + // since messages can't contain \n. + // data: foo\n\bar should be encoded as + // data: foo\r\n + // data: bar\r\n + else if (line[line.Length - _sseLineEnding.Length] != ByteCR) + { + throw new FormatException("Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'"); + } + else + { + EnsureStartsWithDataPrefix(line); + } + + switch (_internalParserState) + { + case InternalParseState.ReadMessageType: + _messageType = GetMessageType(line); + + _internalParserState = InternalParseState.ReadMessagePayload; + + start = lineEnd; + consumed = lineEnd; + break; + case InternalParseState.ReadMessagePayload: + + // Slice away the 'data: ' + var payloadLength = line.Length - (_dataPrefix.Length + _sseLineEnding.Length); + var newData = line.Slice(_dataPrefix.Length, payloadLength).ToArray(); + _data.Add(newData); + + start = lineEnd; + consumed = lineEnd; + break; + case InternalParseState.ReadEndOfMessage: + if (_data.Count > 0) + { + // Find the final size of the payload + var payloadSize = 0; + foreach (var dataLine in _data) + { + payloadSize += dataLine.Length + _newLine.Length; + } + + // Allocate space in the paylod buffer for the data and the new lines. + // Subtract newLine length because we don't want a trailing newline. + var payload = new byte[payloadSize - _newLine.Length]; + + var offset = 0; + foreach (var dataLine in _data) + { + dataLine.CopyTo(payload, offset); + offset += dataLine.Length; + if (offset < payload.Length) + { + _newLine.CopyTo(payload, offset); + offset += _newLine.Length; + } + } + message = new Message(payload, _messageType); + } + else + { + // Empty message + message = new Message(Array.Empty(), _messageType); + } + + consumed = lineEnd; + examined = consumed; + return ParseResult.Completed; + } + if (reader.Peek() == ByteCR) + { + _internalParserState = InternalParseState.ReadEndOfMessage; + } + } + return ParseResult.Incomplete; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private Span ConvertBufferToSpan(ReadableBuffer buffer) + { + if (buffer.IsSingleSpan) + { + return buffer.First.Span; + } + return buffer.ToArray(); + } + + public void Reset() + { + _internalParserState = InternalParseState.ReadMessageType; + _data.Clear(); + } + + private void EnsureStartsWithDataPrefix(ReadOnlySpan line) + { + if (!line.StartsWith(_dataPrefix)) + { + throw new FormatException("Expected the message prefix 'data: '"); + } + } + + private bool IsMessageEnd(ReadOnlySpan line) + { + return line.Length == _sseLineEnding.Length && line.SequenceEqual(_sseLineEnding); + } + + private MessageType GetMessageType(ReadOnlySpan line) + { + EnsureStartsWithDataPrefix(line); + + // Skip the "data: " part of the line + var type = line[_dataPrefix.Length]; + switch (type) + { + case ByteT: + return MessageType.Text; + case ByteB: + throw new NotSupportedException("Support for binary messages has not been implemented yet"); + case ByteC: + return MessageType.Close; + case ByteE: + return MessageType.Error; + default: + throw new FormatException($"Unknown message type: '{(char)type}'"); + } + } + + public enum ParseResult + { + Completed, + Incomplete, + } + + private enum InternalParseState + { + ReadMessageType, + ReadMessagePayload, + ReadEndOfMessage, + Error + } + } +} diff --git a/test/Microsoft.AspNetCore.SignalR.Tests/DefaultTransportFactoryTests.cs b/test/Microsoft.AspNetCore.SignalR.Tests/DefaultTransportFactoryTests.cs index 0e40c662cc..80dad3fc64 100644 --- a/test/Microsoft.AspNetCore.SignalR.Tests/DefaultTransportFactoryTests.cs +++ b/test/Microsoft.AspNetCore.SignalR.Tests/DefaultTransportFactoryTests.cs @@ -33,6 +33,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests [Theory] [InlineData(TransportType.WebSockets, typeof(WebSocketsTransport))] + [InlineData(TransportType.ServerSentEvents, typeof(ServerSentEventsTransport))] [InlineData(TransportType.LongPolling, typeof(LongPollingTransport))] [OSSkipCondition(OperatingSystems.Windows, WindowsVersions.Win7, WindowsVersions.Win2008R2, SkipReason = "No WebSockets Client for this platform")] public void DefaultTransportFactoryCreatesRequestedTransportIfAvailable(TransportType requestedTransport, Type expectedTransportType) @@ -67,7 +68,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests } [Theory] - [InlineData(TransportType.WebSockets, typeof(LongPollingTransport))] + [InlineData(TransportType.WebSockets, typeof(ServerSentEventsTransport))] [InlineData(TransportType.LongPolling, typeof(LongPollingTransport))] public void DefaultTransportFactoryCreatesRequestedTransportIfAvailable_Win7(TransportType requestedTransport, Type expectedTransportType) { diff --git a/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs b/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs index c707b01948..c08d6a3e60 100644 --- a/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs +++ b/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs @@ -78,16 +78,16 @@ namespace Microsoft.AspNetCore.SignalR.Tests var receiveTcs = new TaskCompletionSource(); connection.Received += (data, format) => receiveTcs.TrySetResult(Encoding.UTF8.GetString(data)); connection.Closed += e => + { + if (e != null) { - if (e != null) - { - receiveTcs.TrySetException(e); - } - else - { - receiveTcs.TrySetResult(null); - } - }; + receiveTcs.TrySetException(e); + } + else + { + receiveTcs.TrySetResult(null); + } + }; await connection.StartAsync(transportType); @@ -146,7 +146,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests new[] { new object[] { TransportType.WebSockets }, + new object[] { TransportType.ServerSentEvents }, new object[] { TransportType.LongPolling } }; } -} +} \ No newline at end of file diff --git a/test/Microsoft.AspNetCore.Sockets.Common.Tests/Internal/Formatters/ServerSentEventsParserTests.cs b/test/Microsoft.AspNetCore.Sockets.Common.Tests/Internal/Formatters/ServerSentEventsParserTests.cs new file mode 100644 index 0000000000..7a0b4dd65f --- /dev/null +++ b/test/Microsoft.AspNetCore.Sockets.Common.Tests/Internal/Formatters/ServerSentEventsParserTests.cs @@ -0,0 +1,242 @@ +// Copyright (c) .NET Foundation. All rights reserved. +// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information. + +using System; +using System.Collections.Generic; +using System.IO.Pipelines; +using System.Text; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Sockets.Internal.Formatters; +using Xunit; + +namespace Microsoft.AspNetCore.Sockets.Common.Tests.Internal.Formatters +{ + public class ServerSentEventsParserTests + { + [Theory] + [InlineData("data: T\r\n\r\n", "")] + [InlineData("data: T\r\ndata: \r\r\n\r\n", "\r")] + [InlineData("data: T\r\ndata: A\rB\r\n\r\n", "A\rB")] + [InlineData("data: T\r\ndata: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T\r\ndata: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T\r\ndata: Hello, World\r\n\r\ndata: ", "Hello, World")] + public void ParseSSEMessageSuccessCases(string encodedMessage, string expectedMessage) + { + var buffer = Encoding.UTF8.GetBytes(encodedMessage); + var readableBuffer = ReadableBuffer.Create(buffer); + var parser = new ServerSentEventsMessageParser(); + + var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out Message message); + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult); + Assert.Equal(MessageType.Text, message.Type); + Assert.Equal(consumed, examined); + + var result = Encoding.UTF8.GetString(message.Payload); + Assert.Equal(expectedMessage, result); + } + + [Theory] + [InlineData("data: X\r\n", "Unknown message type: 'X'")] + [InlineData("data: T\n", "Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'")] + [InlineData("data: X\r\n\r\n", "Unknown message type: 'X'")] + [InlineData("data: Not the message type\r\n\r\n", "Unknown message type: 'N'")] + [InlineData("data: T\r\ndata: Hello, World\r\r\n\n", "There was an error in the frame format")] + [InlineData("data: Not the message type\r\r\n", "Unknown message type: 'N'")] + [InlineData("data: T\r\ndata: Hello, World\n\n", "Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'")] + [InlineData("data: T\r\nfoo: Hello, World\r\n\r\n", "Expected the message prefix 'data: '")] + [InlineData("foo: T\r\ndata: Hello, World\r\n\r\n", "Expected the message prefix 'data: '")] + [InlineData("food: T\r\ndata: Hello, World\r\n\r\n", "Expected the message prefix 'data: '")] + [InlineData("data: T\r\ndata: Hello, World\r\n\n", "There was an error in the frame format")] + [InlineData("data: T\r\ndata: Hello\n, World\r\n\r\n", "Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'")] + [InlineData("data: data: \r\n", "Unknown message type: 'd'")] + [InlineData("data: T\r\ndata: Hello, World\r\n\r\\", "Expected a \\r\\n frame ending")] + [InlineData("data: T\r\ndata: Major\r\ndata: Key\rndata: Alert\r\n\r\\", "Expected a \\r\\n frame ending")] + [InlineData("data: T\r\ndata: Major\r\ndata: Key\r\ndata: Alert\r\n\r\\", "Expected a \\r\\n frame ending")] + public void ParseSSEMessageFailureCases(string encodedMessage, string expectedExceptionMessage) + { + var buffer = Encoding.UTF8.GetBytes(encodedMessage); + var readableBuffer = ReadableBuffer.Create(buffer); + var parser = new ServerSentEventsMessageParser(); + + var ex = Assert.Throws(() => { parser.ParseMessage(readableBuffer, out var consumed, out var examined, out Message message); }); + Assert.Equal(expectedExceptionMessage, ex.Message); + } + + [Theory] + [InlineData("")] + [InlineData("data:")] + [InlineData("data: \r")] + [InlineData("data: T\r\nda")] + [InlineData("data: T\r\ndata:")] + [InlineData("data: T\r\ndata: Hello, World")] + [InlineData("data: T\r\ndata: Hello, World\r")] + [InlineData("data: T\r\ndata: Hello, World\r\n")] + [InlineData("data: T\r\ndata: Hello, World\r\n\r")] + public void ParseSSEMessageIncompleteParseResult(string encodedMessage) + { + var buffer = Encoding.UTF8.GetBytes(encodedMessage); + var readableBuffer = ReadableBuffer.Create(buffer); + var parser = new ServerSentEventsMessageParser(); + + var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out Message message); + + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Incomplete, parseResult); + } + + [Theory] + [InlineData("d", "ata: T\r\ndata: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T", "\r\ndata: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T\r", "\ndata: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T\r\n", "data: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T\r\nd", "ata: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T\r\ndata: ", "Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: T\r\ndata: Hello, World", "\r\n\r\n", "Hello, World")] + [InlineData("data: T\r\ndata: Hello, World\r\n", "\r\n", "Hello, World")] + [InlineData("data: T", "\r\ndata: Hello, World\r\n\r\n", "Hello, World")] + [InlineData("data: ", "T\r\ndata: Hello, World\r\n\r\n", "Hello, World")] + public async Task ParseMessageAcrossMultipleReadsSuccess(string encodedMessagePart1, string encodedMessagePart2, string expectedMessage) + { + using (var pipeFactory = new PipeFactory()) + { + var pipe = pipeFactory.Create(); + + // Read the first part of the message + await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(encodedMessagePart1)); + + var result = await pipe.Reader.ReadAsync(); + var parser = new ServerSentEventsMessageParser(); + + var parseResult = parser.ParseMessage(result.Buffer, out var consumed, out var examined, out Message message); + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Incomplete, parseResult); + + pipe.Reader.Advance(consumed, examined); + + // Send the rest of the data and parse the complete message + await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(encodedMessagePart2)); + result = await pipe.Reader.ReadAsync(); + + parseResult = parser.ParseMessage(result.Buffer, out consumed, out examined, out message); + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult); + Assert.Equal(MessageType.Text, message.Type); + Assert.Equal(consumed, examined); + + var resultMessage = Encoding.UTF8.GetString(message.Payload); + Assert.Equal(expectedMessage, resultMessage); + } + } + + [Theory] + [InlineData("data: ", "X\r\n", "Unknown message type: 'X'")] + [InlineData("data: T", "\n", "Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'")] + [InlineData("data: ", "X\r\n\r\n", "Unknown message type: 'X'")] + [InlineData("data: ", "Not the message type\r\n\r\n", "Unknown message type: 'N'")] + [InlineData("data: T\r\n", "data: Hello, World\r\r\n\n", "There was an error in the frame format")] + [InlineData("data:", " Not the message type\r\r\n", "Unknown message type: 'N'")] + [InlineData("data: T\r\n", "data: Hello, World\n\n", "Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'")] + [InlineData("data: T\r\nf", "oo: Hello, World\r\n\r\n", "Expected the message prefix 'data: '")] + [InlineData("foo", ": T\r\ndata: Hello, World\r\n\r\n", "Expected the message prefix 'data: '")] + [InlineData("food:", " T\r\ndata: Hello, World\r\n\r\n", "Expected the message prefix 'data: '")] + [InlineData("data: T\r\ndata: Hello, W", "orld\r\n\n", "There was an error in the frame format")] + [InlineData("data: T\r\nda", "ta: Hello\n, World\r\n\r\n", "Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'")] + [InlineData("data:", " data: \r\n", "Unknown message type: 'd'")] + [InlineData("data: ", "T\r\ndata: Major\r\ndata: Key\r\ndata: Alert\r\n\r\\", "Expected a \\r\\n frame ending")] + public async Task ParseMessageAcrossMultipleReadsFailure(string encodedMessagePart1, string encodedMessagePart2, string expectedMessage) + { + using (var pipeFactory = new PipeFactory()) + { + var pipe = pipeFactory.Create(); + + // Read the first part of the message + await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(encodedMessagePart1)); + + var result = await pipe.Reader.ReadAsync(); + var parser = new ServerSentEventsMessageParser(); + + var parseResult = parser.ParseMessage(result.Buffer, out var consumed, out var examined, out Message message); + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Incomplete, parseResult); + + pipe.Reader.Advance(consumed, examined); + + // Send the rest of the data and parse the complete message + await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(encodedMessagePart2)); + result = await pipe.Reader.ReadAsync(); + + var ex = Assert.Throws(() => parser.ParseMessage(result.Buffer, out consumed, out examined, out message)); + Assert.Equal(expectedMessage, ex.Message); + + } + } + + [Fact] + public async Task ParseMultipleMessages() + { + using (var pipeFactory = new PipeFactory()) + { + var pipe = pipeFactory.Create(); + + var message1 = "data: T\r\ndata: foo\r\n\r\n"; + var message2 = "data: T\r\ndata: bar\r\n\r\n"; + // Read the first part of the message + await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(message1 + message2)); + + var result = await pipe.Reader.ReadAsync(); + var parser = new ServerSentEventsMessageParser(); + + var parseResult = parser.ParseMessage(result.Buffer, out var consumed, out var examined, out var message); + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult); + Assert.Equal(MessageType.Text, message.Type); + Assert.Equal("foo", Encoding.UTF8.GetString(message.Payload)); + Assert.Equal(consumed, result.Buffer.Move(result.Buffer.Start, message1.Length)); + pipe.Reader.Advance(consumed, examined); + Assert.Equal(consumed, examined); + + parser.Reset(); + + result = await pipe.Reader.ReadAsync(); + parseResult = parser.ParseMessage(result.Buffer, out consumed, out examined, out message); + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult); + Assert.Equal(MessageType.Text, message.Type); + Assert.Equal("bar", Encoding.UTF8.GetString(message.Payload)); + pipe.Reader.Advance(consumed, examined); + } + } + + public static IEnumerable MultilineMessages + { + get + { + yield return new object[] { "data: T\r\ndata: Shaolin\r\ndata: Fantastic\r\n\r\n", "Shaolin" + Environment.NewLine + " Fantastic" }; + yield return new object[] { "data: T\r\ndata: The\r\ndata: Get\r\ndata: Down\r\n\r\n", "The" + Environment.NewLine + "Get" + Environment.NewLine + "Down" }; + } + } + + [Theory] + [MemberData(nameof(MultilineMessages))] + public void ParseMessagesWithMultipleDataLines(string encodedMessage, string expectedMessage) + { + var buffer = Encoding.UTF8.GetBytes(encodedMessage); + var readableBuffer = ReadableBuffer.Create(buffer); + var parser = new ServerSentEventsMessageParser(); + + var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out Message message); + Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult); + Assert.Equal(MessageType.Text, message.Type); + Assert.Equal(consumed, examined); + + var result = Encoding.UTF8.GetString(message.Payload); + Assert.Equal(expectedMessage, result); + } + + [Fact] + public void ParseSSEMessageBinaryNotSupported() + { + var encodedMessage = "data: B\r\ndata: \r\n\r\n"; + var buffer = Encoding.UTF8.GetBytes(encodedMessage); + var readableBuffer = ReadableBuffer.Create(buffer); + var parser = new ServerSentEventsMessageParser(); + + var ex = Assert.Throws(() => { parser.ParseMessage(readableBuffer, out var consumed, out var examined, out Message message); }); + Assert.Equal("Support for binary messages has not been implemented yet", ex.Message); + } + } +}