Add configurable SSE parser buffer limit

src/libraries/System.Net.ServerSentEvents/ref/System.Net.ServerSentEvents.cs

@@ -28,6 +28,13 @@ public static partial class SseParser
         public const string EventTypeDefault = "message";
         public static System.Net.ServerSentEvents.SseParser<string> Create(System.IO.Stream sseStream) { throw null; }
         public static System.Net.ServerSentEvents.SseParser<T> Create<T>(System.IO.Stream sseStream, System.Net.ServerSentEvents.SseItemParser<T> itemParser) { throw null; }
+        public static System.Net.ServerSentEvents.SseParser<T> Create<T>(System.IO.Stream sseStream, System.Net.ServerSentEvents.SseParserOptions<T> options) { throw null; }
+    }
+    public sealed partial class SseParserOptions<T>
+    {
+        public SseParserOptions(System.Net.ServerSentEvents.SseItemParser<T> itemParser) { }
+        public System.Net.ServerSentEvents.SseItemParser<T> ItemParser { get { throw null; } }
+        public int MaxBufferSize { get { throw null; } set { } }
     }
     public sealed partial class SseParser<T>
     {

src/libraries/System.Net.ServerSentEvents/src/System.Net.ServerSentEvents.csproj

@@ -20,6 +20,7 @@ System.Net.ServerSentEvents.SseParser</PackageDescription>
     <Compile Include="System\Net\ServerSentEvents\SseItem.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseItemParser.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseParser.cs" />
+    <Compile Include="System\Net\ServerSentEvents\SseParserOptions.cs" />
     <Compile Include="System\Net\ServerSentEvents\ThrowHelper.cs" />
   </ItemGroup>
 

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser.cs

@@ -3,7 +3,6 @@
 
 using System.IO;
 using System.Text;
-
 namespace System.Net.ServerSentEvents
 {
     /// <summary>Provides a parser for parsing server-sent events.</summary>
@@ -32,19 +31,21 @@ public static SseParser<string> Create(Stream sseStream) =>
         /// <param name="itemParser">The parser to use to transform each payload of bytes into a data element.</param>
         /// <returns>The enumerable, which can be enumerated synchronously or asynchronously.</returns>
         /// <exception cref="ArgumentNullException"><paramref name="sseStream"/> or <paramref name="itemParser"/> is null.</exception>
-        public static SseParser<T> Create<T>(Stream sseStream, SseItemParser<T> itemParser)
-        {
-            if (sseStream is null)
-            {
-                ThrowHelper.ThrowArgumentNullException(nameof(sseStream));
-            }
+        public static SseParser<T> Create<T>(Stream sseStream, SseItemParser<T> itemParser) =>
+            Create(sseStream, new SseParserOptions<T>(itemParser));
 
-            if (itemParser is null)
-            {
-                ThrowHelper.ThrowArgumentNullException(nameof(itemParser));
-            }
+        /// <summary>Creates a parser for parsing a <paramref name="sseStream"/> of server-sent events into a sequence of <see cref="SseItem{T}"/> values.</summary>
+        /// <typeparam name="T">Specifies the type of data in each event.</typeparam>
+        /// <param name="sseStream">The stream containing the data to parse.</param>
+        /// <param name="options">The options to use when parsing the stream.</param>
+        /// <returns>The enumerable, which can be enumerated synchronously or asynchronously.</returns>
+        /// <exception cref="ArgumentNullException"><paramref name="sseStream"/> or <paramref name="options"/> is null.</exception>
+        public static SseParser<T> Create<T>(Stream sseStream, SseParserOptions<T> options)
+        {
+            ArgumentNullException.ThrowIfNull(sseStream);
+            ArgumentNullException.ThrowIfNull(options);
 
-            return new SseParser<T>(sseStream, itemParser);
+            return new SseParser<T>(sseStream, options);
         }
     }
 }

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParserOptions.cs

@@ -0,0 +1,37 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+namespace System.Net.ServerSentEvents
+{
+    /// <summary>Provides options for parsing server-sent events.</summary>
+    /// <typeparam name="T">Specifies the type of data parsed from an event.</typeparam>
+    public sealed class SseParserOptions<T>
+    {
+        /// <summary>Initializes a new instance of the <see cref="SseParserOptions{T}"/> class.</summary>
+        /// <param name="itemParser">The parser to use to transform each payload of bytes into a data element.</param>
+        /// <exception cref="ArgumentNullException"><paramref name="itemParser"/> is null.</exception>
+        public SseParserOptions(SseItemParser<T> itemParser)
+        {
+            ArgumentNullException.ThrowIfNull(itemParser);
+
+            ItemParser = itemParser;
+        }
+
+        /// <summary>Gets the parser to use to transform each payload of bytes into a data element.</summary>
+        public SseItemParser<T> ItemParser { get; }
+
+        /// <summary>Gets or sets the maximum buffer size, or -1 to use the default limit.</summary>
+        /// <exception cref="ArgumentOutOfRangeException">The value set is less than -1.</exception>
+        public int MaxBufferSize
+        {
+            get => _maxBufferSize;
+            set
+            {
+                ArgumentOutOfRangeException.ThrowIfLessThan(value, -1);
+                _maxBufferSize = value;
+            }
+        }
+
+        private int _maxBufferSize = -1;
+    }
+}

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser_1.cs

@@ -41,6 +41,9 @@ public sealed class SseParser<T>
             1024;
 #endif
 
+        /// <summary>The maximum amount of data buffered by default.</summary>
+        private const int DefaultMaxBufferSize = 1024 * 1024 * 1024;
+
         /// <summary>The stream to be parsed.</summary>
         private readonly Stream _stream;
         /// <summary>The parser delegate used to transform bytes into a <typeparamref name="T"/>.</summary>
@@ -74,7 +77,7 @@ public sealed class SseParser<T>
         /// <remarks>This can be different than <see cref="_dataLength"/> != 0 if empty data was appended.</remarks>
         private bool _dataAppended;
 
-        private int _maxBufferSize = 1024 * 1024 * 1024;
+        private readonly int _maxBufferSize;
 
         /// <summary>The event type for the next event.</summary>
         private string? _eventType;
@@ -87,11 +90,12 @@ public sealed class SseParser<T>
 
         /// <summary>Initialize the enumerable.</summary>
         /// <param name="stream">The stream to parse.</param>
-        /// <param name="itemParser">The function to use to parse payload bytes into a <typeparamref name="T"/>.</param>
-        internal SseParser(Stream stream, SseItemParser<T> itemParser)
+        /// <param name="options">The options to use to parse the stream.</param>
+        internal SseParser(Stream stream, SseParserOptions<T> options)
         {
             _stream = stream;
-            _itemParser = itemParser;
+            _itemParser = options.ItemParser;
+            _maxBufferSize = options.MaxBufferSize == -1 ? DefaultMaxBufferSize : options.MaxBufferSize;
         }
 
         /// <summary>Gets an enumerable of the server-sent events from this parser.</summary>

src/libraries/System.Net.ServerSentEvents/tests/SseFormatterTests.cs

@@ -176,7 +176,7 @@ public static async Task WriteLargeItems_DataWrittenSuccessfully()
 
             memoryStream.Position = 0;
             int count = 0;
-            foreach (SseItem<byte[]> item in SseParser.Create(memoryStream, (eventType, data) => data.ToArray()).Enumerate())
+            foreach (SseItem<byte[]> item in SseParser.Create(memoryStream, new SseParserOptions<byte[]>((eventType, data) => data.ToArray())).Enumerate())
             {
                 Assert.Equal(expected, item.Data);
                 count++;
@@ -201,7 +201,7 @@ public static async Task WriteAsync_ParserCanRoundtripJsonEvents()
             await SseFormatter.WriteAsync(GetItemsAsync(), stream, FormatJson);
 
             stream.Position = 0;
-            SseParser<MyPoco> parser = SseParser.Create(stream, ParseJson);
+            SseParser<MyPoco> parser = SseParser.Create(stream, new SseParserOptions<MyPoco>(ParseJson));
             await ValidateParseResults(parser.EnumerateAsync());
 
             async IAsyncEnumerable<SseItem<MyPoco>> GetItemsAsync()

src/libraries/System.Net.ServerSentEvents/tests/SseParserTests.cs

@@ -6,7 +6,6 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
-using System.Reflection;
 using System.Text;
 using System.Text.Json;
 using System.Text.Json.Serialization;
@@ -21,15 +20,23 @@ public partial class SseParserTests
         [Fact]
         public void Parse_InvalidArguments_Throws()
         {
-            AssertExtensions.Throws<ArgumentNullException>("sseStream", () => SseParser.Create(null));
-            AssertExtensions.Throws<ArgumentNullException>("sseStream", () => SseParser.Create(null, delegate { return ""; }));
-            AssertExtensions.Throws<ArgumentNullException>("itemParser", () => SseParser.Create<string>(Stream.Null, null));
+            AssertExtensions.Throws<ArgumentNullException>("itemParser", () => new SseParserOptions<string>(null));
+            AssertExtensions.Throws<ArgumentNullException>("sseStream", () => SseParser.Create<string>(null, new SseParserOptions<string>(delegate { return ""; })));
+            AssertExtensions.Throws<ArgumentNullException>("options", () => SseParser.Create<string>(Stream.Null, (SseParserOptions<string>)null));
+        }
+
+        [Fact]
+        public void Options_DefaultMaxBufferSize()
+        {
+            var options = new SseParserOptions<string>(delegate { return ""; });
+
+            Assert.Equal(-1, options.MaxBufferSize);
         }
 
         [Fact]
         public async Task Parse_Sync_SupportsOnlyOneEnumeration_Throws()
         {
-            SseParser<string> parser = SseParser.Create(Stream.Null);
+            SseParser<string> parser = CreateParser(Stream.Null);
             parser.Enumerate().GetEnumerator().MoveNext();
             var e = parser.Enumerate().GetEnumerator();
             var ea = parser.EnumerateAsync().GetAsyncEnumerator();
@@ -40,7 +47,7 @@ public async Task Parse_Sync_SupportsOnlyOneEnumeration_Throws()
         [Fact]
         public async Task Parse_Async_SupportsOnlyOneEnumeration_Throws()
         {
-            SseParser<string> parser = SseParser.Create(Stream.Null);
+            SseParser<string> parser = CreateParser(Stream.Null);
             await parser.EnumerateAsync().GetAsyncEnumerator().MoveNextAsync();
             var ea = parser.EnumerateAsync().GetAsyncEnumerator();
             var e = parser.Enumerate().GetEnumerator();
@@ -210,7 +217,7 @@ public async Task Parse_HtmlSpec_Example4(string newline, bool trickle, bool use
                 $"{newline}",
                 trickle);
 
-            SseParser<string> parser = SseParser.Create(stream);
+            SseParser<string> parser = CreateParser(stream);
             if (useAsync)
             {
                 Assert.Equal(string.Empty, parser.LastEventId);
@@ -266,7 +273,7 @@ public async Task Parse_HtmlSpec_Example4_InheritedIDs(string newline, bool tric
                 $"{newline}",
                 trickle);
 
-            SseParser<string> parser = SseParser.Create(stream);
+            SseParser<string> parser = CreateParser(stream);
             if (useAsync)
             {
                 Assert.Equal(string.Empty, parser.LastEventId);
@@ -416,7 +423,7 @@ public async Task Retry_SetsReconnectionInterval(string newline, bool trickle, b
                 $"{newline}",
                 trickle);
 
-            SseParser<string> parser = SseParser.Create(stream);
+            SseParser<string> parser = CreateParser(stream);
             Assert.Equal(Timeout.InfiniteTimeSpan, parser.ReconnectionInterval);
 
             if (useAsync)
@@ -699,7 +706,7 @@ public async Task Delegate_ThrowsException_Propagates(string newline, bool trick
         {
             using Stream stream = GetStream($"data: hello{newline}{newline}data:world{newline}{newline}", trickle);
 
-            SseParser<string> parser = SseParser.Create<string>(stream, (eventType, bytes) => throw new FormatException(Encoding.UTF8.GetString(bytes.ToArray())));
+            SseParser<string> parser = CreateParser<string>(stream, (eventType, bytes) => throw new FormatException(Encoding.UTF8.GetString(bytes.ToArray())));
 
             FormatException fe;
             if (useAsync)
@@ -723,7 +730,7 @@ public async Task Cancellation_Propagates(bool cancelEnumerator)
         {
             using Stream stream = GetStream($"data: hello\n\ndata:world\n\n", trickle: true);
 
-            SseParser<string> parser = SseParser.Create(stream);
+            SseParser<string> parser = CreateParser(stream);
 
             var cts = new CancellationTokenSource();
             cts.Cancel();
@@ -740,7 +747,7 @@ public void NonGenericEnumerator_ProducesExpectedItems()
         {
             using Stream stream = GetStream($"data: hello\n\ndata:world\n\n", trickle: false);
 
-            IEnumerable sse = SseParser.Create(stream).Enumerate();
+            IEnumerable sse = CreateParser(stream).Enumerate();
             IEnumerator e = sse.GetEnumerator();
 
             Assert.True(e.MoveNext());
@@ -807,7 +814,7 @@ public async Task ArrayPoolRental_PerItem(string newline, bool trickle, bool use
             int count = 0;
             if (useAsync)
             {
-                foreach (var e in SseParser.Create(stream, itemParser).Enumerate())
+                foreach (var e in CreateParser(stream, itemParser).Enumerate())
                 {
                     try
                     {
@@ -826,7 +833,7 @@ public async Task ArrayPoolRental_PerItem(string newline, bool trickle, bool use
             }
             else
             {
-                await foreach (var e in SseParser.Create(stream, itemParser).EnumerateAsync())
+                await foreach (var e in CreateParser(stream, itemParser).EnumerateAsync())
                 {
                     try
                     {
@@ -875,7 +882,7 @@ public async Task ArrayPoolRental_Closure(string newline, bool trickle, bool use
             int count = 0;
             if (useAsync)
             {
-                foreach (var e in SseParser.Create(stream, itemParser).Enumerate())
+                foreach (var e in CreateParser(stream, itemParser).Enumerate())
                 {
                     if ("[DONE]"u8.SequenceEqual(e.Data.Span))
                     {
@@ -886,7 +893,7 @@ public async Task ArrayPoolRental_Closure(string newline, bool trickle, bool use
             }
             else
             {
-                await foreach (var e in SseParser.Create(stream, itemParser).EnumerateAsync())
+                await foreach (var e in CreateParser(stream, itemParser).EnumerateAsync())
                 {
                     if ("[DONE]"u8.SequenceEqual(e.Data.Span))
                     {
@@ -905,18 +912,12 @@ public async Task ArrayPoolRental_Closure(string newline, bool trickle, bool use
         [MemberData(nameof(NewlineAsyncData))]
         public async Task Parse_LongLineCap_Throws(string newline, bool useAsync)
         {
-            // Temporary workaround until we expose limit in public API
-            void ReduceLineLengthLimit(SseParser<string> parser)
-            {
-                Type type = typeof(SseParser<string>);
-                var field = type.GetField("_maxBufferSize", BindingFlags.Instance | BindingFlags.NonPublic);
-                Assert.NotNull(field);
-                field.SetValue(parser, 10 * 1024);
-            }
-
             using Stream stream = new InfiniteLineStream($"data: shortline{newline}{newline}data: ");
-            var parser = SseParser.Create(stream);
-            ReduceLineLengthLimit(parser);
+            var options = new SseParserOptions<string>(static (_, bytes) => Encoding.UTF8.GetString(bytes.ToArray()))
+            {
+                MaxBufferSize = 10 * 1024
+            };
+            var parser = SseParser.Create(stream, options);
 
             if (useAsync)
             {
@@ -961,18 +962,18 @@ private static Stream GetStream(byte[] bytes, bool trickle) =>
 
         private static List<SseItem<string>> ReadAllEvents(Stream stream)
         {
-            return new List<SseItem<string>>(SseParser.Create(stream).Enumerate());
+            return new List<SseItem<string>>(CreateParser(stream).Enumerate());
         }
 
         private static List<SseItem<T>> ReadAllEvents<T>(Stream stream, SseItemParser<T> parser)
         {
-            return new List<SseItem<T>>(SseParser.Create(stream, parser).Enumerate());
+            return new List<SseItem<T>>(CreateParser(stream, parser).Enumerate());
         }
 
         private static async Task<List<SseItem<T>>> ReadAllEventsAsync<T>(Stream stream, SseItemParser<T> parser)
         {
             var list = new List<SseItem<T>>();
-            await foreach (SseItem<T> item in SseParser.Create(stream, parser).EnumerateAsync())
+            await foreach (SseItem<T> item in CreateParser(stream, parser).EnumerateAsync())
             {
                 list.Add(item);
             }
@@ -983,14 +984,20 @@ private static async Task<List<SseItem<T>>> ReadAllEventsAsync<T>(Stream stream,
         private static async Task<List<SseItem<string>>> ReadAllEventsAsync(Stream stream)
         {
             var list = new List<SseItem<string>>();
-            await foreach (SseItem<string> item in SseParser.Create(stream).EnumerateAsync())
+            await foreach (SseItem<string> item in CreateParser(stream).EnumerateAsync())
             {
                 list.Add(item);
             }
 
             return list;
         }
 
+        private static SseParser<string> CreateParser(Stream stream) =>
+            CreateParser(stream, static (_, bytes) => Encoding.UTF8.GetString(bytes.ToArray()));
+
+        private static SseParser<T> CreateParser<T>(Stream stream, SseItemParser<T> itemParser) =>
+            SseParser.Create(stream, new SseParserOptions<T>(itemParser));
+
         /// <summary>Stream where each read reads at most one byte and where every asynchronous operation yields.</summary>
         private sealed class TrickleStream : MemoryStream
         {