-
Notifications
You must be signed in to change notification settings - Fork 5.5k
Add Stream wrappers for memory and text-based types #126669
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
ViveliDuCh
wants to merge
12
commits into
dotnet:main
Choose a base branch
from
ViveliDuCh:stream-wrappers-api
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
12 commits
Select commit
Hold shift + click to select a range
1f1bfe1
First API proposal implementation
ViveliDuCh f7a06f0
API replacements in production code
ViveliDuCh bd5cf98
Implementation update based on latest API Review final consensus
ViveliDuCh fec7a59
Address PR feedback. MemoryStream base change. Spillover buffer for C…
ViveliDuCh 341ad57
Address PR feedback
ViveliDuCh 8882b05
Merge branch 'dotnet:main' into stream-wrappers-api
ViveliDuCh 0fb2ef5
Merge branch 'dotnet:main' into stream-wrappers-api
ViveliDuCh 111aa77
Address PR feedback: fix StringStream flush offset, WritableMemoryStr…
ViveliDuCh 430515a
Merge branch 'dotnet:main' into stream-wrappers-api
ViveliDuCh 0946bfc
Address PR feedback: remove redundant code, improve test coverage and…
ViveliDuCh d5103d1
Merge remote-tracking branch
ViveliDuCh 8e4a441
Address PR feedback: gap-zeroing, disposal checks, cancellation guard…
ViveliDuCh File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Some comments aren't visible on the classic Files Changed page.
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
277 changes: 277 additions & 0 deletions
277
src/libraries/System.Memory/src/System/Buffers/ReadOnlySequenceStream.cs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,277 @@ | ||
| // Licensed to the .NET Foundation under one or more agreements. | ||
| // The .NET Foundation licenses this file to you under the MIT license. | ||
| using System.Threading; | ||
| using System.IO; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace System.Buffers | ||
| { | ||
| /// <summary> | ||
| /// Provides a seekable, read-only <see cref="Stream"/> implementation over a <see cref="ReadOnlySequence{T}"/> of bytes. | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// This type is not thread-safe. Synchronize access if the stream is used concurrently. | ||
| /// The underlying sequence should not be modified while the stream is in use. | ||
| /// Seeking beyond the end of the stream is supported; subsequent reads will return zero bytes. | ||
| /// </remarks> | ||
| public sealed class ReadOnlySequenceStream : Stream | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
| { | ||
| private ReadOnlySequence<byte> _sequence; | ||
| private SequencePosition _position; | ||
| private long _absolutePosition; | ||
| private bool _isDisposed; | ||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="ReadOnlySequenceStream"/> class over the specified <see cref="ReadOnlySequence{Byte}"/>. | ||
| /// </summary> | ||
| /// <param name="source">The <see cref="ReadOnlySequence{Byte}"/> to wrap.</param> | ||
| public ReadOnlySequenceStream(ReadOnlySequence<byte> source) | ||
| { | ||
| _sequence = source; | ||
| _position = source.Start; | ||
| _absolutePosition = 0; | ||
| _isDisposed = false; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override bool CanRead => !_isDisposed; | ||
|
|
||
| /// <inheritdoc /> | ||
| public override bool CanSeek => !_isDisposed; | ||
|
|
||
| /// <inheritdoc /> | ||
| public override bool CanWrite => false; | ||
|
|
||
| private void EnsureNotDisposed() => ObjectDisposedException.ThrowIf(_isDisposed, this); | ||
|
|
||
| /// <inheritdoc /> | ||
| public override long Length | ||
| { | ||
| get | ||
| { | ||
| EnsureNotDisposed(); | ||
| return _sequence.Length; | ||
| } | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override long Position | ||
| { | ||
| get | ||
| { | ||
| EnsureNotDisposed(); | ||
| return _absolutePosition; | ||
| } | ||
| set | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
| { | ||
| EnsureNotDisposed(); | ||
| ArgumentOutOfRangeException.ThrowIfNegative(value); | ||
|
|
||
| if (value >= _sequence.Length) | ||
| { | ||
| _position = _sequence.End; | ||
| } | ||
| else if (value >= _absolutePosition) | ||
| { | ||
| _position = _sequence.GetPosition(value - _absolutePosition, _position); | ||
| } | ||
| else | ||
| { | ||
| _position = _sequence.GetPosition(value, _sequence.Start); | ||
| } | ||
|
|
||
| _absolutePosition = value; | ||
| } | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override int Read(byte[] buffer, int offset, int count) | ||
| { | ||
| ValidateBufferArguments(buffer, offset, count); | ||
| return Read(buffer.AsSpan(offset, count)); | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override int Read(Span<byte> buffer) | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
| { | ||
| EnsureNotDisposed(); | ||
|
|
||
| if (_absolutePosition >= _sequence.Length) | ||
| { | ||
| return 0; | ||
| } | ||
|
|
||
| ReadOnlySequence<byte> remaining = _sequence.Slice(_position); | ||
| int n = (int)Math.Min(remaining.Length, buffer.Length); | ||
| if (n <= 0) | ||
| { | ||
| return 0; | ||
| } | ||
|
|
||
| remaining.Slice(0, n).CopyTo(buffer); | ||
| _position = _sequence.GetPosition(n, _position); | ||
| _absolutePosition += n; | ||
| return n; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override int ReadByte() | ||
| { | ||
| EnsureNotDisposed(); | ||
|
|
||
| byte b = 0; | ||
| return Read(new Span<byte>(ref b)) > 0 ? b : -1; | ||
| } | ||
|
|
||
| /// <inheritdoc/> | ||
| public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) | ||
| { | ||
| ValidateBufferArguments(buffer, offset, count); | ||
| EnsureNotDisposed(); | ||
|
|
||
| if (cancellationToken.IsCancellationRequested) | ||
| { | ||
| return Task.FromCanceled<int>(cancellationToken); | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
| } | ||
|
|
||
| int n = Read(buffer, offset, count); | ||
| return Task.FromResult(n); | ||
| } | ||
|
|
||
| /// <inheritdoc/> | ||
| public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) | ||
| { | ||
| EnsureNotDisposed(); | ||
|
|
||
| if (cancellationToken.IsCancellationRequested) | ||
| { | ||
| return ValueTask.FromCanceled<int>(cancellationToken); | ||
| } | ||
|
|
||
| int n = Read(buffer.Span); | ||
| return new ValueTask<int>(n); | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override void CopyTo(Stream destination, int bufferSize) | ||
| { | ||
| ValidateCopyToArguments(destination, bufferSize); | ||
| EnsureNotDisposed(); | ||
|
|
||
| if (_absolutePosition >= _sequence.Length) | ||
| { | ||
| return; | ||
| } | ||
|
|
||
| ReadOnlySequence<byte> remaining = _sequence.Slice(_position); | ||
| foreach (ReadOnlyMemory<byte> segment in remaining) | ||
| { | ||
| destination.Write(segment.Span); | ||
| } | ||
|
|
||
| _position = _sequence.End; | ||
| _absolutePosition = _sequence.Length; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken) | ||
| { | ||
| ValidateCopyToArguments(destination, bufferSize); | ||
| EnsureNotDisposed(); | ||
|
|
||
| if (cancellationToken.IsCancellationRequested) | ||
| { | ||
| return Task.FromCanceled(cancellationToken); | ||
| } | ||
|
|
||
| if (_absolutePosition >= _sequence.Length) | ||
| { | ||
| return Task.CompletedTask; | ||
| } | ||
|
|
||
| return CopyToAsyncCore(destination, cancellationToken); | ||
| } | ||
|
|
||
| private async Task CopyToAsyncCore(Stream destination, CancellationToken cancellationToken) | ||
| { | ||
| ReadOnlySequence<byte> remaining = _sequence.Slice(_position); | ||
| foreach (ReadOnlyMemory<byte> segment in remaining) | ||
| { | ||
| await destination.WriteAsync(segment, cancellationToken).ConfigureAwait(false); | ||
| } | ||
|
|
||
| _position = _sequence.End; | ||
| _absolutePosition = _sequence.Length; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(SR.NotSupported_UnwritableStream); | ||
|
|
||
| /// <inheritdoc/> | ||
| public override void Write(ReadOnlySpan<byte> buffer) => throw new NotSupportedException(SR.NotSupported_UnwritableStream); | ||
|
|
||
| /// <inheritdoc/> | ||
| public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw new NotSupportedException(SR.NotSupported_UnwritableStream); | ||
|
|
||
| /// <inheritdoc/> | ||
| public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default) => throw new NotSupportedException(SR.NotSupported_UnwritableStream); | ||
|
|
||
| /// <summary> | ||
| /// Sets the position within the current stream. | ||
| /// </summary> | ||
| /// <param name="offset">A byte offset relative to the <paramref name="origin"/> parameter.</param> | ||
| /// <param name="origin">A value of type <see cref="SeekOrigin"/> indicating the reference point used to obtain the new position.</param> | ||
| /// <returns>The new position within the stream.</returns> | ||
| public override long Seek(long offset, SeekOrigin origin) | ||
| { | ||
| EnsureNotDisposed(); | ||
|
|
||
| long absolutePosition = origin switch | ||
| { | ||
| SeekOrigin.Begin => offset, | ||
| SeekOrigin.Current => _absolutePosition + offset, | ||
| SeekOrigin.End => _sequence.Length + offset, | ||
| _ => throw new ArgumentException(SR.Argument_InvalidSeekOrigin, nameof(origin)) | ||
| }; | ||
|
|
||
| if (absolutePosition < 0) | ||
| { | ||
| throw new IOException(SR.IO_SeekBeforeBegin); | ||
| } | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
|
|
||
| if (absolutePosition >= _sequence.Length) | ||
| { | ||
| _position = _sequence.End; | ||
| } | ||
| else if (absolutePosition >= _absolutePosition) | ||
| { | ||
| _position = _sequence.GetPosition(absolutePosition - _absolutePosition, _position); | ||
| } | ||
| else | ||
| { | ||
| _position = _sequence.GetPosition(absolutePosition, _sequence.Start); | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
| } | ||
|
|
||
| _absolutePosition = absolutePosition; | ||
| return absolutePosition; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public override void Flush() { } | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
|
|
||
| /// <inheritdoc /> | ||
| public override Task FlushAsync(CancellationToken cancellationToken) => | ||
| cancellationToken.IsCancellationRequested ? Task.FromCanceled(cancellationToken) : Task.CompletedTask; | ||
|
|
||
| /// <inheritdoc /> | ||
| public override void SetLength(long value) => throw new NotSupportedException(SR.NotSupported_UnwritableStream); | ||
|
|
||
| /// <inheritdoc /> | ||
| protected override void Dispose(bool disposing) | ||
| { | ||
| _isDisposed = true; | ||
| _sequence = default; | ||
| base.Dispose(disposing); | ||
|
ViveliDuCh marked this conversation as resolved.
|
||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.