< Summary

Information
Class: IceRpc.Protobuf.RpcMethods.Internal.PipeReaderExtensions
Assembly: IceRpc.Protobuf
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc.Protobuf/RpcMethods/Internal/PipeReaderExtensions.cs
Tag: 1986_28452893481
Line coverage
85%
Covered lines: 72
Uncovered lines: 12
Coverable lines: 84
Total lines: 188
Line coverage: 85.7%
Branch coverage
80%
Covered branches: 21
Total branches: 26
Branch coverage: 80.7%
Method coverage
100%
Covered methods: 4
Fully covered methods: 2
Total methods: 4
Method coverage: 100%
Full method coverage: 50%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
DecodeProtobufMessageAsync()75%8883.33%
ToAsyncStream(...)100%11100%
ReadProtobufMessageAsync()81.25%171683.72%
DecodeMessageLength()100%22100%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc.Protobuf/RpcMethods/Internal/PipeReaderExtensions.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using Google.Protobuf;
 4using System.Buffers;
 5using System.Buffers.Binary;
 6using System.Diagnostics;
 7using System.IO.Pipelines;
 8
 9namespace IceRpc.Protobuf.RpcMethods.Internal;
 10
 11/// <summary>Provides extension methods for <see cref="PipeReader" />.</summary>
 12internal static class PipeReaderExtensions
 13{
 14    /// <summary>Decodes a Protobuf length-prefixed message from a <see cref="PipeReader" />.</summary>
 15    /// <param name="reader">The <see cref="PipeReader" /> containing the Protobuf length-prefixed message.</param>
 16    /// <param name="parser">The <see cref="MessageParser{T}" /> used to parse the message data.</param>
 17    /// <param name="maxMessageLength">The maximum allowed length.</param>
 18    /// <param name="acceptEmptyPayload">Indicates whether to accept or reject empty payloads.</param>
 19    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 20    /// <returns>The decoded message object.</returns>
 21    /// <remarks>The envelope follows the gRPC Length-Prefixed-Message format: a 1-byte compression flag (0 for
 22    /// uncompressed; 1 indicates compressed and is rejected as not supported), followed by the message length
 23    /// as a 4-byte unsigned big-endian integer, followed by the Protobuf-encoded message bytes. See
 24    /// <see href="https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#data-frames" />.</remarks>
 25    internal static async ValueTask<T> DecodeProtobufMessageAsync<T>(
 26        this PipeReader reader,
 27        MessageParser<T> parser,
 28        int maxMessageLength,
 29        bool acceptEmptyPayload,
 30        CancellationToken cancellationToken) where T : class, IMessage<T>
 4231    {
 4232        T? message = await reader.ReadProtobufMessageAsync(
 4233            parser,
 4234            maxMessageLength,
 4235            cancellationToken).ConfigureAwait(false);
 36
 3537        if (message is null)
 138        {
 139            if (acceptEmptyPayload)
 140            {
 41                // An empty payload represents a default-constructed message. Used for oneway requests.
 142                message = parser.ParseFrom([]);
 143            }
 44            else
 045            {
 046                throw new InvalidDataException(
 047                    "The payload is empty; a length-prefixed Protobuf message was expected.");
 48            }
 149        }
 50        else
 3451        {
 52            // A unary payload must contain exactly one message; any trailing bytes indicate a framing error.
 3453            ReadResult readResult = await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 54            // We never call CancelPendingRead; an interceptor or middleware can but it's not correct.
 3455            if (readResult.IsCanceled)
 056            {
 057                throw new InvalidOperationException("Unexpected call to CancelPendingRead.");
 58            }
 3459            bool hasTrailingBytes = !readResult.Buffer.IsEmpty;
 3460            reader.AdvanceTo(readResult.Buffer.End);
 3461            if (hasTrailingBytes)
 262            {
 263                throw new InvalidDataException(
 264                    "The payload contains unexpected trailing bytes after the Protobuf message.");
 65            }
 3266            Debug.Assert(readResult.IsCompleted);
 3267        }
 68
 3369        return message;
 3370    }
 71
 72    /// <summary>Creates an async stream over a pipe reader to decode Protobuf messages.</summary>
 73    /// <typeparam name="T">The type of the message being decoded.</typeparam>
 74    /// <param name="reader">The pipe reader.</param>
 75    /// <param name="messageParser">The <see cref="MessageParser{T}" /> used to parse the message data.</param>
 76    /// <param name="maxMessageLength">The maximum allowed length.</param>
 77    /// <returns>The async stream to decode and return the streamed messages.</returns>
 78    /// <remarks>The reader ownership is transferred to the returned async stream. The caller should no longer use
 79    /// the reader after this call, and must dispose the returned async stream when done to release the reader.
 80    /// </remarks>
 81    internal static IAsyncStream<T> ToAsyncStream<T>(
 82        this PipeReader reader,
 83        MessageParser<T> messageParser,
 84        int maxMessageLength) where T : class, IMessage<T> =>
 23285        new AsyncStream<T>(reader, messageParser, maxMessageLength);
 86
 87    /// <summary>Reads a single Protobuf length-prefixed message from a <see cref="PipeReader" />.</summary>
 88    /// <typeparam name="T">The type of the message being decoded.</typeparam>
 89    /// <param name="reader">The pipe reader.</param>
 90    /// <param name="messageParser">The <see cref="MessageParser{T}" /> used to parse the message data.</param>
 91    /// <param name="maxMessageLength">The maximum allowed length.</param>
 92    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 93    /// <returns>The decoded message, or <see langword="null" /> when the reader's stream is empty (end of stream).
 94    /// The <see langword="null" /> return value is used as a sentinel to signal end of stream; this is why
 95    /// <typeparamref name="T" /> is constrained to be a reference type.</returns>
 96    internal static async ValueTask<T?> ReadProtobufMessageAsync<T>(
 97        this PipeReader reader,
 98        MessageParser<T> messageParser,
 99        int maxMessageLength,
 100        CancellationToken cancellationToken) where T : class, IMessage<T>
 66023101    {
 66023102        ReadResult readResult = await reader.ReadAtLeastAsync(5, cancellationToken).ConfigureAwait(false);
 103        // We never call CancelPendingRead; an interceptor or middleware can but it's not correct.
 65848104        if (readResult.IsCanceled)
 0105        {
 0106            throw new InvalidOperationException("Unexpected call to CancelPendingRead.");
 107        }
 108
 65848109        if (readResult.Buffer.IsEmpty)
 21110        {
 21111            return null;
 112        }
 113
 65827114        if (readResult.Buffer.Length < 5)
 1115        {
 1116            throw new InvalidDataException(
 1117                $"The payload has {readResult.Buffer.Length} bytes, but 5 bytes were expected.");
 118        }
 119
 65826120        byte compressionFlag = readResult.Buffer.FirstSpan[0];
 65826121        switch (compressionFlag)
 122        {
 123            case 0:
 65823124                break;
 125            case 1:
 126                // The Protobuf envelope defines flag 1 as "compressed". The message is well-formed Protobuf,
 127                // but IceRPC doesn't decompress it.
 1128                throw new NotSupportedException("IceRPC does not support Protobuf compressed messages.");
 129            default:
 130                // The Protobuf envelope only defines flags 0 and 1; any other value is malformed.
 2131                throw new InvalidDataException(
 2132                    $"Invalid Protobuf compression flag {compressionFlag}; expected 0 or 1.");
 133        }
 65823134        int messageLength = DecodeMessageLength(readResult.Buffer.Slice(1, 4), maxMessageLength);
 65821135        reader.AdvanceTo(readResult.Buffer.GetPosition(5));
 136
 137        T message;
 138
 139        try
 65821140        {
 65821141            if (messageLength > 0)
 65797142            {
 65797143                readResult = await reader.ReadAtLeastAsync(messageLength, cancellationToken).ConfigureAwait(false);
 144                // We never call CancelPendingRead; an interceptor or middleware can but it's not correct.
 65797145                if (readResult.IsCanceled)
 0146                {
 0147                    throw new InvalidOperationException("Unexpected call to CancelPendingRead.");
 148                }
 149
 65797150                if (readResult.Buffer.Length < messageLength)
 0151                {
 0152                    throw new InvalidDataException(
 0153                        $"The payload has {readResult.Buffer.Length} bytes, but {messageLength} bytes were expected.");
 154                }
 155
 156                // ParseFrom reads tags until end-of-input and throws InvalidProtocolBufferException on a truncated
 157                // field, so passing an exact-length slice means it either consumes every byte or throws.
 65797158                message = messageParser.ParseFrom(readResult.Buffer.Slice(0, messageLength));
 65795159                reader.AdvanceTo(readResult.Buffer.GetPosition(messageLength));
 65795160            }
 161            else
 24162            {
 24163                message = messageParser.ParseFrom([]);
 24164            }
 65819165        }
 2166        catch (InvalidProtocolBufferException exception)
 2167        {
 2168            throw new InvalidDataException("Failed to decode the Protobuf message.", exception);
 169        }
 65819170        return message;
 171
 172        static int DecodeMessageLength(ReadOnlySequence<byte> buffer, int maxMessageLength)
 65823173        {
 65823174            Debug.Assert(buffer.Length == 4);
 65823175            Span<byte> spanBuffer = stackalloc byte[4];
 65823176            buffer.CopyTo(spanBuffer);
 177            // The Protobuf envelope encodes the length as an unsigned 32-bit big-endian integer. The cast of
 178            // maxMessageLength to uint is safe because it is a non-negative int, and covers both the > int.MaxValue
 179            // case and the > maxMessageLength case in a single comparison.
 65823180            uint messageLength = BinaryPrimitives.ReadUInt32BigEndian(spanBuffer);
 65823181            if (messageLength > (uint)maxMessageLength)
 2182            {
 2183                throw new InvalidDataException("The message length exceeds the maximum value.");
 184            }
 65821185            return (int)messageLength;
 65821186        }
 65840187    }
 188}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

DecodeProtobufMessageAsync()
ToAsyncStream(System.IO.Pipelines.PipeReader,Google.Protobuf.MessageParser`1<T>,System.Int32)
ReadProtobufMessageAsync()
DecodeMessageLength()