< Summary

Information
Class: IceRpc.ResettablePipeReaderDecorator
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/ResettablePipeReaderDecorator.cs
Tag: 1986_28452893481
Line coverage
72%
Covered lines: 145
Uncovered lines: 54
Coverable lines: 199
Total lines: 392
Line coverage: 72.8%
Branch coverage
71%
Covered branches: 46
Total branches: 64
Branch coverage: 71.8%
Method coverage
100%
Covered methods: 14
Fully covered methods: 7
Total methods: 14
Method coverage: 100%
Full method coverage: 50%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_IsResettable()100%11100%
set_IsResettable(...)66.66%9656.25%
.ctor(...)100%11100%
AdvanceTo(...)100%11100%
AdvanceTo(...)80%111078.26%
CancelPendingRead()100%11100%
Complete(...)83.33%66100%
ReadAsync()75%4482.6%
Reset()62.5%9873.68%
TryRead(...)50%10652.17%
ReadAtLeastAsyncCore()50%12644.11%
AdvanceDecoratee()75%88100%
ProcessReadResult(...)100%88100%
ThrowIfCompleted()50%3250%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/ResettablePipeReaderDecorator.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using System.Buffers;
 4using System.Diagnostics;
 5using System.IO.Pipelines;
 6
 7namespace IceRpc;
 8
 9/// <summary>Represents a <see cref="PipeReader" /> decorator that doesn't consume the data from the decoratee to allow
 10/// reading again this data from the beginning after being reset.</summary>
 11/// <remarks><para>The decorator becomes non-resettable if the decoratee's buffered data exceeds the maximum buffer size
 12/// provided to <see cref="ResettablePipeReaderDecorator(PipeReader, int)" /> or if the reading from the decoratee fails
 13/// with an exception other than <see cref="OperationCanceledException"/>.</para>
 14/// <para>Calling <see cref="Complete" /> on the decorator doesn't complete the decoratee to allow reading again the
 15/// data after the decorator is reset. It's therefore important to make the decorator non-resettable by setting <see
 16/// cref="IsResettable" /> to <see langword="false" /> to complete the decoratee.</para></remarks>
 17// The default CopyToAsync implementation is suitable for this reader implementation. It calls ReadAsync/AdvanceTo to
 18// read the data. This ensures that the decorated pipe reader buffered data is not consumed.
 19public sealed class ResettablePipeReaderDecorator : PipeReader
 20{
 21    /// <summary>Gets or sets a value indicating whether this decorator can be reset.</summary>
 22    /// <value><see langword="true"/> if this decorator can be reset; otherwise, <see langword="false"/>. Defaults to
 23    /// <see langword="true"/>.</value>
 24    /// <remarks>This property can only be set to <see langword="false" />. If <see cref="IsResettable"/> is <see
 25    /// langword="true" /> and <see cref="Complete" /> was called, the decoratee is completed.</remarks>
 26    public bool IsResettable
 27    {
 2028        get => _isResettable;
 29
 30        set
 2431        {
 2432            if (value)
 033            {
 034                throw new ArgumentException(
 035                    $"The {nameof(IsResettable)} property cannot be set to true.",
 036                    nameof(value));
 37            }
 38
 2439            if (_isResettable)
 1940            {
 1941                AdvanceDecoratee();
 42
 43                // If Complete was called on this resettable decorator without an intervening Reset, we call Complete
 44                // on the decoratee.
 45
 1946                _isResettable = false;
 1947                if (_isReaderCompleted)
 048                {
 49                    // We complete the decoratee with the saved exception (can be null).
 050                    Complete(_readerCompleteException);
 051                }
 1952            }
 2453        }
 54    }
 55
 56    // The latest consumed given by caller; cleared by Reset and by the first non-resettable AdvanceTo.
 57    private SequencePosition? _consumed;
 58    private readonly PipeReader _decoratee;
 59    // The latest examined given by caller.
 60    private SequencePosition? _examined;
 61
 62    // The highest examined given to _decoratee; not affected by Reset but cleared by the first non-resettable
 63    // AdvanceTo.
 64    private SequencePosition? _highestExamined;
 65
 66    // True when read returned a canceled read result.
 67    private bool _isCanceled;
 68    // True when the caller complete this reader; reset by Reset.
 69    private bool _isReaderCompleted;
 70    private bool _isReadingInProgress;
 3371    private bool _isResettable = true;
 72    private readonly int _maxBufferSize;
 73    private Exception? _readerCompleteException;
 74
 75    // The latest sequence returned by _decoratee; not affected by Reset.
 76    private ReadOnlySequence<byte> _sequence;
 77
 78    /// <summary>Constructs a resettable pipe reader decorator.</summary>
 79    /// <param name="decoratee">The pipe reader being decorated.</param>
 80    /// <param name="maxBufferSize">The maximum size of buffered data in bytes.</param>
 3381    public ResettablePipeReaderDecorator(PipeReader decoratee, int maxBufferSize)
 3382    {
 3383        _decoratee = decoratee;
 3384        _maxBufferSize = maxBufferSize;
 3385    }
 86
 87    /// <summary>Moves forward the pipeline's read cursor to after the consumed data. No data is consumed while
 88    /// <see cref="IsResettable"/> value is true.</summary>
 89    /// <param name="consumed">Marks the extent of the data that has been successfully processed.</param>
 90    /// <seealso cref="PipeReader.AdvanceTo(SequencePosition)"/>
 2691    public override void AdvanceTo(SequencePosition consumed) => AdvanceTo(consumed, consumed);
 92
 93    /// <summary>Moves forward the pipeline's read cursor to after the consumed data. No data is consumed while
 94    /// <see cref="IsResettable"/> value is true.</summary>
 95    /// <param name="consumed">Marks the extent of the data that has been successfully processed.</param>
 96    /// <param name="examined">Marks the extent of the data that has been read and examined.</param>
 97    /// <seealso cref="PipeReader.AdvanceTo(SequencePosition, SequencePosition)"/>
 98    public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)
 3099    {
 100        // If reading returns a canceled read result, _isReadInProgress is set to false since it's not required to call
 101        // AdvanceTo. Calling AdvanceTo after getting a canceled read result is also valid so we don't check if reading
 102        // is in progress in this case.
 30103        if (!_isCanceled && !_isReadingInProgress)
 0104        {
 0105            throw new InvalidOperationException("Cannot call AdvanceTo before reading the PipeReader.");
 106        }
 107
 30108        _isReadingInProgress = false;
 109
 30110        Debug.Assert(_examined is null);
 111
 30112        if (_isResettable)
 21113        {
 21114            ThrowIfCompleted();
 115
 116            // Don't call _decoratee.AdvanceTo just yet. It will be called on the next ReadAsync/TryRead call. This
 117            // way, if Reset is called next, it won't mark the data as examined and the following ReadAsync/TryRead
 118            // call won't block. It will return the buffered data.
 21119            _examined = examined;
 21120            _consumed = consumed;
 21121        }
 122        else
 9123        {
 124            // The examined position given to _decoratee.AdvanceTo must be ever-increasing.
 9125            if (_highestExamined is not null &&
 9126                _sequence.GetOffset(examined) < _sequence.GetOffset(_highestExamined.Value))
 0127            {
 0128                examined = _highestExamined.Value;
 0129            }
 9130            _decoratee.AdvanceTo(consumed, examined);
 131
 132            // The decoratee can now recycle the buffers of the consumed data: the positions saved while this
 133            // decorator was resettable are no longer valid in subsequent read results.
 9134            _consumed = null;
 9135            _highestExamined = null;
 9136        }
 30137    }
 138
 139    /// <summary>Cancels the pending <see cref="ReadAsync(CancellationToken)"/> operation without causing it to throw
 140    /// and without completing the <see cref="PipeReader"/>. If there is no pending operation, this cancels the next
 141    /// operation.</summary>
 142    /// <seealso cref="PipeReader.CancelPendingRead"/>
 143    // This method can be called from another thread so we always forward it to the decoratee directly.
 144    // ReadAsync/ReadAtLeastAsync/TryRead will return IsCanceled as appropriate.
 3145    public override void CancelPendingRead() => _decoratee.CancelPendingRead();
 146
 147    /// <summary>Signals to the producer that the consumer is done reading.</summary>
 148    /// <param name="exception">Optional <see cref="Exception "/> indicating a failure that's causing the pipeline to
 149    /// complete.</param>
 150    /// <seealso cref="PipeReader.Complete(Exception?)"/>
 151    /// <remarks>If <see cref="IsResettable"/> value is true, <see cref="Complete" /> is not called on the decoratee to
 152    /// allow reading again the data after a call to <see cref="Reset" />. To complete the decoratee, <see
 153    /// cref="IsResettable" /> must be set to <see langword="false" />.</remarks>
 154    public override void Complete(Exception? exception = default)
 32155    {
 32156        if (_isResettable)
 10157        {
 10158            if (_isReadingInProgress)
 7159            {
 7160                AdvanceTo(_sequence.Start);
 7161            }
 162
 10163            if (!_isReaderCompleted)
 10164            {
 165                // Only save the first call to Complete
 10166                _isReaderCompleted = true;
 10167                _readerCompleteException = exception;
 10168            }
 169            // we naturally don't complete the decoratee, otherwise this decorator would no longer be resettable
 10170        }
 171        else
 22172        {
 22173            _isReadingInProgress = false;
 22174            _decoratee.Complete(exception);
 22175        }
 32176    }
 177
 178    /// <summary>Asynchronously reads a sequence of bytes from the current <see cref="PipeReader"/>.</summary>
 179    /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
 180    /// <returns>A <see cref="ValueTask{TResult}"/> representing the asynchronous read operation.</returns>
 181    /// <seealso cref="PipeReader.ReadAsync(CancellationToken)"/>
 182    public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancellationToken = default)
 28183    {
 28184        _isReadingInProgress = !_isReadingInProgress ? true :
 28185            throw new InvalidOperationException("Reading is already in progress.");
 186
 28187        ThrowIfCompleted();
 188
 28189        AdvanceDecoratee();
 190
 191        ReadResult readResult;
 192        try
 28193        {
 28194            readResult = await _decoratee.ReadAsync(cancellationToken).ConfigureAwait(false);
 27195            if (readResult.IsCanceled)
 2196            {
 2197                _isCanceled = true;
 2198                _isReadingInProgress = false;
 2199            }
 27200        }
 1201        catch (OperationCanceledException)
 1202        {
 1203            _isReadingInProgress = false;
 1204            throw;
 205        }
 0206        catch
 0207        {
 0208            _isResettable = false;
 0209            throw;
 210        }
 27211        return ProcessReadResult(readResult);
 27212    }
 213
 214    /// <summary>Resets this pipe reader.</summary>
 215    /// <exception cref="InvalidOperationException">Thrown if <see cref="IsResettable" /> is <see langword="false" /> or
 216    /// if reading is in progress.</exception>
 217    public void Reset()
 18218    {
 18219        if (_isResettable)
 18220        {
 18221            if (_isReadingInProgress)
 0222            {
 0223                throw new InvalidOperationException(
 0224                    "The resettable pipe reader decorator cannot be reset while reading is in progress.");
 225            }
 226
 18227            if (_examined is not null)
 6228            {
 229                // Don't commit the caller's examined data on the decoratee. This ensures that the next ReadAsync call
 230                // returns synchronously with the decoratee's buffered data (instead of blocking).
 6231                _decoratee.AdvanceTo(_sequence.Start, _highestExamined ?? _sequence.Start);
 6232                _examined = null;
 6233            }
 234
 18235            _consumed = null;
 18236            _isReaderCompleted = false;
 18237            _readerCompleteException = null;
 18238        }
 239        else
 0240        {
 0241            throw new InvalidOperationException("Cannot reset non-resettable pipe reader decorator.");
 242        }
 18243    }
 244
 245    /// <summary>Attempts to synchronously read data from the <see cref="PipeReader"/>.</summary>
 246    /// <param name="result">When this method returns <see langword="true"/>, this value is set to a
 247    /// <see cref="ReadResult"/> instance that represents the result of the read call; otherwise, this value is set to
 248    /// <see langword="default"/>.</param>
 249    /// <returns><see langword="true"/> if data was available, or if the call was canceled or the writer was completed;
 250    /// otherwise, <see langword="false"/>.</returns>
 251    /// <seealso cref="PipeReader.TryRead(out ReadResult)"/>.
 252    public override bool TryRead(out ReadResult result)
 2253    {
 2254        _isReadingInProgress = !_isReadingInProgress ? true :
 2255            throw new InvalidOperationException("Reading is already in progress.");
 256
 2257        ThrowIfCompleted();
 258
 2259        AdvanceDecoratee();
 260
 261        try
 2262        {
 2263            if (_decoratee.TryRead(out result))
 2264            {
 2265                if (result.IsCanceled)
 0266                {
 0267                    _isCanceled = true;
 0268                    _isReadingInProgress = false;
 0269                }
 2270                result = ProcessReadResult(result);
 2271                return true;
 272            }
 273            else
 0274            {
 0275                _isReadingInProgress = false;
 0276                return false;
 277            }
 278        }
 0279        catch
 0280        {
 0281            _isResettable = false;
 0282            throw;
 283        }
 2284    }
 285
 286    /// <summary>Asynchronously reads a sequence of bytes from the current PipeReader.</summary>
 287    /// <param name="minimumSize">The minimum length that needs to be buffered in order for the call to return.</param>
 288    /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
 289    /// <returns>A <see cref="ValueTask{TResult}"/> representing the asynchronous read operation.</returns>
 290    protected override async ValueTask<ReadResult> ReadAtLeastAsyncCore(
 291        int minimumSize,
 292        CancellationToken cancellationToken = default)
 3293    {
 3294        _isReadingInProgress = !_isReadingInProgress ? true :
 3295            throw new InvalidOperationException("Reading is already in progress.");
 296
 3297        ThrowIfCompleted();
 298
 3299        AdvanceDecoratee();
 300
 3301        long size = (_consumed is null ? 0 : _sequence.GetOffset(_consumed.Value)) + minimumSize;
 302        try
 3303        {
 3304            minimumSize = checked((int)size);
 3305        }
 0306        catch (OverflowException exception)
 0307        {
 308            // In theory this shouldn't happen if _maxBufferSize is set to a reasonable value.
 0309            throw new ArgumentException(
 0310                $"{minimumSize} is too large and would cause the buffered data to be larger than int.MaxValue",
 0311                nameof(minimumSize),
 0312                exception);
 313        }
 314
 315        ReadResult readResult;
 316        try
 3317        {
 3318            readResult = await _decoratee.ReadAtLeastAsync(minimumSize, cancellationToken).ConfigureAwait(false);
 3319            if (readResult.IsCanceled)
 0320            {
 0321                _isCanceled = true;
 0322                _isReadingInProgress = false;
 0323            }
 3324        }
 0325        catch (OperationCanceledException)
 0326        {
 0327            _isReadingInProgress = false;
 0328            throw;
 329        }
 0330        catch
 0331        {
 0332            _isReadingInProgress = false;
 0333            _isResettable = false;
 0334            throw;
 335        }
 3336        return ProcessReadResult(readResult);
 3337    }
 338
 339    /// <summary>Advances the decoratee to the examined position saved by the AdvanceTo call on the decorator.</summary>
 340    /// <remarks>Calling AdvanceTo on the decoratee commits the latest examined data and ensures the next read call will
 341    /// read additional data if all the data was examined. If the decorator is reset, this ensures that the next read
 342    /// call will immediately return the buffered data instead of blocking.</remarks>
 343    private void AdvanceDecoratee()
 52344    {
 52345        _isCanceled = false;
 346
 52347        if (_isResettable && _examined is not null)
 7348        {
 349            // The examined position given to _decoratee.AdvanceTo must be ever-increasing.
 7350            if (_highestExamined is null ||
 7351                _sequence.GetOffset(_examined.Value) > _sequence.GetOffset(_highestExamined.Value))
 7352            {
 7353                _highestExamined = _examined;
 7354            }
 355
 7356            _decoratee.AdvanceTo(_sequence.Start, _highestExamined.Value);
 357
 7358            _examined = null;
 7359        }
 52360    }
 361
 362    private ReadResult ProcessReadResult(ReadResult readResult)
 32363    {
 32364        _sequence = readResult.Buffer;
 365
 32366        if (_consumed is SequencePosition consumed)
 6367        {
 368            // Remove bytes marked as consumed
 6369            readResult = new ReadResult(
 6370                readResult.Buffer.Slice(consumed),
 6371                readResult.IsCanceled,
 6372                readResult.IsCompleted);
 6373        }
 374
 375        // We don't retry when the buffered data exceeds the maximum buffer size.
 32376        if (_isResettable && (_sequence.Length > _maxBufferSize))
 3377        {
 3378            _isResettable = false;
 3379        }
 380
 32381        return readResult;
 32382    }
 383
 384    private void ThrowIfCompleted()
 54385    {
 54386        if (_isReaderCompleted)
 0387        {
 0388            _isResettable = false;
 0389            throw new InvalidOperationException("The pipe reader is completed.");
 390        }
 54391    }
 392}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

get_IsResettable()
set_IsResettable(System.Boolean)
.ctor(System.IO.Pipelines.PipeReader,System.Int32)
AdvanceTo(System.SequencePosition)
AdvanceTo(System.SequencePosition,System.SequencePosition)
CancelPendingRead()
Complete(System.Exception)
ReadAsync()
Reset()
TryRead(System.IO.Pipelines.ReadResult&)
ReadAtLeastAsyncCore()
AdvanceDecoratee()
ProcessReadResult(System.IO.Pipelines.ReadResult)
ThrowIfCompleted()