< 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: 2300_35243572715
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) =>
 2726        new AsyncEnumerablePipeReader<T>(
 2727            asyncEnumerable,
 2728            encodeAction,
 2729            useSegments,
 2730            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.
 2745        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
 11453        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)
 2965        {
 2966            if (!_isCompleted)
 2167            {
 2168                _isCompleted = true;
 69
 70                // Cancel MoveNextAsync if it's still running.
 2171                _cts.Cancel();
 72
 2173                _pipe.Reader.Complete();
 2174                _pipe.Writer.Complete();
 75
 2176                _ = DisposeEnumeratorAsync();
 2177            }
 78
 79            async Task DisposeEnumeratorAsync()
 2180            {
 81                // Make sure MoveNextAsync is completed before disposing the enumerator. Calling DisposeAsync on the
 82                // enumerator while MoveNextAsync is still running is disallowed.
 2183                if (_moveNext is not null)
 284                {
 85                    try
 286                    {
 287                        _ = await _moveNext.ConfigureAwait(false);
 088                    }
 289                    catch
 290                    {
 291                    }
 292                }
 2193                await _asyncEnumerator.DisposeAsync().ConfigureAwait(false);
 2194            }
 2995        }
 96
 97        public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancellationToken = default)
 11798        {
 11799            if (!_pipe.Reader.TryRead(out ReadResult readResult))
 117100            {
 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.
 117104                using CancellationTokenRegistration registration = cancellationToken.UnsafeRegister(
 1105                    cts => ((CancellationTokenSource)cts!).Cancel(),
 117106                    _cts);
 107
 108                bool hasNext;
 109                try
 117110                {
 117111                    if (_moveNext is null)
 26112                    {
 26113                        hasNext = await _asyncEnumerator.MoveNextAsync().ConfigureAwait(false);
 24114                    }
 115                    else
 91116                    {
 91117                        hasNext = await _moveNext.ConfigureAwait(false);
 90118                        _moveNext = null;
 90119                    }
 120
 114121                    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
 22131                    {
 132                        // No need to flush the writer, complete takes care of it.
 22133                        _pipe.Writer.Complete();
 22134                    }
 135
 136                    // There are bytes in the reader or it's completed since we've just flushed or completed the writer.
 114137                    bool ok = _pipe.Reader.TryRead(out readResult);
 114138                    Debug.Assert(ok);
 114139                }
 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                }
 115156            }
 157
 115158            return readResult;
 159
 160            Task<bool>? EncodeElements()
 112161            {
 112162                var encoder = new SliceEncoder(_pipe.Writer);
 163
 112164                Span<byte> sizePlaceholder = default;
 112165                if (_useSegments)
 74166                {
 74167                    sizePlaceholder = encoder.GetPlaceholderSpan(4);
 74168                }
 169
 112170                Task<bool>? result = null;
 171                bool keepEncoding;
 172
 173                do
 131341174                {
 131341175                    _encodeAction(ref encoder, _asyncEnumerator.Current);
 131341176                    ValueTask<bool> moveNext = _asyncEnumerator.MoveNextAsync();
 177
 131341178                    if (moveNext.IsCompletedSuccessfully)
 131311179                    {
 131311180                        bool hasNext = moveNext.Result;
 181
 182                        // If we reached the stream flush threshold, it's time to flush.
 131311183                        if (encoder.EncodedByteCount - sizePlaceholder.Length >= _streamFlushThreshold)
 63184                        {
 63185                            result = hasNext ? Task.FromResult(true) : null;
 63186                            keepEncoding = false;
 63187                        }
 188                        else
 131248189                        {
 131248190                            keepEncoding = hasNext;
 131248191                        }
 131311192                    }
 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                    }
 131341200                }
 131341201                while (keepEncoding);
 202
 112203                if (_useSegments)
 74204                {
 74205                    SliceEncoder.EncodeVarUInt62(
 74206                        (ulong)(encoder.EncodedByteCount - sizePlaceholder.Length),
 74207                        sizePlaceholder);
 74208                }
 112209                return result;
 112210            }
 115211        }
 212
 0213        public override bool TryRead(out ReadResult result) => _pipe.Reader.TryRead(out result);
 214
 27215        internal AsyncEnumerablePipeReader(
 27216            IAsyncEnumerable<T> asyncEnumerable,
 27217            EncodeAction<T> encodeAction,
 27218            bool useSegments,
 27219            SliceEncodeOptions? encodeOptions)
 27220        {
 27221            encodeOptions ??= SliceEncodeOptions.Default;
 27222            _pipe = new Pipe(encodeOptions.PipeOptions);
 27223            _streamFlushThreshold = encodeOptions.StreamFlushThreshold;
 27224            _encodeAction = encodeAction;
 27225            _useSegments = useSegments;
 27226            _asyncEnumerator = asyncEnumerable.GetAsyncEnumerator(_cts.Token);
 27227        }
 228    }
 229}
Generated by: ReportGenerator 5.5.11.0
09/17/2026 - 16:01:28
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&)