112 lines
4.8 KiB
C#
112 lines
4.8 KiB
C#
using System;
|
|
using System.Text;
|
|
using System.Text.Json;
|
|
using InventoryService.Models;
|
|
using RabbitMQ.Client;
|
|
using RabbitMQ.Client.Events;
|
|
|
|
namespace InventoryService.Services;
|
|
|
|
public class RabbitMqConsumerService : BackgroundService
|
|
{
|
|
private readonly string user;
|
|
private readonly string password;
|
|
private readonly string host;
|
|
private readonly int port;
|
|
private readonly IConnectionFactory _factory;
|
|
private readonly StoreContext _storeContext;
|
|
private readonly IHttpClientFactory _httpClientFactory;
|
|
public RabbitMqConsumerService(IConfiguration configuration, StoreContext storeContext, IHttpClientFactory httpClientFactory)
|
|
{
|
|
_storeContext = storeContext ?? throw new ArgumentNullException(nameof(storeContext));
|
|
_httpClientFactory = httpClientFactory ?? throw new ArgumentNullException(nameof(httpClientFactory));
|
|
user = configuration["RabbitMq:User"] ?? throw new ArgumentNullException("RabbitMq:User is not configured");
|
|
password = configuration["RabbitMq:Password"] ?? throw new ArgumentNullException("RabbitMq:Password is not configured");
|
|
host = configuration["RabbitMq:Host"] ?? throw new ArgumentNullException("RabbitMq:Host is not configured");
|
|
port = int.Parse(configuration["RabbitMq:Port"] ?? throw new ArgumentNullException("RabbitMq:Port is not configured")); ;
|
|
_factory = new ConnectionFactory
|
|
{
|
|
UserName = user,
|
|
Password = password,
|
|
HostName = host,
|
|
Port = port
|
|
};
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
stoppingToken.ThrowIfCancellationRequested();
|
|
|
|
var connection = await _factory.CreateConnectionAsync();
|
|
var channel = await connection.CreateChannelAsync();
|
|
|
|
await channel.QueueDeclareAsync(queue: "inventory_queue",
|
|
durable: false,
|
|
exclusive: false,
|
|
autoDelete: false,
|
|
arguments: null);
|
|
var consumer = new AsyncEventingBasicConsumer(channel);
|
|
consumer.ReceivedAsync += async (model, ea) =>
|
|
{
|
|
var body = ea.Body.ToArray();
|
|
var message = Encoding.UTF8.GetString(body);
|
|
Console.WriteLine($"Received message: {message}");
|
|
var purchaseMessage = JsonSerializer.Deserialize<PurchaseMessage>(message);
|
|
if (purchaseMessage == null)
|
|
{
|
|
Console.WriteLine("Failed to deserialize message.");
|
|
return;
|
|
}
|
|
var order = await GetOrderFromMessage(purchaseMessage);
|
|
foreach (var item in order.OrderItems)
|
|
{
|
|
var product = await _storeContext.Products.FindAsync(item.ProductId);
|
|
if (product == null)
|
|
{
|
|
Console.WriteLine($"Product with ID {item.ProductId} not found.");
|
|
continue;
|
|
}
|
|
if (product.Quantity < item.Quantity)
|
|
{
|
|
Console.WriteLine($"Insufficient stock for product {product.Name}. Available: {product.Quantity}, Requested: {item.Quantity}");
|
|
continue;
|
|
}
|
|
product.Quantity -= item.Quantity;
|
|
}
|
|
await _storeContext.SaveChangesAsync();
|
|
Console.WriteLine($"Processed order {order.Id} for customer {order.CustomerId} with {order.OrderItems.Count} items.");
|
|
await channel.BasicAckAsync(ea.DeliveryTag, false);
|
|
};
|
|
await channel.BasicConsumeAsync(queue: "inventory_queue",
|
|
autoAck: false,
|
|
consumer: consumer);
|
|
Console.WriteLine("RabbitMQ consumer started. Waiting for messages...");
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
|
|
}
|
|
|
|
}
|
|
|
|
private async Task<Order> GetOrderFromMessage(PurchaseMessage purchaseMessage)
|
|
{
|
|
string orderServiceUrl = $"http://orderservice:8080/api/order/{purchaseMessage.OrderId}";
|
|
Order order;
|
|
var client = _httpClientFactory.CreateClient("InventoryServiceClient");
|
|
var response = await client.GetAsync(orderServiceUrl);
|
|
|
|
if (!response.IsSuccessStatusCode)
|
|
{
|
|
throw new HttpRequestException($"Order with ID {purchaseMessage.OrderId} not found.");
|
|
}
|
|
|
|
var contentStream = await response.Content.ReadAsStreamAsync();
|
|
// Use JsonSerializer to convert the JSON response into your Order object.
|
|
order = await JsonSerializer.DeserializeAsync<Order>(contentStream, new JsonSerializerOptions { PropertyNameCaseInsensitive = true })
|
|
?? throw new InvalidOperationException("Order could not be parsed or is empty.");
|
|
|
|
return order;
|
|
}
|
|
|
|
}
|