< Summary

Information
Class: IceRpc.Slice.Operations.Internal.AsyncStream<T>
Assembly: IceRpc.Slice
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc.Slice/Operations/Internal/AsyncStream.cs
Tag: 1986_28452893481
Line coverage
97%
Covered lines: 67
Uncovered lines: 2
Coverable lines: 69
Total lines: 166
Line coverage: 97.1%
Branch coverage
92%
Covered branches: 13
Total branches: 14
Branch coverage: 92.8%
Method coverage
100%
Covered methods: 4
Fully covered methods: 3
Total methods: 4
Method coverage: 100%
Full method coverage: 75%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
Dispose()100%44100%
GetAsyncEnumerator(...)100%22100%
EnumerateAsync()87.5%8890.47%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc.Slice/Operations/Internal/AsyncStream.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using System.Buffers;
 4using System.Diagnostics;
 5using System.IO.Pipelines;
 6using System.Runtime.CompilerServices;
 7
 8namespace IceRpc.Slice.Operations.Internal;
 9
 10/// <summary>The default <see cref="IAsyncStream{T}" /> implementation. It wraps a <see cref="PipeReader" /> and
 11/// decodes its bytes into elements of type <typeparamref name="T"/> using a read function and a decode function.
 12/// </summary>
 13internal sealed class AsyncStream<T> : IAsyncStream<T>
 14{
 15    private readonly PipeReader _reader;
 16    private readonly Func<PipeReader, CancellationToken, ValueTask<ReadResult>> _readFunc;
 17    private readonly Func<ReadOnlySequence<byte>, IEnumerable<T>> _decodeBufferFunc;
 18
 19    // Canceled by Dispose when iteration has started, to unblock any pending ReadAsync.
 23720    private readonly CancellationTokenSource _disposeCts = new();
 21
 22    // Set when GetAsyncEnumerator is called. This enforces the single-enumerator contract even if the created
 23    // enumerator is never advanced.
 24    private bool _enumeratorCreated;
 25
 26    // Atomic state used to safely arbitrate ownership of _reader.Complete() between Dispose and the first
 27    // MoveNextAsync.
 28    private int _state;
 29
 30    public void Dispose()
 22531    {
 22532        int original = Interlocked.Exchange(ref _state, (int)State.Disposed);
 33
 22534        switch ((State)original)
 35        {
 36            case State.Initial:
 37                // No iteration could have started (and any future MoveNextAsync will see Disposed and throw).
 38                // Safe to complete the reader directly from this thread.
 639                _reader.Complete();
 640                _disposeCts.Dispose();
 641                break;
 42
 43            case State.Iterating:
 44                // The iterator owns the reader; its finally will complete it. We only signal cancellation here.
 45                // We must not dispose _disposeCts here: a linked CTS inside the iterator may still hold a
 46                // registration on _disposeCts.Token.
 21747                _disposeCts.Cancel();
 21748                break;
 49
 50            case State.Disposed:
 51                // no-op (Dispose called more than once).
 252                break;
 53        }
 22554    }
 55
 56    public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken)
 23557    {
 58        // We don't check for Disposed here: if the stream was disposed, the first MoveNextAsync call on the
 59        // returned enumerator throws ObjectDisposedException (see EnumerateAsync).
 23560        if (_enumeratorCreated)
 161        {
 162            throw new InvalidOperationException($"An {nameof(IAsyncStream<T>)} can only be enumerated once.");
 63        }
 23464        _enumeratorCreated = true;
 23465        return EnumerateAsync(cancellationToken).GetAsyncEnumerator(cancellationToken);
 23466    }
 67
 23768    internal AsyncStream(
 23769        PipeReader reader,
 23770        Func<PipeReader, CancellationToken, ValueTask<ReadResult>> readFunc,
 23771        Func<ReadOnlySequence<byte>, IEnumerable<T>> decodeBufferFunc)
 23772    {
 23773        _reader = reader;
 23774        _readFunc = readFunc;
 23775        _decodeBufferFunc = decodeBufferFunc;
 23776    }
 77
 78    private async IAsyncEnumerable<T> EnumerateAsync([EnumeratorCancellation] CancellationToken cancellationToken)
 23279    {
 80        // Because this async method returns an IAsyncEnumerable<T>, it only starts executing when the caller starts
 81        // iterating (calls MoveNextAsync on the enumerator). It does not execute when EnumerateAsync is called, or
 82        // even when GetAsyncEnumerator is called on the returned IAsyncEnumerable<T>.
 83
 84        // Atomically claim the reader (Idle -> Iterating). This races with Dispose's atomic transition to Disposed;
 85        // whichever transition wins from Idle owns _reader.Complete().
 23286        int original = Interlocked.CompareExchange(ref _state, (int)State.Iterating, (int)State.Initial);
 23287        ObjectDisposedException.ThrowIf(original == (int)State.Disposed, this);
 23088        Debug.Assert(original == (int)State.Initial); // _enumeratorCreated forbids a second iteration.
 89
 90        // Link the caller-provided token with our internal dispose token so that Dispose can unblock ReadAsync.
 23091        using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
 23092            cancellationToken,
 23093            _disposeCts.Token);
 23094        CancellationToken linkedToken = linkedCts.Token;
 95
 96        try
 23097        {
 24298            while (true)
 24299            {
 100                ReadResult readResult;
 101
 102                try
 242103                {
 242104                    readResult = await _readFunc(_reader, linkedToken).ConfigureAwait(false);
 105
 38106                    if (readResult.IsCanceled)
 0107                    {
 108                        // We never call CancelPendingRead; an interceptor or middleware can but it's not correct.
 0109                        throw new InvalidOperationException("Unexpected call to CancelPendingRead.");
 110                    }
 38111                    if (readResult.Buffer.IsEmpty)
 3112                    {
 113                        // An empty buffer means the reader completed with no more bytes, which is how the end of a
 114                        // stream is signaled. (A zero-size Slice segment is rejected when the segment is read.)
 3115                        Debug.Assert(readResult.IsCompleted);
 3116                        yield break;
 117                    }
 35118                }
 202119                catch (OperationCanceledException) when (linkedToken.IsCancellationRequested)
 202120                {
 121                    // Re-issue the cancellation with the caller's token so the OCE that propagates carries the
 122                    // token the caller passed in (not our internal linkedToken). When dispose is the only source,
 123                    // surface dispose-mid-iteration as ObjectDisposedException.
 202124                    cancellationToken.ThrowIfCancellationRequested();
 125
 126                    // Safe to read _state without a barrier: Dispose writes State.Disposed before calling
 127                    // _disposeCts.Cancel(), and observing the cancellation here establishes happens-before
 128                    // with that write.
 200129                    Debug.Assert(_state == (int)State.Disposed);
 200130                    throw new ObjectDisposedException(nameof(AsyncStream<>), "The stream was disposed while reading.");
 131                }
 132
 35133                IEnumerable<T> elements = _decodeBufferFunc(readResult.Buffer);
 33134                _reader.AdvanceTo(readResult.Buffer.End);
 135
 262846136                foreach (T item in elements)
 131375137                {
 131375138                    cancellationToken.ThrowIfCancellationRequested();
 139
 140                    // No memory barrier needed: this read is just an early-out optimization. If we miss a
 141                    // concurrent transition to Disposed, the next ReadAsync call observes the cancellation of
 142                    // _disposeCts (which has its own synchronization) and we surface ObjectDisposedException
 143                    // from the catch block above.
 131374144                    ObjectDisposedException.ThrowIf(_state == (int)State.Disposed, this);
 131374145                    yield return item;
 131372146                }
 147
 30148                if (readResult.IsCompleted)
 18149                {
 18150                    yield break;
 151                }
 12152            }
 153        }
 154        finally
 230155        {
 230156            _reader.Complete();
 230157        }
 23158    }
 159
 160    private enum State
 161    {
 162        Initial = 0,
 163        Iterating = 1,
 164        Disposed = 2
 165    }
 166}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.IO.Pipelines.PipeReader,System.Func`3<System.IO.Pipelines.PipeReader,System.Threading.CancellationToken,System.Threading.Tasks.ValueTask`1<System.IO.Pipelines.ReadResult>>,System.Func`2<System.Buffers.ReadOnlySequence`1<System.Byte>,System.Collections.Generic.IEnumerable`1<T>>)
Dispose()
GetAsyncEnumerator(System.Threading.CancellationToken)
EnumerateAsync()