GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

openai/openai-dotnet

Public

mirrored from https://github.com/openai/openai-dotnetAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
OpenAI_2.1.0-beta.2

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

openai-dotnet/src/Utility

src/Utility/System.Net.ServerSentEvents.cs

623lines · modeblame

RawDownload
Latest commit unavailable.
674e0f77Stephen Toub2 years ago1// Licensed to the .NET Foundation under one or more agreements.
2// The .NET Foundation licenses this file to you under the MIT license.
3
4// This file contains a source copy of:
5// https://github.com/dotnet/runtime/tree/2bd15868f12ace7cee9999af61d5c130b2603f04/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents
6// Once the System.Net.ServerSentEvents package is available, this file should be removed and replaced with a package reference.
7//
8// The only changes made to this code from the original are:
9// - Enabled nullable reference types at file scope, and use a few null suppression operators to work around the lack of [NotNull]
10// - Put into a single file for ease of management (it should not be edited in this repo).
11// - Changed public types to be internal.
12// - Removed a use of a [NotNull] attribute to assist in netstandard2.0 compilation.
13// - Replaced a reference to a .resx string with an inline constant.
14
15#nullable enable
16
17using System.Buffers;
18using System.Collections.Generic;
19using System.Diagnostics;
20using System.Globalization;
21using System.IO;
22using System.Runtime.CompilerServices;
23using System.Text;
24using System.Threading.Tasks;
25using System.Threading;
26
27namespace System.Net.ServerSentEvents
28{
29/// <summary>Represents a server-sent event.</summary>
30/// <typeparam name="T">Specifies the type of data payload in the event.</typeparam>
31internal readonly struct SseItem<T>
32{
33/// <summary>Initializes the server-sent event.</summary>
34/// <param name="data">The event's payload.</param>
35/// <param name="eventType">The event's type.</param>
36public SseItem(T data, string eventType)
37{
38Data = data;
39EventType = eventType;
40}
41
42/// <summary>Gets the event's payload.</summary>
43public T Data { get; }
44
45/// <summary>Gets the event's type.</summary>
46public string EventType { get; }
47}
48
49/// <summary>Encapsulates a method for parsing the bytes payload of a server-sent event.</summary>
50/// <typeparam name="T">Specifies the type of the return value of the parser.</typeparam>
51/// <param name="eventType">The event's type.</param>
52/// <param name="data">The event's payload bytes.</param>
53/// <returns>The parsed <typeparamref name="T"/>.</returns>
54internal delegate T SseItemParser<out T>(string eventType, ReadOnlySpan<byte> data);
55
56/// <summary>Provides a parser for parsing server-sent events.</summary>
57internal static class SseParser
58{
59/// <summary>The default <see cref="SseItem{T}.EventType"/> ("message") for an event that did not explicitly specify a type.</summary>
60public const string EventTypeDefault = "message";
61
62/// <summary>Creates a parser for parsing a <paramref name="sseStream"/> of server-sent events into a sequence of <see cref="SseItem{String}"/> values.</summary>
63/// <param name="sseStream">The stream containing the data to parse.</param>
64/// <returns>
65/// The enumerable of strings, which may be enumerated synchronously or asynchronously. The strings
66/// are decoded from the UTF8-encoded bytes of the payload of each event.
67/// </returns>
68/// <exception cref="ArgumentNullException"><paramref name="sseStream"/> is null.</exception>
69/// <remarks>
70/// This overload has behavior equivalent to calling <see cref="Create{T}(Stream, SseItemParser{T})"/> with a delegate
71/// that decodes the data of each event using <see cref="Encoding.UTF8"/>'s GetString method.
72/// </remarks>
73public static SseParser<string> Create(Stream sseStream) =>
74Create(sseStream, static (_, bytes) => Utf8GetString(bytes));
75
76/// <summary>Creates a parser for parsing a <paramref name="sseStream"/> of server-sent events into a sequence of <see cref="SseItem{T}"/> values.</summary>
77/// <typeparam name="T">Specifies the type of data in each event.</typeparam>
78/// <param name="sseStream">The stream containing the data to parse.</param>
79/// <param name="itemParser">The parser to use to transform each payload of bytes into a data element.</param>
80/// <returns>The enumerable, which may be enumerated synchronously or asynchronously.</returns>
81/// <exception cref="ArgumentNullException"><paramref name="sseStream"/> is null.</exception>
82/// <exception cref="ArgumentNullException"><paramref name="itemParser"/> is null.</exception>
83public static SseParser<T> Create<T>(Stream sseStream, SseItemParser<T> itemParser) =>
84new SseParser<T>(
85sseStream ?? throw new ArgumentNullException(nameof(sseStream)),
86itemParser ?? throw new ArgumentNullException(nameof(itemParser)));
87
88/// <summary>Encoding.UTF8.GetString(bytes)</summary>
89internal static string Utf8GetString(ReadOnlySpan<byte> bytes)
90{
91#if NET
92return Encoding.UTF8.GetString(bytes);
93#else
94unsafe
95{
96fixed (byte* ptr = bytes)
97{
98return ptr is null ?
99string.Empty :
100Encoding.UTF8.GetString(ptr, bytes.Length);
101}
102}
103#endif
104}
105}
106
107/// <summary>Provides a parser for server-sent events information.</summary>
108/// <typeparam name="T">Specifies the type of data parsed from an event.</typeparam>
109internal sealed class SseParser<T>
110{
111// For reference:
112// Specification: https://html.spec.whatwg.org/multipage/server-sent-events.html#server-sent-events
113
114/// <summary>Carriage Return.</summary>
115private const byte CR = (byte)'\r';
116/// <summary>Line Feed.</summary>
117private const byte LF = (byte)'\n';
118/// <summary>Carriage Return Line Feed.</summary>
119private static ReadOnlySpan<byte> CRLF => "\r\n"u8;
120
121/// <summary>The default size of an ArrayPool buffer to rent.</summary>
122/// <remarks>Larger size used by default to minimize number of reads. Smaller size used in debug to stress growth/shifting logic.</remarks>
123private const int DefaultArrayPoolRentSize =
124#if DEBUG
12516;
126#else
1271024;
128#endif
129
130/// <summary>The stream to be parsed.</summary>
131private readonly Stream _stream;
132/// <summary>The parser delegate used to transform bytes into a <typeparamref name="T"/>.</summary>
133private readonly SseItemParser<T> _itemParser;
134
135/// <summary>Indicates whether the enumerable has already been used for enumeration.</summary>
136private int _used;
137
138/// <summary>Buffer, either empty or rented, containing the data being read from the stream while looking for the next line.</summary>
139private byte[] _lineBuffer = [];
140/// <summary>The starting offset of valid data in <see cref="_lineBuffer"/>.</summary>
141private int _lineOffset;
142/// <summary>The length of valid data in <see cref="_lineBuffer"/>, starting from <see cref="_lineOffset"/>.</summary>
143private int _lineLength;
144/// <summary>The index in <see cref="_lineBuffer"/> where a newline ('\r', '\n', or "\r\n") was found.</summary>
145private int _newlineIndex;
146/// <summary>The index in <see cref="_lineBuffer"/> of characters already checked for newlines.</summary>
147/// <remarks>
148/// This is to avoid O(LineLength^2) behavior in the rare case where we have long lines that are built-up over multiple reads.
149/// We want to avoid re-checking the same characters we've already checked over and over again.
150/// </remarks>
151private int _lastSearchedForNewline;
152/// <summary>Set when eof has been reached in the stream.</summary>
153private bool _eof;
154
155/// <summary>Rented buffer containing buffered data for the next event.</summary>
156private byte[]? _dataBuffer;
157/// <summary>The length of valid data in <see cref="_dataBuffer"/>, starting from index 0.</summary>
158private int _dataLength;
159/// <summary>Whether data has been appended to <see cref="_dataBuffer"/>.</summary>
160/// <remarks>This can be different than <see cref="_dataLength"/> != 0 if empty data was appended.</remarks>
161private bool _dataAppended;
162
163/// <summary>The event type for the next event.</summary>
164private string _eventType = SseParser.EventTypeDefault;
165
166/// <summary>Initialize the enumerable.</summary>
167/// <param name="stream">The stream to parse.</param>
168/// <param name="itemParser">The function to use to parse payload bytes into a <typeparamref name="T"/>.</param>
169internal SseParser(Stream stream, SseItemParser<T> itemParser)
170{
171_stream = stream;
172_itemParser = itemParser;
173}
174
175/// <summary>Gets an enumerable of the server-sent events from this parser.</summary>
176/// <exception cref="InvalidOperationException">The parser has already been enumerated. Such an exception may propagate out of a call to <see cref="IEnumerator.MoveNext"/>.</exception>
177public IEnumerable<SseItem<T>> Enumerate()
178{
179// Validate that the parser is only used for one enumeration.
180ThrowIfNotFirstEnumeration();
181
182// Rent a line buffer. This will grow as needed. The line buffer is what's passed to the stream,
183// so we want it to be large enough to reduce the number of reads we need to do when data is
184// arriving quickly. (In debug, we use a smaller buffer to stress the growth and shifting logic.)
185_lineBuffer = ArrayPool<byte>.Shared.Rent(DefaultArrayPoolRentSize);
186try
187{
188// Spec: "Event streams in this format must always be encoded as UTF-8".
189// Skip a UTF8 BOM if it exists at the beginning of the stream. (The BOM is defined as optional in the SSE grammar.)
190while (FillLineBuffer() != 0 && _lineLength < Utf8Bom.Length) ;
191SkipBomIfPresent();
192
193// Process all events in the stream.
194while (true)
195{
196// See if there's a complete line in data already read from the stream. Lines are permitted to
197// end with CR, LF, or CRLF. Look for all of them and if we find one, process the line. However,
198// if we only find a CR and it's at the end of the read data, don't process it now, as we want
199// to process it together with an LF that might immediately follow, rather than treating them
200// as two separate characters, in which case we'd incorrectly process the CR as a line by itself.
201GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength);
202_newlineIndex = _lineBuffer.AsSpan(searchOffset, searchLength).IndexOfAny(CR, LF);
203if (_newlineIndex >= 0)
204{
205_lastSearchedForNewline = -1;
206_newlineIndex += searchOffset;
207if (_lineBuffer[_newlineIndex] is LF || // the newline is LF
208_newlineIndex - _lineOffset + 1 < _lineLength || // we must have CR and we have whatever comes after it
209_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
210{
211// Process the line.
212if (ProcessLine(out SseItem<T> sseItem, out int advance))
213{
214yield return sseItem;
215}
216
217// Move past the line.
218_lineOffset += advance;
219_lineLength -= advance;
220continue;
221}
222}
223else
224{
225// Record the last position searched for a newline. The next time we search,
226// we'll search from here rather than from _lineOffset, in order to avoid searching
227// the same characters again.
228_lastSearchedForNewline = _lineOffset + _lineLength;
229}
230
231// We've processed everything in the buffer we currently can, so if we've already read EOF, we're done.
232if (_eof)
233{
234// Spec: "Once the end of the file is reached, any pending data must be discarded. (If the file ends in the middle of an
235// event, before the final empty line, the incomplete event is not dispatched.)"
236break;
237}
238
239// Read more data into the buffer.
240FillLineBuffer();
241}
242}
243finally
244{
245ArrayPool<byte>.Shared.Return(_lineBuffer);
246if (_dataBuffer is not null)
247{
248ArrayPool<byte>.Shared.Return(_dataBuffer);
249}
250}
251}
252
253/// <summary>Gets an asynchronous enumerable of the server-sent events from this parser.</summary>
254/// <param name="cancellationToken">The cancellation token to use to cancel the enumeration.</param>
255/// <exception cref="InvalidOperationException">The parser has already been enumerated. Such an exception may propagate out of a call to <see cref="IAsyncEnumerator{T}.MoveNextAsync"/>.</exception>
256/// <exception cref="OperationCanceledException">The enumeration was canceled. Such an exception may propagate out of a call to <see cref="IAsyncEnumerator{T}.MoveNextAsync"/>.</exception>
257public async IAsyncEnumerable<SseItem<T>> EnumerateAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
258{
259// Validate that the parser is only used for one enumeration.
260ThrowIfNotFirstEnumeration();
261
262// Rent a line buffer. This will grow as needed. The line buffer is what's passed to the stream,
263// so we want it to be large enough to reduce the number of reads we need to do when data is
264// arriving quickly. (In debug, we use a smaller buffer to stress the growth and shifting logic.)
265_lineBuffer = ArrayPool<byte>.Shared.Rent(DefaultArrayPoolRentSize);
266try
267{
268// Spec: "Event streams in this format must always be encoded as UTF-8".
269// Skip a UTF8 BOM if it exists at the beginning of the stream. (The BOM is defined as optional in the SSE grammar.)
270while (await FillLineBufferAsync(cancellationToken).ConfigureAwait(false) != 0 && _lineLength < Utf8Bom.Length) ;
271SkipBomIfPresent();
272
273// Process all events in the stream.
274while (true)
275{
276// See if there's a complete line in data already read from the stream. Lines are permitted to
277// end with CR, LF, or CRLF. Look for all of them and if we find one, process the line. However,
278// if we only find a CR and it's at the end of the read data, don't process it now, as we want
279// to process it together with an LF that might immediately follow, rather than treating them
280// as two separate characters, in which case we'd incorrectly process the CR as a line by itself.
281GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength);
282_newlineIndex = _lineBuffer.AsSpan(searchOffset, searchLength).IndexOfAny(CR, LF);
283if (_newlineIndex >= 0)
284{
285_lastSearchedForNewline = -1;
286_newlineIndex += searchOffset;
287if (_lineBuffer[_newlineIndex] is LF || // newline is LF
288_newlineIndex - _lineOffset + 1 < _lineLength || // newline is CR, and we have whatever comes after it
289_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
290{
291// Process the line.
292if (ProcessLine(out SseItem<T> sseItem, out int advance))
293{
294yield return sseItem;
295}
296
297// Move past the line.
298_lineOffset += advance;
299_lineLength -= advance;
300continue;
301}
302}
303else
304{
305// Record the last position searched for a newline. The next time we search,
306// we'll search from here rather than from _lineOffset, in order to avoid searching
307// the same characters again.
308_lastSearchedForNewline = searchOffset + searchLength;
309}
310
311// We've processed everything in the buffer we currently can, so if we've already read EOF, we're done.
312if (_eof)
313{
314// Spec: "Once the end of the file is reached, any pending data must be discarded. (If the file ends in the middle of an
315// event, before the final empty line, the incomplete event is not dispatched.)"
316break;
317}
318
319// Read more data into the buffer.
320await FillLineBufferAsync(cancellationToken).ConfigureAwait(false);
321}
322}
323finally
324{
325ArrayPool<byte>.Shared.Return(_lineBuffer);
326if (_dataBuffer is not null)
327{
328ArrayPool<byte>.Shared.Return(_dataBuffer);
329}
330}
331}
332
333/// <summary>Gets the next index and length with which to perform a newline search.</summary>
334private void GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength)
335{
336if (_lastSearchedForNewline > _lineOffset)
337{
338searchOffset = _lastSearchedForNewline;
339searchLength = _lineLength - (_lastSearchedForNewline - _lineOffset);
340}
341else
342{
343searchOffset = _lineOffset;
344searchLength = _lineLength;
345}
346
347Debug.Assert(searchOffset >= _lineOffset, $"{searchOffset}, {_lineLength}");
348Debug.Assert(searchOffset <= _lineOffset + _lineLength, $"{searchOffset}, {_lineOffset}, {_lineLength}");
349Debug.Assert(searchOffset <= _lineBuffer.Length, $"{searchOffset}, {_lineBuffer.Length}");
350
351Debug.Assert(searchLength >= 0, $"{searchLength}");
352Debug.Assert(searchLength <= _lineLength, $"{searchLength}, {_lineLength}");
353}
354
355private int GetNewLineLength()
356{
357Debug.Assert(_newlineIndex - _lineOffset < _lineLength, "Expected to be positioned at a non-empty newline");
358return _lineBuffer.AsSpan(_newlineIndex, _lineLength - (_newlineIndex - _lineOffset)).StartsWith(CRLF) ? 2 : 1;
359}
360
361/// <summary>
362/// If there's no room remaining in the line buffer, either shifts the contents
363/// left or grows the buffer in order to make room for the next read.
364/// </summary>
365private void ShiftOrGrowLineBufferIfNecessary()
366{
367// If data we've read is butting up against the end of the buffer and
368// it's not taking up the entire buffer, slide what's there down to
369// the beginning, making room to read more data into the buffer (since
370// there's no newline in the data that's there). Otherwise, if the whole
371// buffer is full, grow the buffer to accommodate more data, since, again,
372// what's there doesn't contain a newline and thus a line is longer than
373// the current buffer accommodates.
374if (_lineOffset + _lineLength == _lineBuffer.Length)
375{
376if (_lineOffset != 0)
377{
378_lineBuffer.AsSpan(_lineOffset, _lineLength).CopyTo(_lineBuffer);
379if (_lastSearchedForNewline >= 0)
380{
381_lastSearchedForNewline -= _lineOffset;
382}
383_lineOffset = 0;
384}
385else if (_lineLength == _lineBuffer.Length)
386{
387GrowBuffer(ref _lineBuffer!, _lineBuffer.Length * 2);
388}
389}
390}
391
392/// <summary>Processes a complete line from the SSE stream.</summary>
393/// <param name="sseItem">The parsed item if the method returns true.</param>
394/// <param name="advance">How many characters to advance in the line buffer.</param>
395/// <returns>true if an SSE item was successfully parsed; otherwise, false.</returns>
396private bool ProcessLine(out SseItem<T> sseItem, out int advance)
397{
398ReadOnlySpan<byte> line = _lineBuffer.AsSpan(_lineOffset, _newlineIndex - _lineOffset);
399
400// Spec: "If the line is empty (a blank line) Dispatch the event"
401if (line.IsEmpty)
402{
403advance = GetNewLineLength();
404
405if (_dataAppended)
406{
407sseItem = new SseItem<T>(_itemParser(_eventType, _dataBuffer.AsSpan(0, _dataLength)), _eventType);
408_eventType = SseParser.EventTypeDefault;
409_dataLength = 0;
410_dataAppended = false;
411return true;
412}
413
414sseItem = default;
415return false;
416}
417
418// Find the colon separating the field name and value.
419int colonPos = line.IndexOf((byte)':');
420ReadOnlySpan<byte> fieldName;
421ReadOnlySpan<byte> fieldValue;
422if (colonPos >= 0)
423{
424// Spec: "Collect the characters on the line before the first U+003A COLON character (:), and let field be that string."
425fieldName = line.Slice(0, colonPos);
426
427// Spec: "Collect the characters on the line after the first U+003A COLON character (:), and let value be that string.
428// If value starts with a U+0020 SPACE character, remove it from value."
429fieldValue = line.Slice(colonPos + 1);
430if (!fieldValue.IsEmpty && fieldValue[0] == (byte)' ')
431{
432fieldValue = fieldValue.Slice(1);
433}
434}
435else
436{
437// Spec: "using the whole line as the field name, and the empty string as the field value."
438fieldName = line;
439fieldValue = [];
440}
441
442if (fieldName.SequenceEqual("data"u8))
443{
444// Spec: "Append the field value to the data buffer, then append a single U+000A LINE FEED (LF) character to the data buffer."
445// Spec: "If the data buffer's last character is a U+000A LINE FEED (LF) character, then remove the last character from the data buffer."
446
447// If there's nothing currently in the data buffer and we can easily detect that this line is immediately followed by
448// an empty line, we can optimize it to just handle the data directly from the line buffer, rather than first copying
449// into the data buffer and dispatching from there.
450if (!_dataAppended)
451{
452int newlineLength = GetNewLineLength();
453ReadOnlySpan<byte> remainder = _lineBuffer.AsSpan(_newlineIndex + newlineLength, _lineLength - line.Length - newlineLength);
454if (!remainder.IsEmpty &&
455(remainder[0] is LF || (remainder[0] is CR && remainder.Length > 1)))
456{
457advance = line.Length + newlineLength + (remainder.StartsWith(CRLF) ? 2 : 1);
458sseItem = new SseItem<T>(_itemParser(_eventType, fieldValue), _eventType);
459_eventType = SseParser.EventTypeDefault;
460return true;
461}
462}
463
464// We need to copy the data from the data buffer to the line buffer. Make sure there's enough room.
465if (_dataBuffer is null || _dataLength + _lineLength + 1 > _dataBuffer.Length)
466{
467GrowBuffer(ref _dataBuffer, _dataLength + _lineLength + 1);
468}
469
470// Append a newline if there's already content in the buffer.
471// Then copy the field value to the data buffer
472if (_dataAppended)
473{
474_dataBuffer![_dataLength++] = LF;
475}
476fieldValue.CopyTo(_dataBuffer.AsSpan(_dataLength));
477_dataLength += fieldValue.Length;
478_dataAppended = true;
479}
480else if (fieldName.SequenceEqual("event"u8))
481{
482// Spec: "Set the event type buffer to field value."
483_eventType = SseParser.Utf8GetString(fieldValue);
484}
485else if (fieldName.SequenceEqual("id"u8))
486{
487// Spec: "If the field value does not contain U+0000 NULL, then set the last event ID buffer to the field value. Otherwise, ignore the field."
488if (fieldValue.IndexOf((byte)'\0') < 0)
489{
490// Note that fieldValue might be empty, in which case LastEventId will naturally be reset to the empty string. This is per spec.
491LastEventId = SseParser.Utf8GetString(fieldValue);
492}
493}
494else if (fieldName.SequenceEqual("retry"u8))
495{
496// Spec: "If the field value consists of only ASCII digits, then interpret the field value as an integer in base ten,
497// and set the event stream's reconnection time to that integer. Otherwise, ignore the field."
498if (long.TryParse(
499#if NET7_0_OR_GREATER
500fieldValue,
501#else
502SseParser.Utf8GetString(fieldValue),
503#endif
504NumberStyles.None, CultureInfo.InvariantCulture, out long milliseconds))
505{
506ReconnectionInterval = TimeSpan.FromMilliseconds(milliseconds);
507}
508}
509else
510{
511// We'll end up here if the line starts with a colon, producing an empty field name, or if the field name is otherwise unrecognized.
512// Spec: "If the line starts with a U+003A COLON character (:) Ignore the line."
513// Spec: "Otherwise, The field is ignored"
514}
515
516advance = line.Length + GetNewLineLength();
517sseItem = default;
518return false;
519}
520
521/// <summary>Gets the last event ID.</summary>
522/// <remarks>This value is updated any time a new last event ID is parsed. It is not reset between SSE items.</remarks>
523public string LastEventId { get; private set; } = string.Empty; // Spec: "must be initialized to the empty string"
524
525/// <summary>Gets the reconnection interval.</summary>
526/// <remarks>
527/// If no retry event was received, this defaults to <see cref="Timeout.InfiniteTimeSpan"/>, and it will only
528/// ever be <see cref="Timeout.InfiniteTimeSpan"/> in that situation. If a client wishes to retry, the server-sent
529/// events specification states that the interval may then be decided by the client implementation and should be a
530/// few seconds.
531/// </remarks>
532public TimeSpan ReconnectionInterval { get; private set; } = Timeout.InfiniteTimeSpan;
533
534/// <summary>Transitions the object to a used state, throwing if it's already been used.</summary>
535private void ThrowIfNotFirstEnumeration()
536{
537if (Interlocked.Exchange(ref _used, 1) != 0)
538{
539throw new InvalidOperationException("The enumerable may be enumerated only once.");
540}
541}
542
543/// <summary>Reads data from the stream into the line buffer.</summary>
544private int FillLineBuffer()
545{
546ShiftOrGrowLineBufferIfNecessary();
547
548int offset = _lineOffset + _lineLength;
549int bytesRead = _stream.Read(
550#if NET
551_lineBuffer.AsSpan(offset));
552#else
553_lineBuffer, offset, _lineBuffer.Length - offset);
554#endif
555
556if (bytesRead > 0)
557{
558_lineLength += bytesRead;
559}
560else
561{
562_eof = true;
563bytesRead = 0;
564}
565
566return bytesRead;
567}
568
569/// <summary>Reads data asynchronously from the stream into the line buffer.</summary>
570private async ValueTask<int> FillLineBufferAsync(CancellationToken cancellationToken)
571{
572ShiftOrGrowLineBufferIfNecessary();
573
574int offset = _lineOffset + _lineLength;
575int bytesRead = await
576#if NET
577_stream.ReadAsync(_lineBuffer.AsMemory(offset), cancellationToken)
578#else
579new ValueTask<int>(_stream.ReadAsync(_lineBuffer, offset, _lineBuffer.Length - offset, cancellationToken))
580#endif
581.ConfigureAwait(false);
582
583if (bytesRead > 0)
584{
585_lineLength += bytesRead;
586}
587else
588{
589_eof = true;
590bytesRead = 0;
591}
592
593return bytesRead;
594}
595
596/// <summary>Gets the UTF8 BOM.</summary>
597private static ReadOnlySpan<byte> Utf8Bom => [0xEF, 0xBB, 0xBF];
598
599/// <summary>Called at the beginning of processing to skip over an optional UTF8 byte order mark.</summary>
600private void SkipBomIfPresent()
601{
602Debug.Assert(_lineOffset == 0, $"Expected _lineOffset == 0, got {_lineOffset}");
603
604if (_lineBuffer.AsSpan(0, _lineLength).StartsWith(Utf8Bom))
605{
606_lineOffset += 3;
607_lineLength -= 3;
608}
609}
610
611/// <summary>Grows the buffer, returning the existing one to the ArrayPool and renting an ArrayPool replacement.</summary>
612private static void GrowBuffer(ref byte[]? buffer, int minimumLength)
613{
614byte[]? toReturn = buffer;
615buffer = ArrayPool<byte>.Shared.Rent(Math.Max(minimumLength, DefaultArrayPoolRentSize));
616if (toReturn is not null)
617{
618Array.Copy(toReturn, buffer, toReturn.Length);
619ArrayPool<byte>.Shared.Return(toReturn);
620}
621}
622}
623}