Streaming en gRPC, parte I: Streaming unidireccional

Este post es parte del segundo calendario de adviento de C# en español.

Hace bien poco hablábamos de la introducción en .NET/C# de la interfaz IAsyncEnumerable, y de las interesantes posibilidades que abría vistas a la producción y consumo de streams de mensajes.

También hace unos días dimos un repaso al soporte para la implementación de clientes y servidores gRPC lanzado con ASP.NET Core 3. En dicho post hacíamos una pequeña introducción al desarrollo de este tipo de servicios, basados en HTTP/2 y el estándar protobuf.

Como ya comentábamos entonces, a diferencia de los tradicionales servicios tipo REST (HTTP/JSON), gRPC soporta el intercambio de datos en modo streaming, tanto unidireccional como bidireccionalmente. Es decir, en el escenario más complejo podríamos abrir un canal gRPC a través del cual un cliente podría ir enviando paquetes de información de forma asíncrona, y al mismo tiempo utilizarlo para recibir un flujo de datos continuo desde el servidor.

Hoy vamos a ver cómo se combinan estas características viendo un ejemplo de implementación de un servicio gRPC que utiliza streaming unidireccional, en sentido servidor-cliente.

Para ello crearemos un servicio generador de números en streaming que funcionará de la siguiente forma:

• Los clientes invocarán un procedimiento del servidor suministrándole un número entero inicial y un delay expresado en segundos.
• Como respuesta, el servidor creará un stream por el que irá enviando, según la periodicidad indicada, números consecutivos partiendo del especificado en la llamada inicial.
• El proceso finalizará, cerrando en stream, cuando el servidor haya generado 10 números, o bien cuando el cliente sea detenido.

Implementación del lado servidor

Partiendo de un proyecto creado usando la plantilla para servicios ASP.NET Core gRPC, en primer lugar le añadiremos el contrato protobuf del servicio a implementar en el archivo Protos/NumberGenerator.proto, con el siguiente contenido:

syntax = "proto3";

option csharp_namespace = "DemoGrpc";

service NumberGenerator {
  rpc Generate(GenerationOptions) returns (stream GeneratedNumber);
}

message GenerationOptions {
  int32 start = 1;
  int32 delaySeconds = 2;
}

message GeneratedNumber {
  int32 number = 1;
  int64 generatedAtTicks = 2;
}

Aunque no seamos grandes expertos en protobuf, seguro que podemos entender la especificación del servicio. El servicio NumberGenerator define un procedimiento remoto llamado Generate(), que recibe un mensaje de tipo GenerationOptions y retorna un stream de mensajes de tipo GeneratedNumber. Ambos tipos de datos están también definidos en el mismo .proto de forma bastante clara.

Partiendo de este contrato protobuf, el tooling de gRPC para .NET Core generará la clase abstracta DemoGrpc.NumberGenerator.NumberGeneratorBase al compilar el proyecto. Para implementar la lógica de nuestro servicio, simplemente debemos heredar de dicha clase y sobrescribir el método Generate(), por ejemplo como sigue:

public class NumberGeneratorService : NumberGenerator.NumberGeneratorBase
{
    public override async Task Generate(
              GenerationOptions request, 
              IServerStreamWriter<GeneratedNumber> responseStream, 
              ServerCallContext context)
    {
        var current = request.Start;
        while (!context.CancellationToken.IsCancellationRequested)
        {
            var number = new GeneratedNumber()
            {
                Number = current++,
                GeneratedAtTicks = DateTime.Now.Ticks
            };
            await responseStream.WriteAsync(number);
            if (current - request.Start == 10)
                break;
            await Task.Delay(request.DelaySeconds * 1000);
        }
    }
}

Creo que el código habla por si mismo :) Como podéis ver, se trata únicamente de un bucle infinito que va "disparando" enteros a través del stream, introduciendo una espera entre mensaje y mensaje.

Sólo se contemplan dos salidas posibles para el bucle, lo que asegura la finalización de la llamada al generador: cuando el CancellationToken se activa, que indica que el cliente ha dejado de estar a la escucha, o cuando hemos generado diez números.

Bien, hecho esto ya sólo nos falta registrar el endpoint en la clase Startup de ASP.NET Core, de forma que las peticiones al servicio sean rutadas correctamente:

public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    ...
    app.UseRouting();
    ...
    app.UseEndpoints(endpoints =>
    {
        endpoints.MapGrpcService<NumberGeneratorService>();
        ...
    }
}

Implementación del lado cliente

Podríamos implementar un cliente para el servicio anterior en cualquier tipo de proyecto .NET Core, pero lo haremos en forma de aplicación de consola por simplificar.

Lo único especial que necesitamos en ella es añadir una referencia al archivo .proto que contiene la definición del servicio, así como a los siguientes paquetes NuGet:

• Google.Protobuf
• Grpc.Net.Client
• Grpc.Tools

La implementación del cliente del servicio podría ser la siguiente:

class Program
{
    static async Task Main(string[] args)
    {
        var channel = GrpcChannel.ForAddress("https://localhost:5001");
        var client = new NumberGenerator.NumberGeneratorClient(channel);

        var options = new GenerationOptions()
        {
            Start = 10,
            DelaySeconds = 2
        };
        using (var response = client.Generate(options))
        {
            await foreach (var number in response.ResponseStream.ReadAllAsync())
            {
                var time = new DateTime(number.GeneratedAtTicks);
                Console.WriteLine($"{number.Number} generated at {time:T}");
            }
        }
        Console.WriteLine("Finished. Press a key to close.");
        Console.ReadKey();
    }
}

En este código podemos ver la fontanería necesaria para crear el cliente del servicio al que deseamos acceder. Tras ello, y quizás es la parte más llamativa del código, vemos el uso de await foreach sobre el stream asíncrono retornado por la llamada al método ReadAllAsync().

Este bucle foreach iterará mientras en el lado servidor continúe la ejecución del método invocado; en otras palabras, finalizará cuando el servicio haya generado los diez números (o bien cuando el servidor sea detenido).



La verdad es que es alucinante lo sencillo que resulta implementar un escenario streaming unidireccional con los componentes gRPC de ASP.NET Core, porque permiten que nos centremos en nuestro código al aislarnos de toda la complejidad que hay por debajo.

Pero aún daremos una vuelta de tuerca más: en el próximo post veremos cómo implementar streaming bidireccional, es decir, servicios donde cliente y servidor puedan enviar y recibir mensajes de forma asíncrona.

Publicado unidireccionalmente en Variable not found.