Add samples for Azure Queue Storage

2026-06-20 12:23:14 +00:00 · 2026-02-01 08:47:38 +00:00
parent 38ee857784
commit 1755b74692
106 changed files with 64610 additions and 16 deletions
@@ -1,29 +1,90 @@
 using Microsoft.AspNetCore.SignalR;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using System.Text;
+using System.Text.Json;

 namespace OnlineShop.ApiService;

-public class LocationUpdater(
-    IHubContext<LocationHub> locationHub) :
-    BackgroundService
+public sealed class LocationUpdater(
+    IHubContext<LocationHub> locationHub,
+    IConnection rabbitConnection) : BackgroundService
 {
-    protected async override Task
-    ExecuteAsync(
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await using var channel =
+            await rabbitConnection.CreateChannelAsync(cancellationToken: stoppingToken);
+
+        const string queueName = "orders.created";
+
+        await channel.QueueDeclareAsync(
+            queue: queueName,
+            durable: true,
+            exclusive: false,
+            autoDelete: false,
+            arguments: null,
+            cancellationToken: stoppingToken);
+
+        await channel.BasicQosAsync(
+            prefetchSize: 0,
+            prefetchCount: 1,
+            global: false,
+            cancellationToken: stoppingToken);
+
+        var consumer = new AsyncEventingBasicConsumer(channel);
+
+        consumer.ReceivedAsync += async (sender, args) =>
+        {
+            try
+            {
+                var json = Encoding.UTF8.GetString(args.Body.ToArray());
+                var message = JsonSerializer.Deserialize<OrderCreatedMessage>(json);
+
+                if (message is null)
+                {
+                    await channel.BasicNackAsync(
+                        args.DeliveryTag, false, false, stoppingToken);
+                    return;
+                }
+
+                await SetInitialDeliveryLocation(
+                    message.OrderId,
+                    stoppingToken);
+
+                await channel.BasicAckAsync(
+                    args.DeliveryTag,
+                    multiple: false,
+                    cancellationToken: stoppingToken);
+            }
+            catch
+            {
+                await channel.BasicNackAsync(
+                    args.DeliveryTag,
+                    false,
+                    requeue: true,
+                    cancellationToken: stoppingToken);
+            }
+        };
+
+        await channel.BasicConsumeAsync(
+            queue: queueName,
+            autoAck: false,
+            consumer: consumer,
+            cancellationToken: stoppingToken);
+
+        await Task.Delay(Timeout.Infinite, stoppingToken);
+    }
+
+    private async Task SetInitialDeliveryLocation(
+        int orderId,
        CancellationToken cancellationToken)
    {
-        await Task.Delay(5000, cancellationToken);
-        await UpdateLocation(
-            51.5074, -0.1276, cancellationToken);
-
-        await Task.Delay(5000, cancellationToken);
-        await UpdateLocation(
-            51.5074, -0.13, cancellationToken);
-
-        await Task.Delay(5000, cancellationToken);
-        await UpdateLocation(
-            51.508, -0.14, cancellationToken);
+        await Task.Delay(3000, cancellationToken);
+        await UpdateLocation(orderId, 51.5074, -0.1276, cancellationToken);
    }

    private async Task UpdateLocation(
+        int orderId,
        double latitude,
        double longitude,
        CancellationToken cancellationToken)
@@ -35,4 +96,5 @@ public class LocationUpdater(
            cancellationToken);
    }

+    private sealed record OrderCreatedMessage(int OrderId);
 }