Introducción

Las tareas en segundo plano (AKA Background Jobs) son sumamente usadas hoy en día gracias a la facilidad que las nuevas tecnologías y herramientas nos proveen. Tanto la nube como frameworks como .NET, tienen lo necesario para poder tener tareas que corran en procesos distintos al de la aplicación principal.

De esa forma, distribuir la carga de trabajo, reaccionar a eventos o simplemente desacoplar aplicaciones, resulta más sencillo de lo que suena.

En este post veremos como tener una Web API que manda trabajo a un proceso distinto que está hospedado en un lugar distinto.

El código completo, lo puedes encontrar aquí y como siempre, puedes seguirme en @balunatic para cualquier pregunta.

Herramientas Requeridas

Para este post voy a mencionar las herramientas que necesitamos.

Visual Studio y Desarrollo en Azure: Si usas Visual Studio, debes de tener las herramientas de "Desarrollo Azure":
- Aquí automáticamente los emuladores de Storage Account se encontrarán ya instalados.
Azure Functions Tools: Siempre es recomendable tener el host instalado, para correrlo desde consola de ser necesario. Ver más.
Azurite: Si no estas usando visual studio y quieres usar VS Code, puedes hacerlo descargando/corriendo Azurite desde:
- Extensión de VS Code
- npm con npm install -g azurite
- docker con docker pull mcr.microsoft.com/azure-storage/azurite
- O finalmente, usando el código fuente de github
- Más información aquí

Nota 👀: En el post de Audit Logs ya hemos usado Azurite / Storage Emulator. Técnicamente, lo deberías de tener con visual studio, si no, ya sabes cómo instalarlo.

Background jobs

Muchos tipos de aplicaciones requieren que ciertas tareas se ejecuten independientemente de lo que ocurre en la UI. Los background jobs se ejecutan sin necesitar interacción con el usuario, algún trigger inicia el background job y este continúa según se requiera hasta terminar.

Existen varios tipos de background jobs:

Tareas que requieren mucho CPU, como cálculos matemáticos o análisis de modelos
Tareas que requieren mucho Input/Output
Tareas programadas
Tareas con duración prolongada

Estas tareas asíncronas pueden ser "disparadas" de muchas formas, pero se resumen en dos categorías:

Event-driven triggers. La tarea es iniciada en respuesta a un evento, típicamente es una acción de un usuario o un siguiente paso en algún workflow disparado por otro sistema/proceso.
Schedule-driven triggers. La tarea es iniciada basada en una programación de tiempo. Podría ser algo que se ejecuta todas las mañanas o los fines de semana, etc.

Event-driven triggers

Existen varias formas de iniciar tareas en segundo plano:

Algún sistema o UI manda un mensaje por medio de un Queue. El mensaje contiene datos sobre la acción que acaba de suceder, como, por ejemplo, un usuario compró algo y se tiene que procesar el pedido. El background job está en escucha del Queue para estar atento cuando lleguen mensajes nuevos.
Algún sistema o UI guarda/actualiza información en algún almacenamiento. El background job puede estar monitoreando alguna persistencia (como una base de datos) en espera de que suceda algo.
Por medio de llamadas HTTP a un endpoint, como una API. Por medio del endpoint recibe la información pertinente e inicia el procesamiento.

Schedule-driven triggers

Existen varias formas de tareas según una programación de tiempo:

La más común es utilizando expresiones cron

Las tareas programadas son muy comunes, las he usado para procesar pagos de suscripciones. Cada cierto tiempo ejecuto una tarea para determinar si ya es tiempo de cobrarle a los usuarios por alguna suscripción.

También es muy normal utilizarlas para limpieza de base de datos, tal vez a fin de mes queremos reorganizar índices y entre otras aplicaciones que hay.

¿Para qué usar Background Jobs?

Implementar background jobs ayuda principalmente para ofrecer una mejor experiencia al usuario, a veces no se requiere tener esperando al usuario a que una tarea termine. Simplemente la tarea inicia porque el usuario la "disparó" pero esta continúa de manera asíncrona hasta terminar.

Un ejemplo muy sencillo es cuando compras algo en Amazon, por lo general es un proceso asíncrono porque amazon no te cobra hasta saber que el producto está disponible y muchas otras condiciones.

Tareas que duran mucho es muy normal agregarlas en alguna "cola" para que esta se vaya ejecutando con forme se avance otras colas.

También, a veces realmente no es necesario tener al usuario en espera. Ordenar una pizza solo requiere decir que pizza quieres y a partir de ahí el trabajo está en la cocina. Por lo que al final, recibes feedback.

Hay que notificar que una tarea terminó se puede hacer de muchas formas, como mensaje SMS, correo electrónico o una push notification. Realmente depende del tipo de aplicación.

¿Qué es Azure Queue Storage?

Azure Queue Storage nos permite implementar mensajes entre aplicaciones. No es un Message Broker o un Service Bus como otros servicios, Queue Storage es una forma simplificada de mensajería por medio de un almacenamiento en Azure. Pero de igual forma, nos permite mandar mensajes entre distintas aplicaciones y así conectarlos para que interactúen.

Su fácil implementación lo hace atractivo para aplicaciones sin tantos requerimientos, porque no siempre vas a necesitar de un Message Broker o un Service bus.

He usado Queue Storage para distintos proyectos, pero el principal que me viene a la mente es un procesador de pagos y suscripciones que mencioné antes.

Si un Queue falla, este se reintenta, por lo que debemos de tener cuidado si la aplicación falla. Es importante manejar la idempotencia.

Azure Storage Account

Queue Storage son parte de los Storage Accounts de Azure. Un Storage Account cuenta con 4 servicios principales:

Blobs: Almacenamiento de archivos de cualquier tipo
Tables: Información semiestructurada, muy util para big-data
Queues: Almacenamiento de mensajes
Files: Es como montar discos de red en la nube

De igual forma, por ahora solo nos competen los Queues:

Un Storage Account permite tener Teras de información guardada, por lo que nos permite (exageradamente) guardar millones de mensajes. Que al final, los Queues se van liberando con forme se van procesando, así que el almacenamiento no es un concern aquí.

Arquitectura a implementar

Estamos hablando de la parte 12 de esta serie de publicaciones, seguiremos trabajando con el código que ya hemos realizado hasta hoy. De igual forma, aquí de dejo el respositorio para que revises todo el código.

La Web API es la que se encargará recibir solicitudes HTTP de cualquier cliente autenticado. Este mandará mensajes al Queue Storage para que posteriormente el Background Job (AKA Azure Function) lea los mensajes y los procese.

Actualmente ya tenemos una base de datos, un Storage Account y una API. Lo que agregaremos ahora será un Azure Function que estará preguntando si existen mensajes nuevos en el queue.

Azure Functions

Azure Functions es una solución serverless que nos permite escribir menos código, mantener menos infraestructura y ahorrar costos. En lugar de preocuparnos en como deployear y mantener servidores, la infraestructura en la nube nos provee de todos los recursos que necesitamos para siempre enfocarnos en los más importante, en la funcionalidad de nuestro software.

Las funciones de Azure son invocadas por diferentes Triggers (eventos o programados), estos triggers se pueden conectar directamente con distintas tecnologías e infraestructuras de Azure, es muy útil porque como se menciona antes, te ahorra mucho código y tiempo.

Generalmente usas Azure Functions para cuestiones muy específicas, no significa que escribirás toda una API utilizando Azure Functions (aunque podrías) pero aquí te van varios casos de uso:

Si quieres hacer...	Entonces...
Una Web API	Puedes implementar un endpoint con el Trigger HTTP
Procesar archivos subidos	Puedes triggerear un a función cuando se suban archivos en algún storage account
Un workflow	Puedes encadenar funciones utilizando durable functions
Responder a cambios en la BD	Puedes correr lógica personalizada cuando algo pasa en Cosmos DB
Correr tareas programadas	Puedes ejecutar código según expresiones cron
Crear mensajes en Queues	Puedes recibir mensajes usando Queue Storage, Service Bus o Event hubs
Streams IoT	Puedes recolectar información de dispositivos IoT por medio de Azure IoT Hub
Procesar información en tiempo real	Puedes conectar SignalR con tu Azure Function

Sin más preámbulo, veamos que implementaremos hoy...

Procesando Órdenes/Compras con Queues.

El Background Job que haremos en esta ocasión trata de un procesador de compras. Hemos trabajado en los demos anteriores el manejo de productos, por lo que tiene sentido poder ahora "comprarlos".

Crearemos órdenes llamadas Checkout desde la Web API, se creará y se "aceptará" la orden. Después inmediatamente la Web API mandará un mensaje por medio del Queue Storage para que un Azure Function (que está en espera de mensajes) procese la solicitud y haga el trabajo necesario (que, en teoría, sería revisar stock, realizar cobro, facturar, etc).

Siempre te recomiendo que estudies el código que comparto sobre cada publicación, ya que siempre omito detalles de otros temas que, aunque son importantes, no son relevantes al tema del post.

Por lo que, de forma sencilla, lo que agregué en este post para poder hacer las Azure Functions fueron dos entities nuevos: Checkouty CheckoutProduct. Ya contabamos con la clase Product, ahora hicimos una relación entre ellos.

Contrato IQueueService

Este contrato nos servirá para abstraer los detalles de implementación del Queue. Hacerlo de esta forma incluso podríamos cambiar de Queue Storage a RabbitMQ sin tener que modificar el Core de la aplicación.

Por lo que siguiendo la ruta ApplicationCore > Common > Services, tenemos lo siguiente:



namespace MediatrExample.ApplicationCore.Common.Services;
public interface IQueuesService
{
    Task QueueAsync<T>(string queueName, T item);
}

El tipo genérico nos permitirá enviar cualquier tipo de dato como mensaje.

Implementación AzureStorageQueueService

Para usar Queue Storage necesitamos la siguiente librería dentro de ApplicationCore:



dotnet add package Azure.Storage.Queues

Nota 👀: Recuerda que estamos siguiendo un estilo "Vertical Slice Architecture" por lo que casi todo, va en el Application Core

Siguiendo la ruta ApplicationCore > Infrastructure > Services > AzureQueues, tenemos lo siguiente:



using ...;

namespace MediatrExample.ApplicationCore.Infrastructure.Services.AzureQueues;

public class AzureStorageQueueService : IQueuesService
{
    private readonly IConfiguration _config;

    public AzureStorageQueueService(IConfiguration config)
    {
        _config = config;
    }

    public async Task QueueAsync<T>(string queueName, T item)
    {
        var queueClient = new QueueClient(_config.GetConnectionString("StorageAccount"), queueName, new QueueClientOptions
        {
            MessageEncoding = QueueMessageEncoding.Base64
        });

        queueClient.CreateIfNotExists();

        var message = JsonSerializer.Serialize(item);
        await queueClient.SendMessageAsync(message);
    }
}

Por ahora, solo necesitamos enviar mensajes a un Queue. La Lectura se encargará el Azure Function y su Trigger que ya está programado para eso (que más delante lo veremos)

Lo que sucede aquí es muy sencillo, lo único Importante es que el Azure Function está esperando que los mensajes lleguen en Base 64 (no sé por qué 😁).

La connection string es "StorageAccount": "UseDevelopmentStorage=true" por lo que, de nuevo, es importante tener el simulador de storage corriendo

Mandando Mensajes al Queue

Realmente ya tenemos todo "listo" para empezar a mandar mensajes, solo hay que crear un par de Comandos nuevos (recuerden que todo esto es sigiendo CQRS y Vertical Slice Architecture).

Features > Checkouts > Commands > NewCheckout

Para "disparar" este Queue, primero hay que crear una orden, que es la que en teoría nuestro background job debe de obtener para procesar:



using ...;

namespace MediatrExample.ApplicationCore.Features.Checkouts.Commands;

[AuditLog] // <--- del post "Sistema Audiable"
public class NewCheckoutCommand : IRequest
{
    public List<NewCheckoutItems> Products { get; set; } = new();

    public class NewCheckoutItems
    {
        public string ProductId { get; set; } = default!;
        public int Quantity { get; set; }
    }
}

public class NewCheckoutCommandHandler : IRequestHandler<NewCheckoutCommand>
{
    private readonly MyAppDbContext _context;
    private readonly IQueuesService _queuesService;
    private readonly CurrentUser _user;

    public NewCheckoutCommandHandler(MyAppDbContext context, ICurrentUserService currentUserService, IQueuesService queuesService)
    {
        _context = context;
        _queuesService = queuesService;
        _user = currentUserService.User;
    }

    public async Task<Unit> Handle(NewCheckoutCommand request, CancellationToken cancellationToken)
    {
        Checkout newCheckout = await CreateCheckoutAsync(request, cancellationToken);
        await QueueCheckout(newCheckout);

        return Unit.Value;
    }
}

Nuestro comando recibe como parámetro de entrada un listado de productos que se quieren comprar, el handler procesa esta solicitud en dos partes: CreateCheckoutAsync y QueueCheckout.

Este comando tiene como dependencias:

MyAppDbContext: Para acceso a la base de datos
IQueueService: La abstracción para mandar mensajes a un Queue
ICurrentUserSerivce: Necesitamos saber que usuario autenticado está haciendo la compra



private async Task<Checkout> CreateCheckoutAsync(NewCheckoutCommand request, CancellationToken cancellationToken)
{
    var newCheckout = new Checkout
    {
        CheckoutDateTime = DateTime.UtcNow,
        UserId = _user.Id,
        Total = 0
    };

    foreach (var item in request.Products)
    {
        var product = await _context.Products.FindAsync(item.ProductId.FromHashId());

        if (product is null)
        {
            throw new ValidationException(new List<ValidationFailure>
            {
                new ValidationFailure("Error", $"El producto {item.ProductId} no existe")
            });
        }

        var newProduct = new CheckoutProduct
        {
            ProductId = product.ProductId,
            Quantity = item.Quantity,
            UnitPrice = product.Price,
            Total = item.Quantity * product.Price
        };

        newCheckout.Products.Add(newProduct);
    }

    newCheckout.Total = newCheckout.Products.Sum(p => p.Total);

    _context.Checkouts.Add(newCheckout);

    await _context.SaveChangesAsync(cancellationToken);
    return newCheckout;
}

Se crea ahora un nuevo Checkout con los productos que se recibieron por parámetro, si los productos no existen, se mandan errores de validación.

Nota 👀: Aquí hay cosas que se han visto en otros posts, cualquier confusión, siempre puedes preguntarme en mi twitter o abajo en los comentarios.

Y finalmente, se manda un mensaje al queue llamado new-checkouts.



  private async Task QueueCheckout(Checkout newCheckout)
  {
      await _queuesService.QueueAsync("new-checkouts", new NewCheckoutMessage
      {
          CheckoutId = newCheckout.CheckoutId
      });
  }

NewCheckoutMessage es una clase que simplemente contiene el Id del Checkout que se acaba de crear. Esta se encuentra en Common > Messages, con la intención de que existan más mensajes.

DependencyInjection

Registramos una nueva "capa" de dependencias en la clase actual que ya tenemos:



public static IServiceCollection AddInfrastructure(this IServiceCollection services)
{
    services.AddScoped<IQueuesService, AzureStorageQueueService>();

    return services;
}

Hasta ahora habíamos manejado ApplicationCore, Persistence y Security. Ahora, ya agregamos Infrastructure y claro también lo invocamos en Program.cs

Nota 👀: Aquí va a ser útil para agregar más dependencias de servicios de 3eros o infraestructura.

WebApi > CheckoutsController

Y su respecto endpoint para poder ser llamado:



using ...;

namespace MediatrExample.WebApi.Controllers;

[Authorize]
[ApiController]
[Route("api/checkouts")]
public class CheckoutsController : ControllerBase
{
    private readonly IMediator _mediator;

    public CheckoutsController(IMediator mediator)
    {
        _mediator = mediator;
    }

    /// <summary>
    /// Crea una nueva orden de productos
    /// </summary>
    /// <param name="command"></param>
    /// <returns></returns>
    [HttpPost]
    public async Task<IActionResult> NewCheckout([FromBody] NewCheckoutCommand command)
    {
        await _mediator.Send(command);

        return Accepted();
    }
}

Checkout Processor (Azure Function)

Ahora solo resta, crear este background job que estará en espera de estos mensajes que ya mandamos. Se puede hacer de distintas formas, incluso de formas para que sean cloud agnostic y que corran en cualquier nube. La verdad, yo estoy convencido que Azure es la mejor nube que hay 👀.

La forma más fácil de crear un Azure Function es desde visual studio (también se puede con func new utilizando Azure Functions CLI):

Yo lo nombré como MediatRExample.CheckoutProcessor.

Como pueden ver, tenemos muchas opciones de "Triggers", la que nos compete hoy es la Queue trigger.

Teniendo el proyecto creado, verificamos que tengamos instalados los siguientes paquetes (Visual studio me crea mal el template, no sé por qué 😒):



<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Storage" Version="4.0.4" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" Version="1.3.0" OutputItemType="Analyzer" />
<PackageReference Include="Microsoft.Azure.Functions.Worker" Version="1.5.2" />

Así tal cual, deberías de poder compilar el proyecto sin problema.

Configuración de CheckoutProcessor

La forma en como configuramos las dependencias del proyecto es con la intención de reutilizarlas si la Web API ya no es la aplicación final.

Dentro de Program del Azure Function configuraremos la Persistencia y Application Core para reutilizar todo lo que ya tenemos:



using ...;

namespace MediatRExample.CheckoutProcessor;
public class Program
{
    public static void Main()
    {
        var host = new HostBuilder()
            .ConfigureFunctionsWorkerDefaults()
            .ConfigureServices((hostBuilderContext, services) =>
            {
                var configuration = hostBuilderContext.Configuration;
                services.AddApplicationCore();
                services.AddPersistence(configuration);
                services.AddInfrastructure();

                services.AddTransient<ICurrentUserService, WorkerUserService>();
            })
            .Build();

        host.Run();
    }
}

Si recuerdan en el post de autenticación, agregamos una abstracción llamada ICurrentUserService. El Core usa esta abstracción para acceder a la información del usuario autenticado.

En este Function realmente no habrá usuarios autenticados, sino que este será un Background Job de confianza, por lo que simulamos de que está "autenticado" creando la clase WorkerUserService



using MediatrExample.ApplicationCore.Common.Interfaces;
using System;

namespace MediatRExample.CheckoutProcessor;
public class WorkerUserService : ICurrentUserService
{
    public CurrentUser User => new CurrentUser(Guid.Empty.ToString(), "CheckoutProcessor", true);

    public bool IsInRole(string roleName) => true;
}

De esta forma, cualquier infraestructura previamente configurada (como los Logs, Entities auditables y Audit logs) seguirán funcionando sin problema.

Podremos observar cómo trabaja nuestro worker como si fuera un usuario real.

En este punto, ya tenemos configurado todo, incluso MediatR y Entity Framework.

ProcessCheckoutCommand

La lógica le sigue perteneciendo al ApplicationCore, por lo que crearemos un comando que procese las solicitudes y posteriormente lo mandaremos a llamar desde este Azure Function que acabamos de crear.

Esto va en Features > Checkouts > Commands.



using ...;

namespace MediatrExample.ApplicationCore.Features.Checkouts.Commands;

[AuditLog]
public class ProcessCheckoutCommand : IRequest
{
    public int CheckoutId { get; set; }
}

public class ProcessCheckoutCommandHandler : IRequestHandler<ProcessCheckoutCommand>
{
    private readonly MyAppDbContext _context;
    private readonly ILogger<ProcessCheckoutCommandHandler> _logger;

    public ProcessCheckoutCommandHandler(MyAppDbContext context, ILogger<ProcessCheckoutCommandHandler> logger)
    {
        _context = context;
        _logger = logger;
    }

    public async Task<Unit> Handle(ProcessCheckoutCommand request, CancellationToken cancellationToken)
    {
        var checkout = await _context.Checkouts.FindAsync(request.CheckoutId);

        if (checkout is null)
        {
            throw new NotFoundException();
        }

        _logger.LogInformation(
          "Nueva orden recibida con Id {Id}", checkout.CheckoutId);
        _logger.LogInformation(
          "El usuario {UserId} ordenó {ProductCount} producto(s)", checkout.UserId, checkout.Products.Count());

        // Working
        _logger.LogInformation("Realizando cobro...");
        await Task.Delay(5000);
        _logger.LogInformation("Cobro realizado");

        checkout.Processed = true;
        checkout.ProcessedDateTime = DateTime.UtcNow;

        _logger.LogWarning("Se procesó una orden con costo total de {Total:C}", checkout.Total);

        await _context.SaveChangesAsync(cancellationToken);

        return Unit.Value;
    }
}

Aquí realmente estamos simulando trabajo, consultamos el Checkout que se recibió por parámetro (que, a su vez, vendrá de un mensaje del Queue) y después de un tiempo simulado, lo marcamos como done.

Function CheckoutProcessor

Llegó la hora de la verdad, vamos a crear (o editar el Function1 creado por default) ahora un function que reciba los mensajes:



using ...;

namespace MediatRExample.CheckoutProcessor.Functions
{
    public class CheckoutProcessor
    {
        private readonly ILogger _logger;
        private readonly IMediator _mediator;

        public CheckoutProcessor(ILoggerFactory loggerFactory, IMediator mediator)
        {
            _logger = loggerFactory.CreateLogger<CheckoutProcessor>();
            _mediator = mediator;
        }

        [Function("CheckoutProcessor")]
        public async Task Run([QueueTrigger("new-checkouts", Connection = "AzureWebJobsStorage")] string myQueueItem)
        {
            _logger.LogInformation("Nuevo mensaje recibido {Message}", myQueueItem);
            _logger.LogInformation("Procesando...");

            var message = JsonSerializer.Deserialize<NewCheckoutMessage>(myQueueItem);

            await _mediator.Send(new ProcessCheckoutCommand
            {
                CheckoutId = message.CheckoutId
            });
        }
    }
}

Esta función, está en espera de mensajes que existan en el queue new-checkouts del storage AzureWebJobStorage.

Conocemos el formato del mensaje, por lo que lo serializamos de vuelta para posteriormente mandarlo al mediador y que este ejecute el handler.

A este punto, el que "disparó" el evento ya no está en espera de un resultado, aquí se puede durar lo que se tenga que durar. El procesamiento en segundo plano es para eso, pero ya no tenemos en espera al cliente en el UI con un "Cargando...".

La razón de por qué hacemos esto, es también para poder tener estas funciones escaladas horizontalmente y de esta forma poder procesar miles y miles de mensajes de manera concurrente. La nube hace scale-out para que cumpla con nuestras exigencias de procesamiento.

Es importante mencionar, que tenemos que también incluir la configuración del appsettings.json en el local.settings.json del azure function:



{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
    "ConnectionStrings:Default": "Server=(localdb)\\mssqllocaldb;Database=MediatRExample;Trusted_Connection=True;MultipleActiveResultSets=false",
    "UseInMemory": false,
    "AuditLogs:Enabled": true,
    "AuditLogs:ConnectionString": "UseDevelopmentStorage=true"
  }
}

Como estamos reutilizando el ApplicationCore, hay cosas que debemos incluir (todo esto, de posts pasados de la serie).
En resumen, este Azure Function será auditable, mandará logs por medio de [AuditLog] y si quisieramos, también a Application Insights.

Al final, este function es como si fuera otro web api, la estamos configurando con los mismos features que hemos estado haciendo.

Probando los Queues

Y pues por fin, a probar.

Hay que correr los dos proyectos. cuando no estoy debuggeando suelo correr todo en consola porque lo puedo visualizar mejor. Por lo que dentro de WebApi corremos el clásico dotnet run y en otra consola, corremos el azure function pero con el host de las functions (el que hace la chamba de los triggers y bindings). Lo podemos correr con func host start estando dentro de CheckoutProcessor.

Vemos como a la derecha, el Azure Function Host ya detectó la función que creamos: CheckoutProcessor del tipo queueTrigger.

Para poder hacer la prueba y formar el JSON de abajo, necesitamos:

Generar un JWT con el endpoint realizado en este post
Consultas los productos demo, porque tienen HashIds como llaves promarias (visto eso aquí)

Y finalmente, la prueba:

Al llamar la API, se crea una orden pendiente de procesar (es decir Processed = false) y se manda un mensaje al Queue. La respuesta que recibo inmediatamente de la API es un HTTP 202 Accepted, pero el procesamiento ocurre en el Azure Function, ya que este puede durar más y no es necesario tener al cliente esperando.

Si vemos que pasó en consola, vemos que todo se ejecutó como se esperaba:

Primero, se manda la solicitud HTTP a la API y recibimos una respuesta de Accepted

Posteriormente, el Queue se ejecuta en el background job, o sea, el azure function.

Nota 👀: El log dice "El usuario ordenó 0 producto(s)". Esto es por que no consulté los productos con Include() 🤣.

Tal cual funcionan los pipelines de MediatR, aquí también los tenemos funcionando de la misma forma, incluso los AuditLogs:

Un AuditLog de ejemplo:



{
    "Environment": {
        "UserName": "isaac",
        "MachineName": "DELL-G5",
        "DomainName": "DELL-G5",
        "CallingMethodName": "MediatrExample.ApplicationCore.Common.Behaviours.AuditLogsBehavior\u00602\u002B\u003CHandle\u003Ed__4.MoveNext()",
        "AssemblyName": "MediatrExample.ApplicationCore, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null",
        "Culture": "es-MX"
    },
    "EventType": "ProcessCheckoutCommand",
    "StartDate": "2022-05-07T19:49:02.300967Z",
    "EndDate": "2022-05-07T19:49:08.7736892Z",
    "Duration": 6473,
    "User": {
        "Id": "00000000-0000-0000-0000-000000000000",
        "UserName": "CheckoutProcessor",
        "IsAuthenticated": true
    },
    "Request": {
        "CheckoutId": 29
    }
}

Si exploramos la Base de datos, en efecto vimos cómo pasó de crearse a las 19:49:00 y terminar de ser procesado a las 19:49:08 (8 segundos 🤭, al final es un ejemplo).

Conclusión

De esta forma, ya tenemos una arquitectura basada en Background Jobs, donde una Web API manda trabajo a un Worker que se encuentra en otro proceso y otro servidor, esto siendo facilitado gracias a Azure Functions y Queue Storage.

Esta solución fácilmente puede escalarse para que se puedan procesar miles y miles de órdenes. Al ser serverless, no te preocupas de la infraestructura, este se puede escalar tanto como la demanda lo solicite.

También es bueno decir, que, si "vendes" mucho, pues también se te cobrará más, ya que se te cobra por el tiempo de CPU si usas la modalidad Serverless de un Azure Function.

Esto me resulta muy útil, ya tengo proyectos en producción con modalidades totalmente parecidas a esto. No siempre vas a necesitar un Message Broker que a veces puede ser difícil administrar, si los requerimientos son sencillos, facilmente Queue Storage y Azure Functions son una muy buena solución.

[Parte 12] Azure Functions: Background Jobs