From 3be56094aaa1ee48b1b9ea9dcd5f83e2cbf68635 Mon Sep 17 00:00:00 2001 From: MihaZupan Date: Sat, 29 Aug 2026 22:32:18 +0200 Subject: [PATCH] Use ArrayBuffer in SseParser, deduplicate more logic --- .../Common/src/System/Net/ArrayBuffer.cs | 12 +- .../Text/Json/PooledByteBufferWriter.cs | 2 +- .../Zstandard/ZstandardStream.Decompress.cs | 2 +- .../SocketsHttpHandler/Http2Connection.cs | 2 +- .../SocketsHttpHandler/Http3RequestStream.cs | 6 +- .../Http/SocketsHttpHandler/HttpConnection.cs | 6 +- .../Pal.Android/SafeDeleteSslContext.cs | 2 +- .../Security/Pal.OSX/SafeDeleteSslContext.cs | 2 +- .../System/Net/Security/TlsSession.OpenSsl.cs | 5 +- .../PooledByteBufferWriter.cs | 2 +- .../Net/ServerSentEvents/SseParser_1.cs | 316 ++++++------------ 11 files changed, 130 insertions(+), 227 deletions(-) diff --git a/src/libraries/Common/src/System/Net/ArrayBuffer.cs b/src/libraries/Common/src/System/Net/ArrayBuffer.cs index e07ff3ea333ceb..c99c8a943eea06 100644 --- a/src/libraries/Common/src/System/Net/ArrayBuffer.cs +++ b/src/libraries/Common/src/System/Net/ArrayBuffer.cs @@ -59,8 +59,7 @@ public ArrayBuffer(byte[] buffer) public void Dispose() { - _activeStart = 0; - _availableStart = 0; + DiscardAll(); byte[] array = _bytes; _bytes = null!; @@ -77,8 +76,7 @@ public void ClearAndReturnBuffer() Debug.Assert(_usePool); Debug.Assert(_bytes is not null); - _activeStart = 0; - _availableStart = 0; + DiscardAll(); byte[] bufferToReturn = _bytes; _bytes = Array.Empty(); @@ -112,6 +110,12 @@ public void Discard(int byteCount) } } + public void DiscardAll() + { + _activeStart = 0; + _availableStart = 0; + } + public void Commit(int byteCount) { Debug.Assert(byteCount <= AvailableLength); diff --git a/src/libraries/Common/src/System/Text/Json/PooledByteBufferWriter.cs b/src/libraries/Common/src/System/Text/Json/PooledByteBufferWriter.cs index 3b3ad2cce57fb8..7b27b0c12154ed 100644 --- a/src/libraries/Common/src/System/Text/Json/PooledByteBufferWriter.cs +++ b/src/libraries/Common/src/System/Text/Json/PooledByteBufferWriter.cs @@ -33,7 +33,7 @@ public PooledByteBufferWriter(int initialCapacity, Stream stream) : this(initial public int Capacity => _buffer.Capacity; - public void Clear() => _buffer.Discard(_buffer.ActiveLength); + public void Clear() => _buffer.DiscardAll(); public void ClearAndReturnBuffers() => _buffer.ClearAndReturnBuffer(); diff --git a/src/libraries/System.IO.Compression/src/System/IO/Compression/Zstandard/ZstandardStream.Decompress.cs b/src/libraries/System.IO.Compression/src/System/IO/Compression/Zstandard/ZstandardStream.Decompress.cs index 4b70ff2a419cdd..a963a083e0a010 100644 --- a/src/libraries/System.IO.Compression/src/System/IO/Compression/Zstandard/ZstandardStream.Decompress.cs +++ b/src/libraries/System.IO.Compression/src/System/IO/Compression/Zstandard/ZstandardStream.Decompress.cs @@ -436,7 +436,7 @@ private void TryRewindStream(Stream stream) { // Rewind the stream to the exact end of the compressed data stream.Seek(-unconsumedBytes, SeekOrigin.Current); - _buffer.Discard(unconsumedBytes); + _buffer.DiscardAll(); } } diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http2Connection.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http2Connection.cs index fdc6771d0b1d82..3f5659fb7a413b 100644 --- a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http2Connection.cs +++ b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http2Connection.cs @@ -409,7 +409,7 @@ private async Task FlushOutgoingBytesAsync() } _lastPendingWriterShouldFlush = false; - _outgoingBuffer.Discard(_outgoingBuffer.ActiveLength); + _outgoingBuffer.DiscardAll(); } } diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3RequestStream.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3RequestStream.cs index 9b6fb57daa84e1..cca6e545a0d01a 100644 --- a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3RequestStream.cs +++ b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3RequestStream.cs @@ -558,7 +558,7 @@ private async ValueTask WriteRequestContentAsync(ReadOnlyMemory buffer, Ca await _stream.WriteAsync(_sendBuffer.ActiveMemory, cancellationToken).ConfigureAwait(false); await _stream.WriteAsync(buffer, cancellationToken).ConfigureAwait(false); - _sendBuffer.Discard(_sendBuffer.ActiveLength); + _sendBuffer.DiscardAll(); _singleDataFrameWritten = true; } @@ -577,14 +577,14 @@ private async ValueTask WriteRequestContentAsync(ReadOnlyMemory buffer, Ca await _stream.WriteAsync(_sendBuffer.ActiveMemory, cancellationToken).ConfigureAwait(false); await _stream.WriteAsync(buffer, cancellationToken).ConfigureAwait(false); - _sendBuffer.Discard(_sendBuffer.ActiveLength); + _sendBuffer.DiscardAll(); } } private ValueTask FlushSendBufferAsync(bool endStream, CancellationToken cancellationToken) { ReadOnlyMemory toSend = _sendBuffer.ActiveMemory; - _sendBuffer.Discard(toSend.Length); + _sendBuffer.DiscardAll(); return _stream.WriteAsync(toSend, endStream, cancellationToken); } diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnection.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnection.cs index 4744c0af0d11c8..0f7d5ee916f72f 100644 --- a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnection.cs +++ b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnection.cs @@ -1552,7 +1552,7 @@ private void Flush() ReadOnlySpan bytes = _writeBuffer.ActiveSpan; if (bytes.Length > 0) { - _writeBuffer.Discard(bytes.Length); + _writeBuffer.DiscardAll(); WriteToStream(bytes); } } @@ -1562,7 +1562,7 @@ private ValueTask FlushAsync(bool async) ReadOnlyMemory bytes = _writeBuffer.ActiveMemory; if (bytes.Length > 0) { - _writeBuffer.Discard(bytes.Length); + _writeBuffer.DiscardAll(); return WriteToStreamAsync(bytes, async); } return default; @@ -2080,7 +2080,7 @@ private void CompleteResponse() Trace("Unexpected data on connection after response read."); } - _readBuffer.Discard(_readBuffer.ActiveLength); + _readBuffer.DiscardAll(); _connectionClose = true; } diff --git a/src/libraries/System.Net.Security/src/System/Net/Security/Pal.Android/SafeDeleteSslContext.cs b/src/libraries/System.Net.Security/src/System/Net/Security/Pal.Android/SafeDeleteSslContext.cs index aed5b1f56aa41e..5c64358225c8a4 100644 --- a/src/libraries/System.Net.Security/src/System/Net/Security/Pal.Android/SafeDeleteSslContext.cs +++ b/src/libraries/System.Net.Security/src/System/Net/Security/Pal.Android/SafeDeleteSslContext.cs @@ -186,7 +186,7 @@ internal void ReadPendingWrites(ref ProtocolToken token) } token.SetPayload(_outputBuffer.ActiveSpan); - _outputBuffer.Discard(_outputBuffer.ActiveLength); + _outputBuffer.DiscardAll(); } } diff --git a/src/libraries/System.Net.Security/src/System/Net/Security/Pal.OSX/SafeDeleteSslContext.cs b/src/libraries/System.Net.Security/src/System/Net/Security/Pal.OSX/SafeDeleteSslContext.cs index a26029c9f6849f..a2cdfccff59574 100644 --- a/src/libraries/System.Net.Security/src/System/Net/Security/Pal.OSX/SafeDeleteSslContext.cs +++ b/src/libraries/System.Net.Security/src/System/Net/Security/Pal.OSX/SafeDeleteSslContext.cs @@ -319,7 +319,7 @@ internal void ReadPendingWrites(ref ProtocolToken token) } token.SetPayload(_outputBuffer.ActiveSpan); - _outputBuffer.Discard(_outputBuffer.ActiveLength); + _outputBuffer.DiscardAll(); } } diff --git a/src/libraries/System.Net.Security/src/System/Net/Security/TlsSession.OpenSsl.cs b/src/libraries/System.Net.Security/src/System/Net/Security/TlsSession.OpenSsl.cs index 01e1b7fa6e3852..9dab3d38d412ed 100644 --- a/src/libraries/System.Net.Security/src/System/Net/Security/TlsSession.OpenSsl.cs +++ b/src/libraries/System.Net.Security/src/System/Net/Security/TlsSession.OpenSsl.cs @@ -92,10 +92,7 @@ partial void OnServerContextSet() _pendingFdSocket = null; // Discard any managed pre-fetch bytes; ownership transfers to the native BIO now. - if (_socketInBuffer.ActiveLength > 0) - { - _socketInBuffer.Discard(_socketInBuffer.ActiveLength); - } + _socketInBuffer.DiscardAll(); _useFdMode = true; } diff --git a/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/PooledByteBufferWriter.cs b/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/PooledByteBufferWriter.cs index 15df82ba2f2ce1..87fb6dd6d94964 100644 --- a/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/PooledByteBufferWriter.cs +++ b/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/PooledByteBufferWriter.cs @@ -28,7 +28,7 @@ public Span GetSpan(int sizeHint = 0) public ReadOnlyMemory WrittenMemory => _buffer.ActiveMemory; public int Capacity => _buffer.Capacity; public int WrittenCount => _buffer.ActiveLength; - public void Reset() => _buffer.Discard(_buffer.ActiveLength); + public void Reset() => _buffer.DiscardAll(); public void Dispose() => _buffer.Dispose(); } } diff --git a/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser_1.cs b/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser_1.cs index 9a74b758dc15d3..c7a6b219d018dc 100644 --- a/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser_1.cs +++ b/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser_1.cs @@ -1,11 +1,9 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. -using System.Buffers; using System.Collections; using System.Collections.Generic; using System.Diagnostics; -using System.Diagnostics.CodeAnalysis; using System.Globalization; using System.IO; using System.Runtime.CompilerServices; @@ -56,15 +54,11 @@ public sealed class SseParser /// Indicates whether the enumerable has already been used for enumeration. private int _used; - /// Buffer, either empty or rented, containing the data being read from the stream while looking for the next line. - private byte[] _lineBuffer = []; - /// The starting offset of valid data in . - private int _lineOffset; - /// The length of valid data in , starting from . - private int _lineLength; - /// The index in where a newline ('\r', '\n', or "\r\n") was found. + /// Buffer containing the data being read from the stream while looking for the next line. + private ArrayBuffer _lineBuffer = new(initialSize: 0, usePool: true); + /// The index relative to the start of the line buffer's active region where a newline ('\r', '\n', or "\r\n") was found. private int _newlineIndex; - /// The index in of characters already checked for newlines. + /// The index relative to the start of the line buffer's active region of characters already checked for newlines. /// /// This is to avoid O(LineLength^2) behavior in the rare case where we have long lines that are built-up over multiple reads. /// We want to avoid re-checking the same characters we've already checked over and over again. @@ -73,12 +67,10 @@ public sealed class SseParser /// Set when eof has been reached in the stream. private bool _eof; - /// Rented buffer containing buffered data for the next event. - private byte[]? _dataBuffer; - /// The length of valid data in , starting from index 0. - private int _dataLength; + /// Buffer containing buffered data for the next event. + private ArrayBuffer _dataBuffer = new(initialSize: 0, usePool: true); /// Whether data has been appended to . - /// This can be different than != 0 if empty data was appended. + /// This can be different than != 0 if empty data was appended. private bool _dataAppended; private readonly int _maxBufferSize; @@ -115,53 +107,24 @@ public IEnumerable> Enumerate() // Validate that the parser is only used for one enumeration. ThrowIfNotFirstEnumeration(); - // Rent a line buffer. This will grow as needed. The line buffer is what's passed to the stream, - // so we want it to be large enough to reduce the number of reads we need to do when data is - // arriving quickly. (In debug, we use a smaller buffer to stress the growth and shifting logic.) - _lineBuffer = ArrayPool.Shared.Rent(DefaultArrayPoolRentSize); try { // Spec: "Event streams in this format must always be encoded as UTF-8". // Skip a UTF8 BOM if it exists at the beginning of the stream. (The BOM is defined as optional in the SSE grammar.) - while (FillLineBuffer() != 0 && _lineLength < Utf8Bom.Length) ; + while (FillLineBuffer() != 0 && _lineBuffer.ActiveLength < Utf8Bom.Length) ; SkipBomIfPresent(); // Process all events in the stream. while (true) { - // See if there's a complete line in data already read from the stream. Lines are permitted to - // end with CR, LF, or CRLF. Look for all of them and if we find one, process the line. However, - // if we only find a CR and it's at the end of the read data, don't process it now, as we want - // to process it together with an LF that might immediately follow, rather than treating them - // as two separate characters, in which case we'd incorrectly process the CR as a line by itself. - GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength); - _newlineIndex = _lineBuffer.AsSpan(searchOffset, searchLength).IndexOfAny(CR, LF); - if (_newlineIndex >= 0) + if (TryProcessLine(out SseItem? sseItem)) { - _lastSearchedForNewline = -1; - _newlineIndex += searchOffset; - if (_lineBuffer[_newlineIndex] is LF || // the newline is LF - _newlineIndex - _lineOffset + 1 < _lineLength || // we must have CR and we have whatever comes after it - _eof) // if we get here, we know we have a CR at the end of the buffer, so it's definitely the whole newline if we've hit EOF + if (sseItem.HasValue) { - // Process the line. - if (ProcessLine(out SseItem sseItem, out int advance)) - { - yield return sseItem; - } - - // Move past the line. - _lineOffset += advance; - _lineLength -= advance; - continue; + yield return sseItem.GetValueOrDefault(); } - } - else - { - // Record the last position searched for a newline. The next time we search, - // we'll search from here rather than from _lineOffset, in order to avoid searching - // the same characters again. - _lastSearchedForNewline = _lineOffset + _lineLength; + + continue; } // We've processed everything in the buffer we currently can, so if we've already read EOF, we're done. @@ -178,11 +141,8 @@ public IEnumerable> Enumerate() } finally { - ArrayPool.Shared.Return(_lineBuffer); - if (_dataBuffer is not null) - { - ArrayPool.Shared.Return(_dataBuffer); - } + _lineBuffer.Dispose(); + _dataBuffer.Dispose(); } } @@ -195,53 +155,24 @@ public async IAsyncEnumerable> EnumerateAsync([EnumeratorCancellation // Validate that the parser is only used for one enumeration. ThrowIfNotFirstEnumeration(); - // Rent a line buffer. This will grow as needed. The line buffer is what's passed to the stream, - // so we want it to be large enough to reduce the number of reads we need to do when data is - // arriving quickly. (In debug, we use a smaller buffer to stress the growth and shifting logic.) - _lineBuffer = ArrayPool.Shared.Rent(DefaultArrayPoolRentSize); try { // Spec: "Event streams in this format must always be encoded as UTF-8". // Skip a UTF8 BOM if it exists at the beginning of the stream. (The BOM is defined as optional in the SSE grammar.) - while (await FillLineBufferAsync(cancellationToken).ConfigureAwait(false) != 0 && _lineLength < Utf8Bom.Length) ; + while (await FillLineBufferAsync(cancellationToken).ConfigureAwait(false) != 0 && _lineBuffer.ActiveLength < Utf8Bom.Length) ; SkipBomIfPresent(); // Process all events in the stream. while (true) { - // See if there's a complete line in data already read from the stream. Lines are permitted to - // end with CR, LF, or CRLF. Look for all of them and if we find one, process the line. However, - // if we only find a CR and it's at the end of the read data, don't process it now, as we want - // to process it together with an LF that might immediately follow, rather than treating them - // as two separate characters, in which case we'd incorrectly process the CR as a line by itself. - GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength); - _newlineIndex = _lineBuffer.AsSpan(searchOffset, searchLength).IndexOfAny(CR, LF); - if (_newlineIndex >= 0) + if (TryProcessLine(out SseItem? sseItem)) { - _lastSearchedForNewline = -1; - _newlineIndex += searchOffset; - if (_lineBuffer[_newlineIndex] is LF || // newline is LF - _newlineIndex - _lineOffset + 1 < _lineLength || // newline is CR, and we have whatever comes after it - _eof) // if we get here, we know we have a CR at the end of the buffer, so it's definitely the whole newline if we've hit EOF + if (sseItem.HasValue) { - // Process the line. - if (ProcessLine(out SseItem sseItem, out int advance)) - { - yield return sseItem; - } - - // Move past the line. - _lineOffset += advance; - _lineLength -= advance; - continue; + yield return sseItem.GetValueOrDefault(); } - } - else - { - // Record the last position searched for a newline. The next time we search, - // we'll search from here rather than from _lineOffset, in order to avoid searching - // the same characters again. - _lastSearchedForNewline = searchOffset + searchLength; + + continue; } // We've processed everything in the buffer we currently can, so if we've already read EOF, we're done. @@ -258,101 +189,83 @@ public async IAsyncEnumerable> EnumerateAsync([EnumeratorCancellation } finally { - ArrayPool.Shared.Return(_lineBuffer); - if (_dataBuffer is not null) - { - ArrayPool.Shared.Return(_dataBuffer); - } + _lineBuffer.Dispose(); + _dataBuffer.Dispose(); } } - /// Gets the next index and length with which to perform a newline search. - private void GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength) + /// Tries to process a complete line from data already read from the stream. + /// The parsed item if processing the line dispatched an event; otherwise, . + /// if a complete line was processed; otherwise, . + private bool TryProcessLine(out SseItem? sseItem) { - if (_lastSearchedForNewline > _lineOffset) + // See if there's a complete line in data already read from the stream. Lines are permitted to + // end with CR, LF, or CRLF. Look for all of them and if we find one, process the line. However, + // if we only find a CR and it's at the end of the read data, don't process it now, as we want + // to process it together with an LF that might immediately follow, rather than treating them + // as two separate characters, in which case we'd incorrectly process the CR as a line by itself. + ReadOnlySpan lineBuffer = _lineBuffer.ActiveReadOnlySpan; + int searchOffset = Math.Max(_lastSearchedForNewline, 0); + _newlineIndex = lineBuffer.Slice(searchOffset).IndexOfAny(CR, LF); + if (_newlineIndex >= 0) { - searchOffset = _lastSearchedForNewline; - searchLength = _lineLength - (_lastSearchedForNewline - _lineOffset); + _lastSearchedForNewline = -1; + _newlineIndex += searchOffset; + if (lineBuffer[_newlineIndex] is LF || // the newline is LF + _newlineIndex + 1 < lineBuffer.Length || // we must have CR and we have whatever comes after it + _eof) // if we get here, we know we have a CR at the end of the buffer, so it's definitely the whole newline if we've hit EOF + { + // Process the line. + sseItem = ProcessLine(out SseItem item) ? item : null; + return true; + } } else { - searchOffset = _lineOffset; - searchLength = _lineLength; + // Record the last position searched for a newline. The next time we search, + // we'll search from here rather than from the beginning, in order to avoid searching + // the same characters again. + _lastSearchedForNewline = lineBuffer.Length; } - Debug.Assert(searchOffset >= _lineOffset, $"{searchOffset}, {_lineLength}"); - Debug.Assert(searchOffset <= _lineOffset + _lineLength, $"{searchOffset}, {_lineOffset}, {_lineLength}"); - Debug.Assert(searchOffset <= _lineBuffer.Length, $"{searchOffset}, {_lineBuffer.Length}"); - - Debug.Assert(searchLength >= 0, $"{searchLength}"); - Debug.Assert(searchLength <= _lineLength, $"{searchLength}, {_lineLength}"); - } - - private int GetNewLineLength() - { - Debug.Assert(_newlineIndex - _lineOffset < _lineLength, "Expected to be positioned at a non-empty newline"); - return _lineBuffer.AsSpan(_newlineIndex, _lineLength - (_newlineIndex - _lineOffset)).StartsWith(CRLF) ? 2 : 1; + sseItem = null; + return false; } - /// - /// If there's no room remaining in the line buffer, either shifts the contents - /// left or grows the buffer in order to make room for the next read. - /// - private void ShiftOrGrowLineBufferIfNecessary() + private int GetNewLineLength(ReadOnlySpan lineBuffer) { - // If data we've read is butting up against the end of the buffer and - // it's not taking up the entire buffer, slide what's there down to - // the beginning, making room to read more data into the buffer (since - // there's no newline in the data that's there). Otherwise, if the whole - // buffer is full, grow the buffer to accommodate more data, since, again, - // what's there doesn't contain a newline and thus a line is longer than - // the current buffer accommodates. - if (_lineOffset + _lineLength == _lineBuffer.Length) - { - if (_lineOffset != 0) - { - _lineBuffer.AsSpan(_lineOffset, _lineLength).CopyTo(_lineBuffer); - if (_lastSearchedForNewline >= 0) - { - _lastSearchedForNewline -= _lineOffset; - } - _lineOffset = 0; - } - else if (_lineLength == _lineBuffer.Length) - { - GrowBuffer(ref _lineBuffer, (uint)_lineBuffer.Length + 1); - } - } - - // Storage available for at least one byte - Debug.Assert(_lineOffset + _lineLength < _lineBuffer.Length); + Debug.Assert(_newlineIndex < lineBuffer.Length, "Expected to be positioned at a non-empty newline"); + return lineBuffer.Slice(_newlineIndex).StartsWith(CRLF) ? 2 : 1; } /// Processes a complete line from the SSE stream. /// The parsed item if the method returns true. - /// How many characters to advance in the line buffer. /// true if an SSE item was successfully parsed; otherwise, false. - private bool ProcessLine(out SseItem sseItem, out int advance) + private bool ProcessLine(out SseItem sseItem) { - ReadOnlySpan line = _lineBuffer.AsSpan(_lineOffset, _newlineIndex - _lineOffset); + ReadOnlySpan lineBuffer = _lineBuffer.ActiveReadOnlySpan; + ReadOnlySpan line = lineBuffer.Slice(0, _newlineIndex); // Spec: "If the line is empty (a blank line) Dispatch the event" if (line.IsEmpty) { - advance = GetNewLineLength(); + int advance = GetNewLineLength(lineBuffer); if (_dataAppended) { - T data = _itemParser(_eventType ?? SseParser.EventTypeDefault, _dataBuffer.AsSpan(0, _dataLength)); + T data = _itemParser(_eventType ?? SseParser.EventTypeDefault, _dataBuffer.ActiveReadOnlySpan); sseItem = new SseItem(data, _eventType) { EventId = _eventId, ReconnectionInterval = _nextReconnectionInterval }; _eventType = null; _eventId = null; _nextReconnectionInterval = null; - _dataLength = 0; + _dataBuffer.DiscardAll(); _dataAppended = false; + + _lineBuffer.Discard(advance); return true; } + _lineBuffer.Discard(advance); sseItem = default; return false; } @@ -391,32 +304,45 @@ private bool ProcessLine(out SseItem sseItem, out int advance) // into the data buffer and dispatching from there. if (!_dataAppended) { - int newlineLength = GetNewLineLength(); - ReadOnlySpan remainder = _lineBuffer.AsSpan(_newlineIndex + newlineLength, _lineLength - line.Length - newlineLength); + int newlineLength = GetNewLineLength(lineBuffer); + ReadOnlySpan remainder = lineBuffer.Slice(_newlineIndex + newlineLength); if (!remainder.IsEmpty && (remainder[0] is LF || (remainder[0] is CR && remainder.Length > 1))) { - advance = line.Length + newlineLength + (remainder.StartsWith(CRLF) ? 2 : 1); T data = _itemParser(_eventType ?? SseParser.EventTypeDefault, fieldValue); sseItem = new SseItem(data, _eventType) { EventId = _eventId, ReconnectionInterval = _nextReconnectionInterval }; _eventType = null; _eventId = null; _nextReconnectionInterval = null; + + _lineBuffer.Discard(line.Length + newlineLength + (remainder.StartsWith(CRLF) ? 2 : 1)); return true; } } // We need to copy the data from the line buffer to the data buffer. Make sure there's enough room. - GrowBuffer(ref _dataBuffer, (uint)_dataLength + (uint)_lineLength + 1); + int requiredAvailableSpace = lineBuffer.Length + 1; + if (_dataBuffer.AvailableLength < requiredAvailableSpace) + { + if (requiredAvailableSpace > _maxBufferSize - _dataBuffer.ActiveLength) + { + throw new InvalidDataException(SR.InvalidDataException_SseExceededMaxLength); + } + + _dataBuffer.EnsureAvailableSpace( + _dataBuffer.Capacity == 0 ? Math.Max(requiredAvailableSpace, DefaultArrayPoolRentSize) : requiredAvailableSpace); + } // Append a newline if there's already content in the buffer. // Then copy the field value to the data buffer + Span destination = _dataBuffer.AvailableSpan; + int bytesWritten = 0; if (_dataAppended) { - _dataBuffer[_dataLength++] = LF; + destination[bytesWritten++] = LF; } - fieldValue.CopyTo(_dataBuffer.AsSpan(_dataLength)); - _dataLength += fieldValue.Length; + fieldValue.CopyTo(destination.Slice(bytesWritten)); + _dataBuffer.Commit(bytesWritten + fieldValue.Length); _dataAppended = true; } else if (fieldName.SequenceEqual("event"u8)) @@ -458,7 +384,7 @@ private bool ProcessLine(out SseItem sseItem, out int advance) // Spec: "Otherwise, The field is ignored" } - advance = line.Length + GetNewLineLength(); + _lineBuffer.Discard(line.Length + GetNewLineLength(lineBuffer)); sseItem = default; return false; } @@ -488,19 +414,19 @@ private void ThrowIfNotFirstEnumeration() /// Reads data from the stream into the line buffer. private int FillLineBuffer() { - ShiftOrGrowLineBufferIfNecessary(); - - int offset = _lineOffset + _lineLength; + EnsureLineBufferAvailableSpace(); int bytesRead = _stream.Read( #if NET - _lineBuffer.AsSpan(offset)); + _lineBuffer.AvailableSpan); #else - _lineBuffer, offset, _lineBuffer.Length - offset); + _lineBuffer.DangerousGetUnderlyingBuffer(), + _lineBuffer.ActiveStartOffset + _lineBuffer.ActiveLength, + _lineBuffer.AvailableLength); #endif if (bytesRead > 0) { - _lineLength += bytesRead; + _lineBuffer.Commit(bytesRead); } else { @@ -514,14 +440,12 @@ private int FillLineBuffer() /// Reads data asynchronously from the stream into the line buffer. private async ValueTask FillLineBufferAsync(CancellationToken cancellationToken) { - ShiftOrGrowLineBufferIfNecessary(); - - int offset = _lineOffset + _lineLength; - int bytesRead = await _stream.ReadAsync(_lineBuffer.AsMemory(offset), cancellationToken).ConfigureAwait(false); + EnsureLineBufferAvailableSpace(); + int bytesRead = await _stream.ReadAsync(_lineBuffer.AvailableMemory, cancellationToken).ConfigureAwait(false); if (bytesRead > 0) { - _lineLength += bytesRead; + _lineBuffer.Commit(bytesRead); } else { @@ -532,50 +456,28 @@ private async ValueTask FillLineBufferAsync(CancellationToken cancellationT return bytesRead; } - /// Gets the UTF8 BOM. - private static ReadOnlySpan Utf8Bom => [0xEF, 0xBB, 0xBF]; - - /// Called at the beginning of processing to skip over an optional UTF8 byte order mark. - private void SkipBomIfPresent() + private void EnsureLineBufferAvailableSpace() { - Debug.Assert(_lineOffset == 0, $"Expected _lineOffset == 0, got {_lineOffset}"); - - if (_lineBuffer.AsSpan(0, _lineLength).StartsWith(Utf8Bom)) + if (_lineBuffer.AvailableLength == 0) { - _lineOffset += 3; - _lineLength -= 3; - } - } - - /// Grows the buffer, returning the existing one to the ArrayPool and renting an ArrayPool replacement. - private void GrowBuffer([NotNull] ref byte[]? buffer, uint minimumSize) - { - int currentSize = buffer?.Length ?? 0; - - uint preferredSize = (uint)currentSize * 2; - preferredSize = Math.Min(preferredSize, (uint)_maxBufferSize); - preferredSize = Math.Max(preferredSize, DefaultArrayPoolRentSize); - Debug.Assert(preferredSize <= int.MaxValue); - - if (minimumSize > _maxBufferSize) - { - throw new InvalidDataException(SR.InvalidDataException_SseExceededMaxLength); - } - Debug.Assert(minimumSize <= int.MaxValue); + if (_lineBuffer.ActiveLength >= _maxBufferSize) + { + throw new InvalidDataException(SR.InvalidDataException_SseExceededMaxLength); + } - if (buffer is not null && currentSize >= minimumSize) - { - return; + _lineBuffer.EnsureAvailableSpace(_lineBuffer.Capacity == 0 ? DefaultArrayPoolRentSize : 1); } + } - int rentedSize = Math.Max((int)preferredSize, (int)minimumSize); + /// Gets the UTF8 BOM. + private static ReadOnlySpan Utf8Bom => [0xEF, 0xBB, 0xBF]; - byte[]? toReturn = buffer; - buffer = ArrayPool.Shared.Rent(rentedSize); - if (toReturn is not null) + /// Called at the beginning of processing to skip over an optional UTF8 byte order mark. + private void SkipBomIfPresent() + { + if (_lineBuffer.ActiveReadOnlySpan.StartsWith(Utf8Bom)) { - Array.Copy(toReturn, buffer, toReturn.Length); - ArrayPool.Shared.Return(toReturn); + _lineBuffer.Discard(Utf8Bom.Length); } } }