fix #204 by implementing SSE formatter (#210)

src/Microsoft.AspNetCore.Sockets.Common/Formatters/BinaryMessageFormatter.cs

@@ -5,7 +5,7 @@ using System;
 using System.Binary;
 using System.IO.Pipelines;
 
-namespace Microsoft.AspNetCore.Sockets
+namespace Microsoft.AspNetCore.Sockets.Formatters
 {
     internal static class BinaryMessageFormatter
     {

src/Microsoft.AspNetCore.Sockets.Common/Formatters/MessageFormatter.cs

@@ -1,9 +1,9 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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;
 
-namespace Microsoft.AspNetCore.Sockets
+namespace Microsoft.AspNetCore.Sockets.Formatters
 {
     public static class MessageFormatter
     {
@@ -16,6 +16,7 @@ namespace Microsoft.AspNetCore.Sockets
                 // giving it to us. Hence we throw, instead of returning false.
                 throw new InvalidOperationException("Cannot format message where endOfMessage is false using this format");
             }
+
             return format == MessageFormat.Text ?
                 TextMessageFormatter.TryFormatMessage(message, buffer, out bytesWritten) :
                 BinaryMessageFormatter.TryFormatMessage(message, buffer, out bytesWritten);

src/Microsoft.AspNetCore.Sockets.Common/Formatters/ServerSentEventsMessageFormatter.cs

@@ -0,0 +1,186 @@
+// 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.IO.Pipelines;
+using System.Text;
+
+namespace Microsoft.AspNetCore.Sockets.Formatters
+{
+    public static class ServerSentEventsMessageFormatter
+    {
+        private static readonly Span<byte> DataPrefix = new byte[] { (byte)'d', (byte)'a', (byte)'t', (byte)'a', (byte)':', (byte)' ' };
+        private static readonly Span<byte> Newline = new byte[] { (byte)'\r', (byte)'\n' };
+
+        private const byte LineFeed = (byte)'\n';
+        private const byte TextTypeFlag = (byte)'T';
+        private const byte BinaryTypeFlag = (byte)'B';
+        private const byte CloseTypeFlag = (byte)'C';
+        private const byte ErrorTypeFlag = (byte)'E';
+
+        public static bool TryFormatMessage(Message message, Span<byte> buffer, out int bytesWritten)
+        {
+            if (!message.EndOfMessage)
+            {
+                // This is a truely exceptional condition since we EXPECT callers to have already
+                // buffered incomplete messages and synthesized the correct, complete message before
+                // giving it to us. Hence we throw, instead of returning false.
+                throw new InvalidOperationException("Cannot format message where endOfMessage is false using this format");
+            }
+
+            // Need at least: Length of 'data: ', one character type, one \r\n, and the trailing \r\n
+            if (buffer.Length < DataPrefix.Length + 1 + Newline.Length + Newline.Length)
+            {
+                bytesWritten = 0;
+                return false;
+            }
+            DataPrefix.CopyTo(buffer);
+            buffer = buffer.Slice(DataPrefix.Length);
+            if (!TryFormatType(buffer, message.Type))
+            {
+                bytesWritten = 0;
+                return false;
+            }
+            buffer = buffer.Slice(1);
+
+            Newline.CopyTo(buffer);
+            buffer = buffer.Slice(Newline.Length);
+
+            // Write the payload
+            if (!TryFormatPayload(message.Payload.Buffer, message.Type, buffer, out var writtenForPayload))
+            {
+                bytesWritten = 0;
+                return false;
+            }
+            buffer = buffer.Slice(writtenForPayload);
+
+            if (buffer.Length < Newline.Length)
+            {
+                bytesWritten = 0;
+                return false;
+            }
+            Newline.CopyTo(buffer);
+
+            bytesWritten = DataPrefix.Length + Newline.Length + 1 + writtenForPayload + Newline.Length;
+            return true;
+        }
+
+        private static bool TryFormatPayload(ReadableBuffer payload, MessageType type, Span<byte> buffer, out int bytesWritten)
+        {
+            // Short-cut for empty payload
+            if (payload.Length == 0)
+            {
+                bytesWritten = 0;
+                return true;
+            }
+
+            var writtenSoFar = 0;
+            if (type == MessageType.Binary)
+            {
+                // TODO: We're going to need to fix this as part of https://github.com/aspnet/SignalR/issues/192
+                var message = Convert.ToBase64String(payload.ToArray());
+                var encodedSize = DataPrefix.Length + Encoding.UTF8.GetByteCount(message) + Newline.Length;
+                if (buffer.Length < encodedSize)
+                {
+                    bytesWritten = 0;
+                    return false;
+                }
+                DataPrefix.CopyTo(buffer);
+                buffer = buffer.Slice(DataPrefix.Length);
+
+                var array = Encoding.UTF8.GetBytes(message);
+                array.CopyTo(buffer);
+                buffer = buffer.Slice(array.Length);
+
+                Newline.CopyTo(buffer);
+                writtenSoFar += encodedSize;
+                buffer.Slice(Newline.Length);
+            }
+            else
+            {
+                while (true)
+                {
+                    // Seek to the end of buffer or newline
+                    var sliced = payload.TrySliceTo(LineFeed, out var slice, out var cursor);
+
+                    if (!TryFormatLine(sliced ? slice : payload, buffer, out var writtenByLine))
+                    {
+                        bytesWritten = 0;
+                        return false;
+                    }
+                    buffer = buffer.Slice(writtenByLine);
+                    writtenSoFar += writtenByLine;
+
+                    if (sliced)
+                    {
+                        payload = payload.Slice(payload.Move(cursor, 1));
+                    }
+                    else
+                    {
+                        break;
+                    }
+                }
+            }
+
+            bytesWritten = writtenSoFar;
+            return true;
+        }
+
+        private static bool TryFormatLine(ReadableBuffer slice, Span<byte> buffer, out int bytesWritten)
+        {
+            // We're going to write the whole thing. HOWEVER, if the last byte is a '\r', we want to truncate it
+            // because it was the '\r' in a '\r\n' newline sequence
+            // This won't require an additional byte in the buffer because after this line we have to write a newline sequence anyway.
+            var writtenSoFar = 0;
+            if (buffer.Length < DataPrefix.Length + slice.Length)
+            {
+                bytesWritten = 0;
+                return false;
+            }
+            DataPrefix.CopyTo(buffer);
+            writtenSoFar += DataPrefix.Length;
+            buffer = buffer.Slice(DataPrefix.Length);
+
+            slice.CopyTo(buffer);
+            var sliceTo = slice.Length;
+            if (sliceTo > 0 && buffer[sliceTo - 1] == '\r')
+            {
+                sliceTo -= 1;
+            }
+            writtenSoFar += sliceTo;
+            buffer = buffer.Slice(sliceTo);
+
+            if (buffer.Length < Newline.Length)
+            {
+                bytesWritten = 0;
+                return false;
+            }
+            writtenSoFar += Newline.Length;
+            Newline.CopyTo(buffer);
+
+            bytesWritten = writtenSoFar;
+            return true;
+        }
+
+        private static bool TryFormatType(Span<byte> buffer, MessageType type)
+        {
+            switch (type)
+            {
+                case MessageType.Text:
+                    buffer[0] = TextTypeFlag;
+                    return true;
+                case MessageType.Binary:
+                    buffer[0] = BinaryTypeFlag;
+                    return true;
+                case MessageType.Close:
+                    buffer[0] = CloseTypeFlag;
+                    return true;
+                case MessageType.Error:
+                    buffer[0] = ErrorTypeFlag;
+                    return true;
+                default:
+                    return false;
+            }
+        }
+    }
+}

src/Microsoft.AspNetCore.Sockets.Common/Formatters/TextMessageFormatter.cs

@@ -5,7 +5,7 @@ using System;
 using System.IO.Pipelines;
 using System.Text;
 
-namespace Microsoft.AspNetCore.Sockets
+namespace Microsoft.AspNetCore.Sockets.Formatters
 {
     internal static class TextMessageFormatter
     {

src/Microsoft.AspNetCore.Sockets.Common/MessageFormat.cs

@@ -1,10 +1,6 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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.Text;
-
 namespace Microsoft.AspNetCore.Sockets
 {
     public enum MessageFormat

src/Microsoft.AspNetCore.Sockets.Common/Microsoft.AspNetCore.Sockets.Common.csproj

@@ -8,6 +8,7 @@
     <NoWarn>$(NoWarn);CS1591</NoWarn>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
     <PackageTags>aspnetcore;signalr</PackageTags>
+    <RootNamespace>Microsoft.AspNetCore.Sockets</RootNamespace>
   </PropertyGroup>
 
   <ItemGroup>

test/Microsoft.AspNetCore.Sockets.Client.Tests/HubConnectionTests.cs

@@ -1,13 +1,16 @@
-using Microsoft.AspNetCore.SignalR.Tests.Common;
-using Microsoft.AspNetCore.Sockets.Client;
-using Microsoft.Extensions.Logging;
-using Moq;
-using Moq.Protected;
+// 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.Net;
 using System.Net.Http;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.SignalR.Tests.Common;
+using Microsoft.AspNetCore.Sockets.Client;
+using Microsoft.Extensions.Logging;
+using Moq;
+using Moq.Protected;
 using Xunit;
 
 namespace Microsoft.AspNetCore.SignalR.Client.Tests

test/Microsoft.AspNetCore.Sockets.Common.Tests/Formatters/BinaryMessageFormatterTests.cs

@@ -4,9 +4,10 @@
 using System;
 using System.Collections.Generic;
 using System.IO.Pipelines;
+using Microsoft.AspNetCore.Sockets.Tests;
 using Xunit;
 
-namespace Microsoft.AspNetCore.Sockets.Tests
+namespace Microsoft.AspNetCore.Sockets.Formatters.Tests
 {
     public partial class BinaryMessageFormatterTests
     {
@@ -167,5 +168,26 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             Assert.False(MessageFormatter.TryParseMessage(encoded, MessageFormat.Binary, out var message, out var consumed));
             Assert.Equal(0, consumed);
         }
+
+        [Fact]
+        public void InsufficientWriteBufferSpace()
+        {
+            const int ExpectedSize = 13;
+            var message = MessageTestUtils.CreateMessage("Test", MessageType.Text);
+
+            byte[] buffer;
+            int bufferSize;
+            int written;
+            for (bufferSize = 0; bufferSize < 13; bufferSize++)
+            {
+                buffer = new byte[bufferSize];
+                Assert.False(MessageFormatter.TryFormatMessage(message, buffer, MessageFormat.Binary, out written));
+                Assert.Equal(0, written);
+            }
+
+            buffer = new byte[bufferSize];
+            Assert.True(MessageFormatter.TryFormatMessage(message, buffer, MessageFormat.Binary, out written));
+            Assert.Equal(ExpectedSize, written);
+        }
     }
 }

test/Microsoft.AspNetCore.Sockets.Common.Tests/Formatters/ServerSentEventsMessageFormatterTests.cs

@@ -0,0 +1,86 @@
+// 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.IO.Pipelines;
+using System.Text;
+using Microsoft.AspNetCore.Sockets.Tests;
+using Xunit;
+
+namespace Microsoft.AspNetCore.Sockets.Formatters.Tests
+{
+    public class ServerSentEventsMessageFormatterTests
+    {
+        [Fact]
+        public void InsufficientWriteBufferSpace()
+        {
+            const int ExpectedSize = 23;
+            var message = MessageTestUtils.CreateMessage("Test", MessageType.Text);
+
+            byte[] buffer;
+            int bufferSize;
+            int written;
+            for (bufferSize = 0; bufferSize < 23; bufferSize++)
+            {
+                buffer = new byte[bufferSize];
+                Assert.False(ServerSentEventsMessageFormatter.TryFormatMessage(message, buffer, out written));
+                Assert.Equal(0, written);
+            }
+
+            buffer = new byte[bufferSize];
+            Assert.True(ServerSentEventsMessageFormatter.TryFormatMessage(message, buffer, out written));
+            Assert.Equal(ExpectedSize, written);
+        }
+
+        [Fact]
+        public void WriteInvalidMessages()
+        {
+            var message = new Message(ReadableBuffer.Create(new byte[0]).Preserve(), MessageType.Binary, endOfMessage: false);
+            var ex = Assert.Throws<InvalidOperationException>(() =>
+                ServerSentEventsMessageFormatter.TryFormatMessage(message, Span<byte>.Empty, out var written));
+            Assert.Equal("Cannot format message where endOfMessage is false using this format", ex.Message);
+        }
+
+        [Theory]
+        [InlineData("data: T\r\n\r\n", MessageType.Text, "")]
+        [InlineData("data: T\r\ndata: Hello, World\r\n\r\n", MessageType.Text, "Hello, World")]
+        [InlineData("data: T\r\ndata: Hello\r\ndata: World\r\n\r\n", MessageType.Text, "Hello\r\nWorld")]
+        [InlineData("data: T\r\ndata: Hello\r\ndata: World\r\n\r\n", MessageType.Text, "Hello\nWorld")]
+        [InlineData("data: T\r\ndata: Hello\r\ndata: \r\n\r\n", MessageType.Text, "Hello\n")]
+        [InlineData("data: T\r\ndata: Hello\r\ndata: \r\n\r\n", MessageType.Text, "Hello\r\n")]
+        [InlineData("data: C\r\n\r\n", MessageType.Close, "")]
+        [InlineData("data: C\r\ndata: Hello, World\r\n\r\n", MessageType.Close, "Hello, World")]
+        [InlineData("data: C\r\ndata: Hello\r\ndata: World\r\n\r\n", MessageType.Close, "Hello\r\nWorld")]
+        [InlineData("data: C\r\ndata: Hello\r\ndata: World\r\n\r\n", MessageType.Close, "Hello\nWorld")]
+        [InlineData("data: C\r\ndata: Hello\r\ndata: \r\n\r\n", MessageType.Close, "Hello\n")]
+        [InlineData("data: C\r\ndata: Hello\r\ndata: \r\n\r\n", MessageType.Close, "Hello\r\n")]
+        [InlineData("data: E\r\n\r\n", MessageType.Error, "")]
+        [InlineData("data: E\r\ndata: Hello, World\r\n\r\n", MessageType.Error, "Hello, World")]
+        [InlineData("data: E\r\ndata: Hello\r\ndata: World\r\n\r\n", MessageType.Error, "Hello\r\nWorld")]
+        [InlineData("data: E\r\ndata: Hello\r\ndata: World\r\n\r\n", MessageType.Error, "Hello\nWorld")]
+        [InlineData("data: E\r\ndata: Hello\r\ndata: \r\n\r\n", MessageType.Error, "Hello\n")]
+        [InlineData("data: E\r\ndata: Hello\r\ndata: \r\n\r\n", MessageType.Error, "Hello\r\n")]
+        public void WriteTextMessage(string encoded, MessageType messageType, string payload)
+        {
+            var message = MessageTestUtils.CreateMessage(payload, messageType);
+
+            var buffer = new byte[256];
+            Assert.True(ServerSentEventsMessageFormatter.TryFormatMessage(message, buffer, out var written));
+
+            Assert.Equal(encoded, Encoding.UTF8.GetString(buffer, 0, written));
+        }
+
+        [Theory]
+        [InlineData("data: B\r\n\r\n", new byte[0])]
+        [InlineData("data: B\r\ndata: q83v\r\n\r\n", new byte[] { 0xAB, 0xCD, 0xEF })]
+        public void WriteBinaryMessage(string encoded, byte[] payload)
+        {
+            var message = MessageTestUtils.CreateMessage(payload);
+
+            var buffer = new byte[256];
+            Assert.True(ServerSentEventsMessageFormatter.TryFormatMessage(message, buffer, out var written));
+
+            Assert.Equal(encoded, Encoding.UTF8.GetString(buffer, 0, written));
+        }
+    }
+}

test/Microsoft.AspNetCore.Sockets.Common.Tests/Formatters/TextMessageFormatterTests.cs

@@ -5,9 +5,10 @@ using System;
 using System.Collections.Generic;
 using System.IO.Pipelines;
 using System.Text;
+using Microsoft.AspNetCore.Sockets.Tests;
 using Xunit;
 
-namespace Microsoft.AspNetCore.Sockets.Tests
+namespace Microsoft.AspNetCore.Sockets.Formatters.Tests
 {
     public class TextMessageFormatterTests
     {
@@ -152,5 +153,26 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             Assert.False(MessageFormatter.TryParseMessage(buffer, MessageFormat.Text, out var message, out var consumed));
             Assert.Equal(0, consumed);
         }
+
+        [Fact]
+        public void InsufficientWriteBufferSpace()
+        {
+            const int ExpectedSize = 9;
+            var message = MessageTestUtils.CreateMessage("Test", MessageType.Text);
+
+            byte[] buffer;
+            int bufferSize;
+            int written;
+            for (bufferSize = 0; bufferSize < 9; bufferSize++)
+            {
+                buffer = new byte[bufferSize];
+                Assert.False(MessageFormatter.TryFormatMessage(message, buffer, MessageFormat.Text, out written));
+                Assert.Equal(0, written);
+            }
+
+            buffer = new byte[bufferSize];
+            Assert.True(MessageFormatter.TryFormatMessage(message, buffer, MessageFormat.Text, out written));
+            Assert.Equal(ExpectedSize, written);
+        }
     }
 }

test/Microsoft.AspNetCore.Sockets.Common.Tests/MessageTestUtils.cs

@@ -1,4 +1,7 @@
-using System.IO.Pipelines;
+// 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.IO.Pipelines;
 using System.Text;
 using Xunit;