Notes

using Azure.Storage.Files.DataLake;
using Azure.Storage.Files.DataLake.Models;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using System;
using System.IO;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;

public static class ServiceBusTrigger
{
[FunctionName("Topic1TriggerFunction")]
public static async Task RunTopic1(
[ServiceBusTrigger("%ServiceBusTopics:Topic1:Name%", "%SubscriptionName%", Connection = "ServiceBusConnectionString")]
Message message,
string topicName,
string subscriptionName,
string serviceBusConnectionString,
string adlsAccountName,
string adlsFileSystemName,
string adlsConnectionString,
string environment,
ILogger log,
ExecutionContext context)
{
await ProcessMessage(message, topicName, subscriptionName, serviceBusConnectionString, adlsAccountName, adlsFileSystemName, adlsConnectionString, environment, log, context);
}

[FunctionName("Topic2TriggerFunction")]
public static async Task RunTopic2(
[ServiceBusTrigger("%ServiceBusTopics:Topic2:Name%", "%SubscriptionName%", Connection = "ServiceBusConnectionString")]
Message message,
string topicName,
string subscriptionName,
string serviceBusConnectionString,
string adlsAccountName,
string adlsFileSystemName,
string adlsConnectionString,
string environment,
ILogger log,
ExecutionContext context)
{
await ProcessMessage(message, topicName, subscriptionName, serviceBusConnectionString, adlsAccountName, adlsFileSystemName, adlsConnectionString, environment, log, context);
}

[FunctionName("Topic3TriggerFunction")]
public static async Task RunTopic3(
[ServiceBusTrigger("%ServiceBusTopics:Topic3:Name%", "%SubscriptionName%", Connection = "ServiceBusConnectionString")]
Message message,
string topicName,
string subscriptionName,
string serviceBusConnectionString,
string adlsAccountName,
string adlsFileSystemName,
string adlsConnectionString,
string environment,
ILogger log,
ExecutionContext context)
{
await ProcessMessage(message, topicName, subscriptionName, serviceBusConnectionString, adlsAccountName, adlsFileSystemName, adlsConnectionString, environment, log, context);
}

private static async Task ProcessMessage(Message message, string topicName, string subscriptionName, string serviceBusConnectionString, string adlsAccountName, string adlsFileSystemName, string adlsConnectionString, string environment, ILogger log, ExecutionContext context)
{
// Récupérer l'identifiant du message
string messageId = message.MessageId;

// Récupérer la date d'envoi du message
DateTime enqueuedTimeUtc = message.SystemProperties.EnqueuedTimeUtc;

// Récupérer le contenu du message
string messageBody = Encoding.UTF8.GetString(message.Body);

// Convertir le contenu du message en JSON
string messageJson = JsonSerializer.Serialize(messageBody);

// Enregistrer le message au format JSON dans ADLS Gen2
await SaveMessageToADLS(messageId, enqueuedTimeUtc, messageJson, adlsAccountName, adlsFileSystemName, topicName, environment, adlsConnectionString, log, context);

// Traiter le message reçu
log.LogInformation($"Message reçu depuis le topic '{topicName}' et la souscription '{subscriptionName}': {messageBody}");
log.LogInformation($"Identifiant du message : {messageId}");
log.LogInformation($"Date d'envoi du message : {enqueuedTimeUtc}");
}

private static async Task SaveMessageToADLS(string messageId, DateTime enqueuedTimeUtc, string messageJson, string accountName, string fileSystemName, string topicName, string environment, string storageConnectionString, ILogger log, ExecutionContext context)
{
// Configuration de l'accès à ADLS Gen2
string year = enqueuedTimeUtc.Year.ToString();
string month = enqueuedTimeUtc.Month.ToString("D2");
string day = enqueuedTimeUtc.Day.ToString("D2");

string directoryName = $"{environment}/silver/{topicName}/{year}/{month}/{day}"; // Utiliser la structure : environment/silver/topicName/année/mois/jour
string fileName = $"{messageId}_{enqueuedTimeUtc:yyyyMMddHHmmss}.json"; // Nom du fichier basé sur l'identifiant du message et la date d'envoi

// Créer un client pour accéder à l'ADLS Gen2
DataLakeFileSystemClient fileSystemClient = new DataLakeFileSystemClient(storageConnectionString, accountName, fileSystemName);

// Créer un répertoire s'il n'existe pas déjà
await fileSystemClient.CreateDirectoryAsync(directoryName);

// Créer ou écrire dans le fichier avec le contenu du message JSON
DataLakeFileClient fileClient = fileSystemClient.GetFileClient($"{directoryName}/{fileName}");
using (MemoryStream stream = new MemoryStream(Encoding.UTF8.GetBytes(messageJson)))
{
await fileClient.UploadAsync(stream, true);
}
}
}