Hybride Integration mit Azure Service Bus

<span class="hljs-keyword">using</span> Microsoft.Azure.Functions.Worker;<br/><span class="hljs-keyword">using</span> Microsoft.Extensions.Logging;<br/><span class="hljs-keyword">using</span> Microsoft.AspNetCore.Http;<br/><span class="hljs-keyword">using</span> Microsoft.AspNetCore.Mvc;<br/><span class="hljs-keyword">using</span> System.IO.Compression;<br/><span class="hljs-keyword">using</span> Azure.Messaging.ServiceBus;<br/><span class="hljs-keyword">namespace</span> <span class="hljs-title">FunctionApp</span><br/>{<br/>  <span class="hljs-keyword">public</span> <span class="hljs-keyword">class</span> <span class="hljs-title">EmpfangeZipBestelldatei</span><br/>  {<br/>    <span class="hljs-keyword">private</span> <span class="hljs-keyword">readonly</span> ILogger<EmpfangeZipBestelldatei> <br/>      _logger;<br/>    <span class="hljs-keyword">private</span> <span class="hljs-keyword">readonly</span> ServiceBusClient <br/>      _serviceBusClient;<br/>    <span class="hljs-keyword">private</span> <span class="hljs-keyword">readonly</span> ServiceBusSender <br/>      _serviceBusSender;<br/>    <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-title">EmpfangeZipBestelldatei</span>(</span><br/><span class="hljs-function"><span class="hljs-params">        ILogger<EmpfangeZipBestelldatei> logger</span>)</span><br/><span class="hljs-function">    </span>{<br/>      logger = logger;<br/>      serviceBusClient = <span class="hljs-keyword">new</span> ServiceBusClient(<br/>        <span class="hljs-string">"<CONNECTIONSTRING>"</span>);<br/>      serviceBusSender = _serviceBusClient.CreateSender(<br/>        <span class="hljs-string">"<QUEUENAME>"</span>);<br/>    }<br/>    [Function(<span class="hljs-string">"EmpfangeZipBestelldatei"</span>)]<br/>    <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">async</span> Task<IActionResult> <span class="hljs-title">Run</span>(<span class="hljs-params">[HttpTrigger(</span></span><br/><span class="hljs-function"><span class="hljs-params">        AuthorizationLevel.Function, <span class="hljs-string">"post"</span></span>)] </span><br/><span class="hljs-function">        HttpRequest req)</span><br/><span class="hljs-function">    </span>{<br/>      logger.LogInformation(<span class="hljs-string">"Received a request to </span><br/><span class="hljs-string">        process and send notifications."</span>);<br/>      <span class="hljs-keyword">if</span> (!req.HasFormContentType || <br/>          !req.Form.Files.Any())<br/>      {<br/>        <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> BadRequestObjectResult(<br/>          <span class="hljs-string">"Please upload a ZIP file."</span>);<br/>      }<br/>      <span class="hljs-keyword">var</span> file = req.Form.Files.First();<br/>      <span class="hljs-keyword">if</span> (file.Length == <span class="hljs-number">0</span> || !file.FileName.EndsWith(<br/>          <span class="hljs-string">".zip"</span>, StringComparison.OrdinalIgnoreCase))<br/>      {<br/>        <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> BadRequestObjectResult(<br/>          <span class="hljs-string">"The uploaded file is not a valid ZIP file."</span>);<br/>      }<br/>      <span class="hljs-keyword">try</span><br/>      {<br/>        <span class="hljs-keyword">using</span> (<span class="hljs-keyword">var</span> zipStream = file.OpenReadStream())<br/>        {<br/>          <span class="hljs-keyword">using</span> (<span class="hljs-keyword">var</span> archive = <span class="hljs-keyword">new</span> ZipArchive(<br/>              zipStream, ZipArchiveMode.Read))<br/>          {<br/>            <span class="hljs-keyword">foreach</span> (<span class="hljs-keyword">var</span> entry <span class="hljs-keyword">in</span> archive.Entries)<br/>            {<br/>              <span class="hljs-keyword">using</span> (<span class="hljs-keyword">var</span> entryStream = entry.Open())<br/>              {<br/>                <span class="hljs-keyword">string</span> fileContent = <span class="hljs-keyword">new</span> StreamReader(<br/>                  entryStream).ReadToEnd();<br/>                logger.LogInformation(<span class="hljs-string">$"<span class="hljs-subst">{fileContent}</span>"</span>);<br/>                <span class="hljs-keyword">await</span> <br/>                  _serviceBusSender.SendMessageAsync(<br/>                  <span class="hljs-keyword">new</span> ServiceBusMessage(fileContent));<br/>              }<br/>            }<br/>          }<br/>        }<br/>        <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> OkObjectResult(<span class="hljs-string">"ZIP file processed </span><br/><span class="hljs-string">          and notifications sent successfully."</span>);<br/>      }<br/>      <span class="hljs-keyword">catch</span> (Exception ex)<br/>      {<br/>        logger.LogError(<br/>          <span class="hljs-string">$"An error occurred: <span class="hljs-subst">{ex.Message}</span>"</span>);<br/>        <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> StatusCodeResult(<br/>          StatusCodes.Status500InternalServerError);<br/>      }<br/>    }<br/>  }<br/>} 

_serviceBusClient = 
  new ServiceBusClient("<CONNECTIONSTRING>");
_serviceBusSender = 
  _serviceBusClient.CreateSender("<QUEUENAME>"); 

[Function("EmpfangeZipBestelldatei")]
public async Task<IActionResult> Run(
  [HttpTrigger(AuthorizationLevel.Function, 
  "post")] HttpRequest req){...} 

await _serviceBusSender
  .SendMessageAsync(
  new ServiceBusMessage(
  fileContent)); 

using Azure.Messaging.ServiceBus;<br/>namespace ServiceBusWorker;<br/>public class Worker : BackgroundService<br/>{<br/>  private readonly ILogger<Worker> _logger;<br/>  private readonly ServiceBusClient _serviceBusClient;<br/>  private readonly ServiceBusProcessor <br/>    _serviceBusProcessor;<br/>  public Worker(ILogger<Worker> logger)<br/>  {<br/>    logger = logger;<br/>    serviceBusClient = new ServiceBusClient(<br/>      "<CONNECTIONSTRING>");<br/>    serviceBusProcessor = _serviceBusClient<br/>      .CreateProcessor("<QUEUENAME>");<br/>    serviceBusProcessor.ProcessMessageAsync += <br/>      _serviceBusProcessor_ProcessMessageAsync;<br/>    serviceBusProcessor.ProcessErrorAsync += <br/>      _serviceBusProcessor_ProcessErrorAsync;<br/>  }<br/>  private async Task <br/>      _serviceBusProcessor_ProcessMessageAsync(<br/>      ProcessMessageEventArgs arg)<br/>  {<br/>    var bestellung = arg.Message.Body.ToString();<br/>    // Hier findet nun die Verarbeitung der Bestellung <br/>    // statt. Validierung etc., Transfer ins ERP<br/>    // Wenn keine Fehler aufgetreten sind, kann die<br/>    // Nachricht im Service Bus gelöscht werden.<br/>    logger.LogInformation(<br/>      $"On-Premises-Worker: Bestellung empfangen. <br/>      Inhalt der Datei: {bestellung}");<br/>    await arg.CompleteMessageAsync(arg.Message);<br/>  }<br/>  private Task _serviceBusProcessor_ProcessErrorAsync(<br/>      ProcessErrorEventArgs arg)<br/>  {<br/>    logger.LogCritical("Fehler bei der Verarbeitung <br/>      der Bestelldateien.");<br/>    return Task.CompletedTask;<br/>  }<br/>  protected override async Task ExecuteAsync(<br/>      CancellationToken stoppingToken)<br/>  {<br/>    await _serviceBusProcessor.StartProcessingAsync(<br/>      stoppingToken);<br/>    while (!stoppingToken.IsCancellationRequested)<br/>    {<br/>      if (logger.IsEnabled(LogLevel.Information))<br/>      {<br/>        logger.LogInformation("Worker running at: <br/>          {time}", DateTimeOffset.Now);<br/>      }<br/>      await Task.Delay(1000, stoppingToken);<br/>    }<br/>  }<br/>} 

serviceBusClient = 
  new ServiceBusClient("<CONNECTIONSTRING>");
serviceBusProcessor = _serviceBusClient.CreateProcessor(
  "<QUEUENAME>"); 

serviceBusProcessor.ProcessMessageAsync += 
  _serviceBusProcessor_ProcessMessageAsync;
serviceBusProcessor.ProcessErrorAsync += 
  _serviceBusProcessor_ProcessErrorAsync; 

await arg.CompleteMessageAsync(arg.Message); 

await _serviceBusProcessor.StartProcessingAsync(
  stoppingToken);