‫بررسی قابلیت‌های جریان داده در MediatR با استفاده از IStreamRequestHandler

Stream Requests چیست؟

• مدیریت حجم بالای داده‌ها: انتقال داده‌ها بدون اشغال حجم زیادی از حافظه.
• پردازش عملیات طولانی: امکان ارسال داده‌ها به مرور زمان، حتی قبل از تکمیل کل پردازش.
• پشتیبانی از ارتباط‌های جریانی: مناسب برای پروتکل‌هایی مثل WebSocket یا gRPC.

تعریف IStreamRequestHandler

public interface IStreamRequestHandler<TRequest, TResponse>
    where TRequest : IStreamRequest<TResponse>
{
    IAsyncEnumerable<TResponse> Handle(TRequest request, CancellationToken cancellationToken);
}

توضیحات:

• TRequest: نوع درخواست که باید از IStreamRequest<TResponse> ارث‌بری کند.
• TResponse: نوع داده‌هایی که به صورت جریانی بازگردانده می‌شوند.
• Handle: متدی برای تولید جریان داده‌ها به صورت IAsyncEnumerable.

نکات کلیدی در مورد IStreamRequestHandler

1. کارایی حافظه (Memory Efficiency): داده‌ها به صورت تکه‌تکه پردازش و ارسال می‌شوند، که حافظه کمتری مصرف می‌کند.
2. پشتیبانی از لغو (Cancellation): استفاده از CancellationToken برای توقف پردازش در صورت نیاز.
3. عملکرد ناهمزمان (Asynchronous): بسیار مناسب برای عملیات‌هایی مانند کوئری‌های پایگاه داده یا درخواست به APIهای خارجی.
4. بهینه‌سازی عملکرد (Performance Optimization): این الگو برای داده‌های با حجم بالا بسیار کاربردی و مؤثر است.

موارد استفاده عملی

• جستجوهای جریانی: بازگرداندن مجموعه داده‌ها از پایگاه داده به صورت تکه‌تکه.
• به‌روزرسانی‌های زنده (Live Updates): سیستم‌های Real-Time مثل SignalR یا WebSocket.
• پردازش‌های طولانی‌مدت: مدیریت داده‌هایی که به زمان زیادی برای پردازش نیاز دارند.
• gRPC Streaming: ارسال جریان داده‌ها به کلاینت‌های gRPC.

مثال 1: بازگرداندن لیست محصولات به صورت جریانی

public class GetProductsStream : IStreamRequest<Product>
{
    public int CategoryId { get; set; }
}

public class GetProductsStreamHandler : IStreamRequestHandler<GetProductsStream, Product>
{
    private readonly ProductDbContext _dbContext;

    public GetProductsStreamHandler(ProductDbContext dbContext)
    {
        _dbContext = dbContext;
    }

    public async IAsyncEnumerable<Product> Handle(GetProductsStream request, [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        var products = _dbContext.Products
                                 .Where(p => p.CategoryId == request.CategoryId)
                                 .AsAsyncEnumerable();

        await foreach (var product in products.WithCancellation(cancellationToken))
        {
            yield return product;
        }
    }
}

public async Task StreamProducts(IMediator mediator, int categoryId)
{
    var stream = mediator.CreateStream(new GetProductsStream { CategoryId = categoryId });

    await foreach (var product in stream)
    {
        Console.WriteLine($"Product: {product.Name}");
    }
}

مثال 2: Live Updates در سیستم‌های Real-Time

public class GetOrderUpdates : IStreamRequest<OrderUpdate>
{
    public int UserId { get; set; }
}

public class GetOrderUpdatesHandler : IStreamRequestHandler<GetOrderUpdates, OrderUpdate>
{
    private readonly OrderUpdateService _orderUpdateService;

    public GetOrderUpdatesHandler(OrderUpdateService orderUpdateService)
    {
        _orderUpdateService = orderUpdateService;
    }

    public async IAsyncEnumerable<OrderUpdate> Handle(GetOrderUpdates request, [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (var update in _orderUpdateService.GetOrderUpdatesAsync(request.UserId, cancellationToken))
        {
            yield return update;
        }
    }
}

public class OrderHub : Hub
{
    private readonly IMediator _mediator;

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

    public async Task StreamOrderUpdates(int userId)
    {
        var stream = _mediator.CreateStream(new GetOrderUpdates { UserId = userId });

        await foreach (var update in stream)
        {
            await Clients.Caller.SendAsync("ReceiveOrderUpdate", update);
        }
    }
}

const connection = new signalR.HubConnectionBuilder()
    .withUrl("/orderHub")
    .build();

connection.on("ReceiveOrderUpdate", (update) => {
    console.log("Order Update:", update);
});

connection.start().then(() => {
    connection.invoke("StreamOrderUpdates", userId);
});

مثال 3: gRPC Streaming

public class GetWeatherUpdates : IStreamRequest<WeatherUpdate>
{
    public string City { get; set; }
}

public class GetWeatherUpdatesHandler : IStreamRequestHandler<GetWeatherUpdates, WeatherUpdate>
{
    private readonly WeatherService _weatherService;

    public GetWeatherUpdatesHandler(WeatherService weatherService)
    {
        _weatherService = weatherService;
    }

    public async IAsyncEnumerable<WeatherUpdate> Handle(GetWeatherUpdates request, [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (var update in _weatherService.GetUpdatesAsync(request.City, cancellationToken))
        {
            yield return update;
        }
    }
}

پیاده‌سازی gRPC Service

public class WeatherGrpcService : WeatherService.WeatherServiceBase
{
    private readonly IMediator _mediator;

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

    public override async Task GetWeatherStream(GetWeatherRequest request, IServerStreamWriter<WeatherUpdateResponse> responseStream, ServerCallContext context)
    {
        var stream = _mediator.CreateStream(new GetWeatherUpdates { City = request.City });

        await foreach (var update in stream.WithCancellation(context.CancellationToken))
        {
            await responseStream.WriteAsync(new WeatherUpdateResponse
            {
                Temperature = update.Temperature,
                Humidity = update.Humidity,
                Timestamp = update.Timestamp.ToString()
            });
        }
    }
}

تعریف gRPC Proto

syntax = "proto3";

service WeatherService {
    rpc GetWeatherStream(GetWeatherRequest) returns (stream WeatherUpdateResponse);
}

message GetWeatherRequest {
    string city = 1;
}

message WeatherUpdateResponse {
    double temperature = 1;
    double humidity = 2;
    string timestamp = 3;
}

کلاینت gRPC (C#):

var channel = GrpcChannel.ForAddress("https://localhost:5001");
var client = new WeatherService.WeatherServiceClient(channel);

using var stream = client.GetWeatherStream(new GetWeatherRequest { City = "Tehran" });

await foreach (var update in stream.ResponseStream.ReadAllAsync())
{
    Console.WriteLine($"Temperature: {update.Temperature}, Humidity: {update.Humidity}");
}

نتیجه‌گیری