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 · modecode

RawDownload
Latest commit unavailable.
1// 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>
31 internal 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>
36 public SseItem(T data, string eventType)
37 {
38 Data = data;
39 EventType = eventType;
40 }
41
42 /// <summary>Gets the event's payload.</summary>
43 public T Data { get; }
44
45 /// <summary>Gets the event's type.</summary>
46 public 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>
54 internal delegate T SseItemParser<out T>(string eventType, ReadOnlySpan<byte> data);
55
56 /// <summary>Provides a parser for parsing server-sent events.</summary>
57 internal 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>
60 public 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>
73 public static SseParser<string> Create(Stream sseStream) =>
74 Create(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>
83 public static SseParser<T> Create<T>(Stream sseStream, SseItemParser<T> itemParser) =>
84 new SseParser<T>(
85 sseStream ?? throw new ArgumentNullException(nameof(sseStream)),
86 itemParser ?? throw new ArgumentNullException(nameof(itemParser)));
87
88 /// <summary>Encoding.UTF8.GetString(bytes)</summary>
89 internal static string Utf8GetString(ReadOnlySpan<byte> bytes)
90 {
91#if NET
92 return Encoding.UTF8.GetString(bytes);
93#else
94 unsafe
95 {
96 fixed (byte* ptr = bytes)
97 {
98 return ptr is null ?
99 string.Empty :
100 Encoding.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>
109 internal 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>
115 private const byte CR = (byte)'\r';
116 /// <summary>Line Feed.</summary>
117 private const byte LF = (byte)'\n';
118 /// <summary>Carriage Return Line Feed.</summary>
119 private 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>
123 private const int DefaultArrayPoolRentSize =
124#if DEBUG
125 16;
126#else
127 1024;
128#endif
129
130 /// <summary>The stream to be parsed.</summary>
131 private readonly Stream _stream;
132 /// <summary>The parser delegate used to transform bytes into a <typeparamref name="T"/>.</summary>
133 private readonly SseItemParser<T> _itemParser;
134
135 /// <summary>Indicates whether the enumerable has already been used for enumeration.</summary>
136 private 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>
139 private byte[] _lineBuffer = [];
140 /// <summary>The starting offset of valid data in <see cref="_lineBuffer"/>.</summary>
141 private int _lineOffset;
142 /// <summary>The length of valid data in <see cref="_lineBuffer"/>, starting from <see cref="_lineOffset"/>.</summary>
143 private int _lineLength;
144 /// <summary>The index in <see cref="_lineBuffer"/> where a newline ('\r', '\n', or "\r\n") was found.</summary>
145 private 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>
151 private int _lastSearchedForNewline;
152 /// <summary>Set when eof has been reached in the stream.</summary>
153 private bool _eof;
154
155 /// <summary>Rented buffer containing buffered data for the next event.</summary>
156 private byte[]? _dataBuffer;
157 /// <summary>The length of valid data in <see cref="_dataBuffer"/>, starting from index 0.</summary>
158 private 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>
161 private bool _dataAppended;
162
163 /// <summary>The event type for the next event.</summary>
164 private 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>
169 internal 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>
177 public IEnumerable<SseItem<T>> Enumerate()
178 {
179 // Validate that the parser is only used for one enumeration.
180 ThrowIfNotFirstEnumeration();
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);
186 try
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.)
190 while (FillLineBuffer() != 0 && _lineLength < Utf8Bom.Length) ;
191 SkipBomIfPresent();
192
193 // Process all events in the stream.
194 while (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.
201 GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength);
202 _newlineIndex = _lineBuffer.AsSpan(searchOffset, searchLength).IndexOfAny(CR, LF);
203 if (_newlineIndex >= 0)
204 {
205 _lastSearchedForNewline = -1;
206 _newlineIndex += searchOffset;
207 if (_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.
212 if (ProcessLine(out SseItem<T> sseItem, out int advance))
213 {
214 yield return sseItem;
215 }
216
217 // Move past the line.
218 _lineOffset += advance;
219 _lineLength -= advance;
220 continue;
221 }
222 }
223 else
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.
232 if (_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.)"
236 break;
237 }
238
239 // Read more data into the buffer.
240 FillLineBuffer();
241 }
242 }
243 finally
244 {
245 ArrayPool<byte>.Shared.Return(_lineBuffer);
246 if (_dataBuffer is not null)
247 {
248 ArrayPool<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>
257 public async IAsyncEnumerable<SseItem<T>> EnumerateAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
258 {
259 // Validate that the parser is only used for one enumeration.
260 ThrowIfNotFirstEnumeration();
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);
266 try
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.)
270 while (await FillLineBufferAsync(cancellationToken).ConfigureAwait(false) != 0 && _lineLength < Utf8Bom.Length) ;
271 SkipBomIfPresent();
272
273 // Process all events in the stream.
274 while (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.
281 GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength);
282 _newlineIndex = _lineBuffer.AsSpan(searchOffset, searchLength).IndexOfAny(CR, LF);
283 if (_newlineIndex >= 0)
284 {
285 _lastSearchedForNewline = -1;
286 _newlineIndex += searchOffset;
287 if (_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.
292 if (ProcessLine(out SseItem<T> sseItem, out int advance))
293 {
294 yield return sseItem;
295 }
296
297 // Move past the line.
298 _lineOffset += advance;
299 _lineLength -= advance;
300 continue;
301 }
302 }
303 else
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.
312 if (_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.)"
316 break;
317 }
318
319 // Read more data into the buffer.
320 await FillLineBufferAsync(cancellationToken).ConfigureAwait(false);
321 }
322 }
323 finally
324 {
325 ArrayPool<byte>.Shared.Return(_lineBuffer);
326 if (_dataBuffer is not null)
327 {
328 ArrayPool<byte>.Shared.Return(_dataBuffer);
329 }
330 }
331 }
332
333 /// <summary>Gets the next index and length with which to perform a newline search.</summary>
334 private void GetNextSearchOffsetAndLength(out int searchOffset, out int searchLength)
335 {
336 if (_lastSearchedForNewline > _lineOffset)
337 {
338 searchOffset = _lastSearchedForNewline;
339 searchLength = _lineLength - (_lastSearchedForNewline - _lineOffset);
340 }
341 else
342 {
343 searchOffset = _lineOffset;
344 searchLength = _lineLength;
345 }
346
347 Debug.Assert(searchOffset >= _lineOffset, $"{searchOffset}, {_lineLength}");
348 Debug.Assert(searchOffset <= _lineOffset + _lineLength, $"{searchOffset}, {_lineOffset}, {_lineLength}");
349 Debug.Assert(searchOffset <= _lineBuffer.Length, $"{searchOffset}, {_lineBuffer.Length}");
350
351 Debug.Assert(searchLength >= 0, $"{searchLength}");
352 Debug.Assert(searchLength <= _lineLength, $"{searchLength}, {_lineLength}");
353 }
354
355 private int GetNewLineLength()
356 {
357 Debug.Assert(_newlineIndex - _lineOffset < _lineLength, "Expected to be positioned at a non-empty newline");
358 return _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>
365 private 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.
374 if (_lineOffset + _lineLength == _lineBuffer.Length)
375 {
376 if (_lineOffset != 0)
377 {
378 _lineBuffer.AsSpan(_lineOffset, _lineLength).CopyTo(_lineBuffer);
379 if (_lastSearchedForNewline >= 0)
380 {
381 _lastSearchedForNewline -= _lineOffset;
382 }
383 _lineOffset = 0;
384 }
385 else if (_lineLength == _lineBuffer.Length)
386 {
387 GrowBuffer(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>
396 private bool ProcessLine(out SseItem<T> sseItem, out int advance)
397 {
398 ReadOnlySpan<byte> line = _lineBuffer.AsSpan(_lineOffset, _newlineIndex - _lineOffset);
399
400 // Spec: "If the line is empty (a blank line) Dispatch the event"
401 if (line.IsEmpty)
402 {
403 advance = GetNewLineLength();
404
405 if (_dataAppended)
406 {
407 sseItem = new SseItem<T>(_itemParser(_eventType, _dataBuffer.AsSpan(0, _dataLength)), _eventType);
408 _eventType = SseParser.EventTypeDefault;
409 _dataLength = 0;
410 _dataAppended = false;
411 return true;
412 }
413
414 sseItem = default;
415 return false;
416 }
417
418 // Find the colon separating the field name and value.
419 int colonPos = line.IndexOf((byte)':');
420 ReadOnlySpan<byte> fieldName;
421 ReadOnlySpan<byte> fieldValue;
422 if (colonPos >= 0)
423 {
424 // Spec: "Collect the characters on the line before the first U+003A COLON character (:), and let field be that string."
425 fieldName = 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."
429 fieldValue = line.Slice(colonPos + 1);
430 if (!fieldValue.IsEmpty && fieldValue[0] == (byte)' ')
431 {
432 fieldValue = fieldValue.Slice(1);
433 }
434 }
435 else
436 {
437 // Spec: "using the whole line as the field name, and the empty string as the field value."
438 fieldName = line;
439 fieldValue = [];
440 }
441
442 if (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.
450 if (!_dataAppended)
451 {
452 int newlineLength = GetNewLineLength();
453 ReadOnlySpan<byte> remainder = _lineBuffer.AsSpan(_newlineIndex + newlineLength, _lineLength - line.Length - newlineLength);
454 if (!remainder.IsEmpty &&
455 (remainder[0] is LF || (remainder[0] is CR && remainder.Length > 1)))
456 {
457 advance = line.Length + newlineLength + (remainder.StartsWith(CRLF) ? 2 : 1);
458 sseItem = new SseItem<T>(_itemParser(_eventType, fieldValue), _eventType);
459 _eventType = SseParser.EventTypeDefault;
460 return 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.
465 if (_dataBuffer is null || _dataLength + _lineLength + 1 > _dataBuffer.Length)
466 {
467 GrowBuffer(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
472 if (_dataAppended)
473 {
474 _dataBuffer![_dataLength++] = LF;
475 }
476 fieldValue.CopyTo(_dataBuffer.AsSpan(_dataLength));
477 _dataLength += fieldValue.Length;
478 _dataAppended = true;
479 }
480 else if (fieldName.SequenceEqual("event"u8))
481 {
482 // Spec: "Set the event type buffer to field value."
483 _eventType = SseParser.Utf8GetString(fieldValue);
484 }
485 else 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."
488 if (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.
491 LastEventId = SseParser.Utf8GetString(fieldValue);
492 }
493 }
494 else 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."
498 if (long.TryParse(
499#if NET7_0_OR_GREATER
500 fieldValue,
501#else
502 SseParser.Utf8GetString(fieldValue),
503#endif
504 NumberStyles.None, CultureInfo.InvariantCulture, out long milliseconds))
505 {
506 ReconnectionInterval = TimeSpan.FromMilliseconds(milliseconds);
507 }
508 }
509 else
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
516 advance = line.Length + GetNewLineLength();
517 sseItem = default;
518 return 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>
523 public 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>
532 public 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>
535 private void ThrowIfNotFirstEnumeration()
536 {
537 if (Interlocked.Exchange(ref _used, 1) != 0)
538 {
539 throw new InvalidOperationException("The enumerable may be enumerated only once.");
540 }
541 }
542
543 /// <summary>Reads data from the stream into the line buffer.</summary>
544 private int FillLineBuffer()
545 {
546 ShiftOrGrowLineBufferIfNecessary();
547
548 int offset = _lineOffset + _lineLength;
549 int bytesRead = _stream.Read(
550#if NET
551 _lineBuffer.AsSpan(offset));
552#else
553 _lineBuffer, offset, _lineBuffer.Length - offset);
554#endif
555
556 if (bytesRead > 0)
557 {
558 _lineLength += bytesRead;
559 }
560 else
561 {
562 _eof = true;
563 bytesRead = 0;
564 }
565
566 return bytesRead;
567 }
568
569 /// <summary>Reads data asynchronously from the stream into the line buffer.</summary>
570 private async ValueTask<int> FillLineBufferAsync(CancellationToken cancellationToken)
571 {
572 ShiftOrGrowLineBufferIfNecessary();
573
574 int offset = _lineOffset + _lineLength;
575 int bytesRead = await
576#if NET
577 _stream.ReadAsync(_lineBuffer.AsMemory(offset), cancellationToken)
578#else
579 new ValueTask<int>(_stream.ReadAsync(_lineBuffer, offset, _lineBuffer.Length - offset, cancellationToken))
580#endif
581 .ConfigureAwait(false);
582
583 if (bytesRead > 0)
584 {
585 _lineLength += bytesRead;
586 }
587 else
588 {
589 _eof = true;
590 bytesRead = 0;
591 }
592
593 return bytesRead;
594 }
595
596 /// <summary>Gets the UTF8 BOM.</summary>
597 private 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>
600 private void SkipBomIfPresent()
601 {
602 Debug.Assert(_lineOffset == 0, $"Expected _lineOffset == 0, got {_lineOffset}");
603
604 if (_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>
612 private static void GrowBuffer(ref byte[]? buffer, int minimumLength)
613 {
614 byte[]? toReturn = buffer;
615 buffer = ArrayPool<byte>.Shared.Rent(Math.Max(minimumLength, DefaultArrayPoolRentSize));
616 if (toReturn is not null)
617 {
618 Array.Copy(toReturn, buffer, toReturn.Length);
619 ArrayPool<byte>.Shared.Return(toReturn);
620 }
621 }
622 }
623}