PriceQueueTriggerFA.cs

using System.Net;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Microsoft.Extensions.Configuration;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker.Extensions.ServiceBus;
using PriceMS.DTO;
using PriceMS.Transformer;
using System.Xml.Linq;
using System.Xml;
using System.Text;

namespace PriceMS.Function;

public class PriceQueueTriggerFA
{
    private readonly ILogger<PriceQueueTriggerFA> _logger;
    private readonly IPriceTransFormerService _priceTransFormerService;
    private readonly ServiceBusReceiver _receiver;
    private readonly ServiceBusSender _sender;

    public PriceQueueTriggerFA(ILogger<PriceQueueTriggerFA> logger, IPriceTransFormerService priceTransFormerService, ServiceBusClient client)
    {
        _logger = logger;
        _priceTransFormerService = priceTransFormerService;
        _receiver = client.CreateReceiver("sbq-pl-a-001-01");
        _sender = client.CreateSender("sbq-pl-a-001-02");

    }

    [Function("PriceQueueTriggerFA")]
    public async Task ConsumePriceMessage(
        [ServiceBusTrigger("sbq-pl-a-001-01", Connection ="ServiceBusConnectionString")]
        ServiceBusReceivedMessage message)
    {
        _logger.LogInformation("Received Message from Queue: {queueName}", "sbq-pl-a-001-01");

        if (message != null)
        {
            string messagePayload = message.Body.ToString();
            _logger.LogInformation("Message ID: {MessageId}", message.MessageId);
            _logger.LogInformation("Delivery Count: {DeliveryCount}", message.DeliveryCount);
            //_logger.LogInformation("Message: {message}", messagePayload);
        }
        else
        {
            _logger.LogError("Message Body is null");
        }


        List<InputDTO> priceDataList = JsonConvert.DeserializeObject<List<InputDTO>>(message!.Body.ToString()) ?? throw new NullReferenceException("Message Body can't be null.");

        //transformation
        List<XDocument> transformedPayload = _priceTransFormerService.TransformPricePayload(priceDataList);

        //send the transformed msg to service bus

        try
        {
            foreach (XDocument xmlPayload in transformedPayload)
            {
                //_logger.LogInformation(xmlPayload.ToString());
                using var ms = new MemoryStream();
                using (var writer = XmlWriter.Create(ms, new XmlWriterSettings
                {
                    Encoding = new UTF8Encoding(false),
                    Indent = true
                }))
                {
                    xmlPayload.Save(writer);
                }

                ServiceBusMessage serviceBusMessage = new ServiceBusMessage(BinaryData.FromBytes(ms.ToArray()))
                {
                    CorrelationId = Guid.NewGuid().ToString(),
                    ContentType = "application/xml"
                };
                serviceBusMessage.ApplicationProperties["recordType"] = "price";
                serviceBusMessage.ApplicationProperties["source"] = "PriceQueueTriggerFA";
                await _sender.SendMessageAsync(serviceBusMessage);
            }
        }
        catch (Exception exception)
        {
            _logger.LogError(exception.Message);
        }

        await Task.CompletedTask;
    }
}