Introducción a Dapr .NET SDK (4/N): Pub/Sub

Continuando con la serie de posts sobre #Dapr, en este articulo nos ponemos manos a la obra con el patrón Publicador/Subscriptor (Publisher/Subscriber). Además, vamos a ver como trabajar con este Building Block en Azure y en particular con un componente basado en Azure Service Bus. Incluso ¿por que no?, usar el Azure Redis que ya usamos en posts anteriores.

A continuación, el listado completo de posts relacionados sobre los que iré trabajando y publicando:

Para comenzar, lo primero es crear un nuevo recurso en Azure de tipo Servcie Bus, con un nombre cualquiera (por ejemplo «dapr-servicebus-ns«). Podemos usar para ello, bien az cli, como se muestra a continuación, o bien, el portal de Azure, como hemos hecho en otros casos.

Importante: El sku/layer que tenemos que seleccionar para la creación de Service Bus, ha de ser Standard o Premium. Basic no soporta el uso de «topic», que es la entidad que precisamente usa Dapr.

az servicebus namespace create --resource-group <RG-NAME> --name dapr-demo-ns --location westeurope --sky <Standard|Premium>

Una vez disponemos de nuestro Service Bus, continuamos con los siguientes pasos. Y, recordemos que todo el código de ejemplo podemos encontrarlo aquí, en mi github:

Publisher

Crear un proyecto de consola y añadir la referencia (paquete nuget) «Dapr.Client«.

Crear una nueva clase «PublishEventExample.cs«, con el siguiente código:

public class PublishEventExample : Example
{
    private static readonly string pubsubName = "pubsub";
    public override string DisplayName => "Publishing Events";
    public override async Task RunAsync(CancellationToken cancellationToken)
    {
        using var client = new DaprClientBuilder().Build();
            
        string[] Summaries = new[]
        {
            "Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot"
        };
        var rng = new Random();
        var eventData = new WeatherForecast
        {
            Date = DateTime.Now,
            TemperatureC = rng.Next(-20, 55),
            Summary = Summaries[rng.Next(Summaries.Length)]
        };
        await client.PublishEventAsync(pubsubName, 
                                        topicName: "forecast",
                                        eventData,
                                        cancellationToken);
        Console.WriteLine("Published forecast event!");
    }
    public class WeatherForecast
    {
        public DateTime Date { get; set; }
        public int TemperatureC { get; set; }
        public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
        public string Summary { get; set; }
    }
}

Nota: El método que realiza la publicación del topic es «PublishEventAsync«, que además de la publicación, crea el topic si este no existe.

Crear un nuevo componente de tipo «pubsub.azure.servicebus» con el nombre «pubsub» y guardarlo en la carpeta raiz «dapr/components» con el nombre «servicebus-publish.pubsub.yaml«.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pubsub
  namespace: default
spec:
  type: pubsub.azure.servicebus
  version: v1
  metadata:
  - name: connectionString # Required    
    value: "Endpoint=sb://<SERVICEBUS-NAMESPACE>.servicebus.windows.net/;SharedAccessKeyName=<KEY-NAME>;SharedAccessKey=<KEY>"

Ejecutar el programa, mediante la siguiente instrucción:

dapr run --app-id publisher -- dotnet run 0

Subscriber

Para el subscriptor creamos una web api a partir de la plantilla de ejemplo y actualizamos el Controller como sigue:

[ApiController]
[Route("[Controller]")]
public class WeatherForecastController : ControllerBase
{       
    private readonly ILogger<WeatherForecastController> _logger;
    public WeatherForecastController(ILogger<WeatherForecastController> logger)
    {
        _logger = logger;
    }
    [Topic("pubsub","forecast")]
    [HttpPost]        
    public async Task<ActionResult> Get([FromBody] WeatherForecast weatherForecast)
    {
        _logger.LogInformation($"Forecast for today ({weatherForecast.Date.DayOfWeek}) " +
            $"is: {weatherForecast.TemperatureC}C. " +
            $"Take care about the {weatherForecast.Summary}.");
            
        return await Task.FromResult(Ok(weatherForecast));           
    }
}

Y, modificamos el «startup.cs» tal y como vemos a continuación:

public void ConfigureServices(IServiceCollection services)
{
    services.AddControllers().AddDapr();
    services.AddSwaggerGen(c =>
    {
        c.SwaggerDoc("v1", new OpenApiInfo { Title = "subscriber", Version = "v1" });
    });
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    if (env.IsDevelopment())
    {
        app.UseDeveloperExceptionPage();
        app.UseSwagger();
        app.UseSwaggerUI(c => c.SwaggerEndpoint("/swagger/v1/swagger.json", "subscriber v1"));
    }
    
    app.UseRouting();
    app.UseCloudEvents();
    app.UseAuthorization();
    app.UseEndpoints(endpoints =>
    {
        // 3) This is the Dapr subscribe handler.
        // It will automatically create an endpoint on dapr/subscribe that returns all topics that the application subscribes to (derived via the Topic attribute).
        // This is required for Dapr to know on what it has to subscribe.
        endpoints.MapSubscribeHandler();
        endpoints.MapControllers();
    });
}

En este punto es importante destacar las tres instrucciones principales (en negrita):

.AddDapr(). Integra Dapr en la pipeline de MVC y registra una instancia DaprClient en el contenedor de dependencias.
app.UseCloudEvents(). Midleware para desempaquetar cada solicitud con el Content-type application/cloudevents+json. Lo que permite la lectura directa del payload. Nota: CloudEvents es un estandard de formato de mensajes que provee mecanismos para la descripción de eventos, cuya especificación podemos encontrar aquí.
endpoints.MapSubscribeHandler(). Requerido para que Dapr pueda conocer a que tiene que susbscribirse. Automáticamente crea un endpoint en Dapr que devuelve todos los topics a los que se subscribe la aplicación (dependiendo del atributo Topic).

El subscriptor hará uso del mismo componente que el publicador, dado que toda la configuración es exactamente la misma. Por tanto, la ejecución se realiza indicando el mismo valor para el parámetro «–components-path»:

dapr run --app-id subscriber --app-port 5000 --components-path ../dapr/components -- dotnet run

Adicionalmente, podríamos evitar la dependencia del código con Dapr, haciendo uso de este otro tipo de componentes «Susbscription«. Es decir, creando un nuevo «topic» de manera declarativa como se indica a continuación:

apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:
  name: weatherforecast-subscription
spec:
  pubsubname: pubsub
  topic: forecast
  route: /weatherforecast

En este caso, eliminaríamos el atributo «Topic()» de nuestro método Post en el controlador y por contra actualizaríamos el código para trabajar directametne con la subscripción. Algo similar a como podemos ver en este ejemplo.

Finalmente, en este post, no incluimos el ejemplo para el caso particular de un componente Azure Redis dado que la configuración es la misma que ya vimos en el post (Statement management).

Happy #Dapr codding !!!

Referencias:

Introducción a Dapr .NET SDK (4/N): Pub/Sub

Publisher

Subscriber

9 comentarios sobre “Introducción a Dapr .NET SDK (4/N): Pub/Sub”

Deja una respuesta Cancelar la respuesta

Publisher

Subscriber

Comparte esto:

Me gusta esto:

Relacionado

9 comentarios sobre “Introducción a Dapr .NET SDK (4/N): Pub/Sub”

Deja una respuesta Cancelar la respuesta