Streaming and Errors

Server, client and bidirectional streaming, status codes instead of exceptions, and interceptors for cross-cutting work.

Syntax

public override async Task Method(TRequest request, IServerStreamWriter<TReply> responseStream, ServerCallContext context)

Any rpc can carry a stream of messages instead of a single one. Put the stream keyword on the request, on the reply, or on both in the contract, and the generated method changes shape to match.

The three shapes

Server streaming takes the request plus an IServerStreamWriter<T>, and each WriteAsync sends one message. Client streaming takes an IAsyncStreamReader<T> and returns one reply after reading the requests. Bidirectional streaming takes both. A reader and a writer each allow one thread at a time, and the call ends when your method returns, so a background task must finish before that or its writes land on a closed call. In the example, Count with a limit of 3 streams values 1, 2 and 3 and ends with status OK; a limit of 0 streams nothing and ends with InvalidArgument.

Errors are statuses

Every call ends with a status code. OK means success. To send a specific failure, throw RpcException with a Status: InvalidArgument, NotFound, PermissionDenied, Unavailable. Any other exception also ends the call, but as Unknown with the fixed text Exception was thrown by handler, because the real message could leak internals. EnableDetailedErrors sends the real one and belongs in development only. On the client, awaiting the call throws RpcException, and StatusCode is what to branch on.

Interceptors

An interceptor is a class deriving from Interceptor that wraps every call on a server or a channel: logging, metrics, authentication, turning domain exceptions into statuses. Override the method for the shape you want, such as UnaryServerHandler, and call continuation to let the call proceed. Server interceptors are created per request from the service container unless you register the type yourself.

Example

// Protos/count.proto: package count.v1; option csharp_namespace = "Count.V1";
// service Counter { rpc Count (CountRequest) returns (stream CountReply); }
// message CountRequest { int32 limit = 1; }
// message CountReply { int32 value = 1; }

using Count.V1;
using Grpc.Core;

var builder = WebApplication.CreateBuilder(args);

// true would send real exception messages to callers. Keep it false outside development.
builder.Services.AddGrpc(options => options.EnableDetailedErrors = false);

var app = builder.Build();
app.MapGrpcService<CounterService>();
app.Run();

public class CounterService : Counter.CounterBase
{
    public override async Task Count(
        CountRequest request,
        IServerStreamWriter<CountReply> responseStream,
        ServerCallContext context)
    {
        if (request.Limit < 1)
        {
            // Ends the call with this code before a single message is written.
            throw new RpcException(new Status(StatusCode.InvalidArgument, "limit must be 1 or more"));
        }

        for (var i = 1; i <= request.Limit; i++)
        {
            // The caller may walk away. Stop writing when the call is cancelled.
            if (context.CancellationToken.IsCancellationRequested)
            {
                return;
            }

            await responseStream.WriteAsync(new CountReply { Value = i });
        }
    }
}

When to use it

Stream a long result set to a caller instead of building one large reply
Accept a stream of uploaded samples and reply with one summary
Return NotFound or PermissionDenied so callers can act on the reason, not a generic failure
Log every failed call in one place rather than in every service method

More examples

Reading a server stream on the client

A streaming client method returns immediately with a call object; the messages arrive as you read them. The failure status can arrive at any point, including midway through, so the loop belongs inside the try rather than the other way round. ReadAllAsync and the MoveNext and Current pair are the same mechanism, and seeing both matters because older code and other gRPC languages use the longhand form.

using Count.V1;
using Grpc.Core;
using Grpc.Net.Client;

using var channel = GrpcChannel.ForAddress("https://localhost:5001");
var client = new Counter.CounterClient(channel);

// The call object is disposable. Disposing it early cancels the stream.
using var call = client.Count(new CountRequest { Limit = 3 });

try
{
    // Shorthand: ReadAllAsync() turns the reader into an async sequence.
    await foreach (CountReply reply in call.ResponseStream.ReadAllAsync())
    {
        Console.WriteLine(reply.Value);
    }

    // The same loop longhand, using the reader directly. Only one of these
    // can run per call, because reading consumes the stream.
    // while (await call.ResponseStream.MoveNext(CancellationToken.None))
    // {
    //     Console.WriteLine(call.ResponseStream.Current.Value);
    // }
}
catch (RpcException ex) when (ex.StatusCode == StatusCode.InvalidArgument)
{
    Console.WriteLine($"rejected: {ex.Status.Detail}");
}

// Prints 1, 2 and 3 on separate lines.

Client streaming and bidirectional streaming

Client streaming suits uploads: a sensor sending samples, or a batch of rows, where the caller sends many messages and wants one summary back. Bidirectional streaming suits a long-lived conversation, such as a client feeding readings in while the server pushes alerts out. The reader and the writer are safe to use from different threads from each other, which is what lets a bidirectional method read and write at the same time, but each one on its own is single-threaded.

using Grpc.Core;
using Metrics.V1;

public class MetricsService : Metrics.MetricsBase
{
    // rpc Upload (stream Sample) returns (UploadReply);
    // Many requests in, one reply at the end.
    public override async Task<UploadReply> Upload(
        IAsyncStreamReader<Sample> requestStream,
        ServerCallContext context)
    {
        var count = 0;
        double total = 0;

        await foreach (Sample sample in requestStream.ReadAllAsync())
        {
            count++;
            total += sample.Value;
        }

        return new UploadReply { Count = count, Total = total };
    }

    // rpc Watch (stream Sample) returns (stream Alert);
    // Reads and writes on the same call. One reply per request here,
    // but the two directions are independent in general.
    public override async Task Watch(
        IAsyncStreamReader<Sample> requestStream,
        IServerStreamWriter<Alert> responseStream,
        ServerCallContext context)
    {
        await foreach (Sample sample in requestStream.ReadAllAsync())
        {
            if (sample.Value > 100)
            {
                await responseStream.WriteAsync(new Alert { Name = sample.Name });
            }
        }
    }
}

A server interceptor that logs failed calls

One class now covers every unary method on every service, which is the point: this is where logging, metrics and authorisation checks belong rather than repeated at the top of each method. Interceptors run in registration order, and global ones run before per-service ones added with AddServiceOptions. Override ServerStreamingServerHandler, ClientStreamingServerHandler or DuplexStreamingServerHandler to cover the streaming shapes; a client-side interceptor overrides AsyncUnaryCall and friends instead, and is attached with AddInterceptor or channel.Intercept.

using Grpc.Core;
using Grpc.Core.Interceptors;
using Microsoft.Extensions.Logging;

public class StatusLoggingInterceptor : Interceptor
{
    private readonly ILogger<StatusLoggingInterceptor> _logger;

    public StatusLoggingInterceptor(ILogger<StatusLoggingInterceptor> logger)
    {
        _logger = logger;
    }

    // TRequest and TResponse come from the base class, one pair per call.
    public override async Task<TResponse> UnaryServerHandler<TRequest, TResponse>(
        TRequest request,
        ServerCallContext context,
        UnaryServerMethod<TRequest, TResponse> continuation)
    {
        try
        {
            // continuation runs the next interceptor, or the service method.
            return await continuation(request, context);
        }
        catch (RpcException ex)
        {
            _logger.LogWarning("{Method} failed with {StatusCode}", context.Method, ex.StatusCode);
            throw;
        }
    }
}

// In Program.cs:
// builder.Services.AddGrpc(options =>
// {
//     options.Interceptors.Add<StatusLoggingInterceptor>();
// });

Discussion

Be the first to comment on this lesson.

The three shapes

Errors are statuses

Interceptors

Example

When to use it

More examples

Reading a server stream on the client

Client streaming and bidirectional streaming

A server interceptor that logs failed calls

Related

Discussion