Add Service Bus Sessions Example

src/Sample.Api/Startup.cs

@@ -60,6 +60,7 @@ public void ConfigureServices(IServiceCollection services)
                 {
                     cfg.Host(Configuration.GetConnectionString("AzureServiceBus"));
 
+                    cfg.Publish<OrderShippedBase>(configurator => configurator.Exclude = true);
                     cfg.Send<OrderShipped>(s => s.UseSessionIdFormatter(c => c.Message.OrderId.ToString("D")));
                 });
             });

src/Sample.Contracts/OrderShipped.cs

@@ -3,9 +3,13 @@ namespace Sample.Contracts
     using System;
 
 
-    public record OrderShipped
+    public record OrderShippedBase
     {
         public Guid OrderId { get; init; }
         public DateTimeOffset Timestamp { get; init; }
     }
+
+    public record OrderShipped : OrderShippedBase
+    {
+    }
 }

src/Sample.Worker/Consumers/OrderShippedConsumer.cs

@@ -0,0 +1,25 @@
+using System.Threading.Tasks;
+using MassTransit;
+using Microsoft.Extensions.Logging;
+using Sample.Contracts;
+
+namespace Sample.Worker.Consumers
+{
+    public class OrderShippedConsumer :
+        IConsumer<OrderShipped>
+    {
+        private readonly ILogger<OrderShippedConsumer> _logger;
+
+        public OrderShippedConsumer(ILogger<OrderShippedConsumer> logger)
+        {
+            _logger = logger;
+        }
+
+        public Task Consume(ConsumeContext<OrderShipped> context)
+        {
+            _logger.LogInformation("Order Shipped with OrderId: {OrderId}. SessionId {SessionId}.", 
+                context.Message.OrderId, context.SessionId());
+            return Task.CompletedTask;
+        }
+    }
+}

src/Sample.Worker/Program.cs

@@ -1,3 +1,5 @@
+using System;
+
 namespace Sample.Worker
 {
     using System.Threading.Tasks;
@@ -29,6 +31,10 @@ public static IHostBuilder CreateHostBuilder(string[] args)
 
                         x.AddConsumer<SubmitOrderConsumer>();
                         x.AddConsumer<OrderSubmittedConsumer>();
+                        x.AddConsumer<OrderShippedConsumer>(e =>
+                        {
+                            e.UseTimeout(c => c.Timeout = TimeSpan.FromSeconds(10));
+                        });
 
                         x.AddSagaStateMachine<OrderShipmentStateMachine, OrderShipmentState, OrderShipmentSagaDefinition>()
                             .MessageSessionRepository();
@@ -48,6 +54,19 @@ public static IHostBuilder CreateHostBuilder(string[] args)
                                 e.ConfigureConsumer<OrderSubmittedConsumer>(context);
                             });
 
+                            cfg.Publish<OrderShippedBase>(configurator => configurator.Exclude = true);
+                            cfg.SubscriptionEndpoint<OrderShipped>("order-shipped-consumer", e =>
+                            {
+                                e.ConfigureConsumeTopology = false;
+                                e.AutoStart = true;
+                                e.AutoDeleteOnIdle = TimeSpan.FromDays(30);
+                                e.MaxDeliveryCount = 2; 
+                                e.RequiresSession = true;
+                                e.MaxConcurrentCalls = 8; //like AZ Function default. 8 Concurrent consumers handling 1 ServiceBus Session
+                                e.PrefetchCount = 0;
+                                e.ConfigureConsumer<OrderShippedConsumer>(context);
+                            });
+
                             cfg.ConfigureEndpoints(context);
                         });
                     });