< Summary

Information
Class: IceRpc.Slice.Operations.AsyncEnumerableExtensions
Assembly: IceRpc.Slice
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc.Slice/Operations/AsyncEnumerableExtensions.cs
Tag: 1986_28452893481
Line coverage
95%
Covered lines: 119
Uncovered lines: 6
Coverable lines: 125
Total lines: 229
Line coverage: 95.2%
Branch coverage
90%
Covered branches: 29
Total branches: 32
Branch coverage: 90.6%
Method coverage
80%
Covered methods: 8
Fully covered methods: 4
Total methods: 10
Method coverage: 80%
Full method coverage: 40%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
ToPipeReader(...)100%11100%
.ctor(...)50%22100%
AdvanceTo(...)100%11100%
AdvanceTo(...)100%210%
CancelPendingRead()100%11100%
Complete(...)100%22100%
DisposeEnumeratorAsync()100%2291.66%
ReadAsync()85.71%141492.3%
EncodeElements()100%1212100%
TryRead(...)100%210%

File(s)

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

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using System.Diagnostics;
 4using System.IO.Pipelines;
 5using ZeroC.Slice.Codec;
 6
 7namespace IceRpc.Slice.Operations;
 8
 9/// <summary>Provides an extension method for <see cref="IAsyncEnumerable{T}" /> to encode elements into a <see
 10/// cref="PipeReader"/>.</summary>
 11public static class AsyncEnumerableExtensions
 12{
 13    /// <summary>Encodes an async enumerable into a stream of bytes represented by a <see cref="PipeReader"/>.</summary>
 14    /// <typeparam name="T">The async enumerable element type.</typeparam>
 15    /// <param name="asyncEnumerable">The async enumerable to encode into a stream of bytes.</param>
 16    /// <param name="encodeAction">The action used to encode one element.</param>
 17    /// <param name="useSegments"><see langword="true" /> if an element can be encoded on a variable number of bytes;
 18    /// otherwise, <see langword="false" />.</param>
 19    /// <param name="encodeOptions">The Slice encode options.</param>
 20    /// <returns>A pipe reader that represents the encoded stream of bytes.</returns>
 21    public static PipeReader ToPipeReader<T>(
 22        this IAsyncEnumerable<T> asyncEnumerable,
 23        EncodeAction<T> encodeAction,
 24        bool useSegments,
 25        SliceEncodeOptions? encodeOptions = null) =>
 2526        new AsyncEnumerablePipeReader<T>(
 2527            asyncEnumerable,
 2528            encodeAction,
 2529            useSegments,
 2530            encodeOptions);
 31
 32    // Overriding ReadAtLeastAsyncCore or CopyToAsync methods for this reader is not critical since this reader is
 33    // mostly used by the IceRPC core to copy the encoded data for the enumerable to the network stream. This copy
 34    // doesn't use these methods.
 35#pragma warning disable CA1001 // Types that own disposable fields should be disposable.
 36    private class AsyncEnumerablePipeReader<T> : PipeReader
 37#pragma warning restore CA1001
 38    {
 39        // Disposed in Complete.
 40        private readonly IAsyncEnumerator<T> _asyncEnumerator;
 41
 42        // We don't dispose _cts because it's not necessary
 43        // (see https://github.com/dotnet/runtime/issues/29970#issuecomment-717840778) and we can't easily dispose it
 44        // when no one is using it since CancelPendingRead can be called by another thread after Complete is called.
 2545        private readonly CancellationTokenSource _cts = new();
 46        private readonly EncodeAction<T> _encodeAction;
 47        private bool _isCompleted;
 48        private readonly bool _useSegments;
 49        private readonly int _streamFlushThreshold;
 50        private Task<bool>? _moveNext;
 51        private readonly Pipe _pipe;
 52
 11253        public override void AdvanceTo(SequencePosition consumed) => _pipe.Reader.AdvanceTo(consumed);
 54
 55        public override void AdvanceTo(SequencePosition consumed, SequencePosition examined) =>
 056            _pipe.Reader.AdvanceTo(consumed, examined);
 57
 58        public override void CancelPendingRead()
 159        {
 160            _pipe.Reader.CancelPendingRead();
 161            _cts.Cancel();
 162        }
 63
 64        public override void Complete(Exception? exception = null)
 2665        {
 2666            if (!_isCompleted)
 1967            {
 1968                _isCompleted = true;
 69
 70                // Cancel MoveNextAsync if it's still running.
 1971                _cts.Cancel();
 72
 1973                _pipe.Reader.Complete();
 1974                _pipe.Writer.Complete();
 75
 1976                _ = DisposeEnumeratorAsync();
 1977            }
 78
 79            async Task DisposeEnumeratorAsync()
 1980            {
 81                // Make sure MoveNextAsync is completed before disposing the enumerator. Calling DisposeAsync on the
 82                // enumerator while MoveNextAsync is still running is disallowed.
 1983                if (_moveNext is not null)
 284                {
 85                    try
 286                    {
 287                        _ = await _moveNext.ConfigureAwait(false);
 088                    }
 289                    catch
 290                    {
 291                    }
 292                }
 1993                await _asyncEnumerator.DisposeAsync().ConfigureAwait(false);
 1994            }
 2695        }
 96
 97        public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancellationToken = default)
 11598        {
 11599            if (!_pipe.Reader.TryRead(out ReadResult readResult))
 115100            {
 101                // If no more buffered data to read, fill the pipe with new data.
 102
 103                // If ReadAsync is canceled, cancel the enumerator iteration to ensure MoveNextAsync below completes.
 115104                using CancellationTokenRegistration registration = cancellationToken.UnsafeRegister(
 1105                    cts => ((CancellationTokenSource)cts!).Cancel(),
 115106                    _cts);
 107
 108                bool hasNext;
 109                try
 115110                {
 115111                    if (_moveNext is null)
 24112                    {
 24113                        hasNext = await _asyncEnumerator.MoveNextAsync().ConfigureAwait(false);
 22114                    }
 115                    else
 91116                    {
 91117                        hasNext = await _moveNext.ConfigureAwait(false);
 90118                        _moveNext = null;
 90119                    }
 120
 112121                    if (hasNext && EncodeElements() is Task<bool> moveNext)
 92122                    {
 123                        // Flush does not block because the pipe is configured to not pause flush.
 92124                        ValueTask<FlushResult> valueTask = _pipe.Writer.FlushAsync(CancellationToken.None);
 92125                        Debug.Assert(valueTask.IsCompletedSuccessfully);
 126
 92127                        _moveNext = moveNext;
 128                        // And the next ReadAsync will await _moveNext.
 92129                    }
 130                    else
 20131                    {
 132                        // No need to flush the writer, complete takes care of it.
 20133                        _pipe.Writer.Complete();
 20134                    }
 135
 136                    // There are bytes in the reader or it's completed since we've just flushed or completed the writer.
 112137                    bool ok = _pipe.Reader.TryRead(out readResult);
 112138                    Debug.Assert(ok);
 112139                }
 2140                catch (OperationCanceledException) when (_cts.IsCancellationRequested)
 2141                {
 2142                    cancellationToken.ThrowIfCancellationRequested();
 143
 1144                    if (_pipe.Reader.TryRead(out readResult) && readResult.IsCanceled)
 1145                    {
 146                        // Ok: return canceled readResult once after calling CancelPendingRead.
 147                        // Note that we can't return a canceled read result with a bogus buffer since the caller must
 148                        // be able to call reader.AdvanceTo with this buffer.
 1149                    }
 150                    else
 0151                    {
 0152                        throw new NotSupportedException(
 0153                            "Cannot resume reading an AsyncEnumerablePipeReader after canceling a ReadAsync or calling C
 154                    }
 1155                }
 113156            }
 157
 113158            return readResult;
 159
 160            Task<bool>? EncodeElements()
 110161            {
 110162                var encoder = new SliceEncoder(_pipe.Writer);
 163
 110164                Span<byte> sizePlaceholder = default;
 110165                if (_useSegments)
 72166                {
 72167                    sizePlaceholder = encoder.GetPlaceholderSpan(4);
 72168                }
 169
 110170                Task<bool>? result = null;
 171                bool keepEncoding;
 172
 173                do
 131337174                {
 131337175                    _encodeAction(ref encoder, _asyncEnumerator.Current);
 131337176                    ValueTask<bool> moveNext = _asyncEnumerator.MoveNextAsync();
 177
 131337178                    if (moveNext.IsCompletedSuccessfully)
 131307179                    {
 131307180                        bool hasNext = moveNext.Result;
 181
 182                        // If we reached the stream flush threshold, it's time to flush.
 131307183                        if (encoder.EncodedByteCount - sizePlaceholder.Length >= _streamFlushThreshold)
 63184                        {
 63185                            result = hasNext ? moveNext.AsTask() : null;
 63186                            keepEncoding = false;
 63187                        }
 188                        else
 131244189                        {
 131244190                            keepEncoding = hasNext;
 131244191                        }
 131307192                    }
 193                    else
 30194                    {
 195                        // If we can't get the next element synchronously, we return the move next task and end the loop
 196                        // to flush the encoded elements.
 30197                        result = moveNext.AsTask();
 30198                        keepEncoding = false;
 30199                    }
 131337200                }
 131337201                while (keepEncoding);
 202
 110203                if (_useSegments)
 72204                {
 72205                    SliceEncoder.EncodeVarUInt62(
 72206                        (ulong)(encoder.EncodedByteCount - sizePlaceholder.Length),
 72207                        sizePlaceholder);
 72208                }
 110209                return result;
 110210            }
 113211        }
 212
 0213        public override bool TryRead(out ReadResult result) => _pipe.Reader.TryRead(out result);
 214
 25215        internal AsyncEnumerablePipeReader(
 25216            IAsyncEnumerable<T> asyncEnumerable,
 25217            EncodeAction<T> encodeAction,
 25218            bool useSegments,
 25219            SliceEncodeOptions? encodeOptions)
 25220        {
 25221            encodeOptions ??= SliceEncodeOptions.Default;
 25222            _pipe = new Pipe(encodeOptions.PipeOptions);
 25223            _streamFlushThreshold = encodeOptions.StreamFlushThreshold;
 25224            _encodeAction = encodeAction;
 25225            _useSegments = useSegments;
 25226            _asyncEnumerator = asyncEnumerable.GetAsyncEnumerator(_cts.Token);
 25227        }
 228    }
 229}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

ToPipeReader(System.Collections.Generic.IAsyncEnumerable`1<T>,ZeroC.Slice.Codec.EncodeAction`1<T>,System.Boolean,IceRpc.Slice.SliceEncodeOptions)
.ctor(System.Collections.Generic.IAsyncEnumerable`1<T>,ZeroC.Slice.Codec.EncodeAction`1<T>,System.Boolean,IceRpc.Slice.SliceEncodeOptions)
AdvanceTo(System.SequencePosition)
AdvanceTo(System.SequencePosition,System.SequencePosition)
CancelPendingRead()
Complete(System.Exception)
DisposeEnumeratorAsync()
ReadAsync()
EncodeElements()
TryRead(System.IO.Pipelines.ReadResult&)