Add DynamoDB 4 Sample (Transactions)

samples/aws/dynamodb-transactions/DynamoDB_3/Client/Program.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -7,16 +8,52 @@
 Console.Title = "Client";
 
 var builder = Host.CreateApplicationBuilder(args);
-builder.Services.AddHostedService<InputLoopService>();
 
 var endpointConfiguration = new EndpointConfiguration("Samples.DynamoDB.Transactions.Client");
 endpointConfiguration.UseSerialization<SystemJsonSerializer>();
 endpointConfiguration.UseTransport<LearningTransport>();
 
-
 Console.WriteLine("Press any key, the application is starting");
 Console.ReadKey();
 Console.WriteLine("Starting...");
 
 builder.UseNServiceBus(endpointConfiguration);
-await builder.Build().RunAsync();
+
+var host = builder.Build();
+
+// Get the required services
+var messageSession = host.Services.GetRequiredService<IMessageSession>();
+// Register a cancellation token to gracefully handle application shutdown
+var ct = host.Services.GetRequiredService<IHostApplicationLifetime>().ApplicationStopping;
+
+Console.WriteLine("Press 'S' to send a StartOrder message to the server endpoint.");
+Console.WriteLine("Press Ctrl+C to shut down.");
+
+// Wait for user input to publish messages
+while (!ct.IsCancellationRequested)
+{
+    if (!Console.KeyAvailable)
+    {
+        // If no key is pressed, wait for a short time before checking again
+        await Task.Delay(100, CancellationToken.None);
+        continue;
+    }
+
+    var key = Console.ReadKey();
+    Console.WriteLine();
+
+    if (key.Key == ConsoleKey.S)
+    {
+        var orderId = Guid.NewGuid();
+        var startOrder = new StartOrder
+        {
+            OrderId = orderId
+        };
+
+        await messageSession.Send("Samples.DynamoDB.Transactions.Server", startOrder);
+        Console.WriteLine($"StartOrder Message sent to Server with OrderId {orderId}");
+    }
+}
+
+// Wait for the host to stop gracefully
+await host.StopAsync();

samples/aws/dynamodb-transactions/DynamoDB_3/Server/Program.cs

@@ -4,7 +4,6 @@
 using Microsoft.Extensions.Hosting;
 using NServiceBus;
 
-
 Console.Title = "Server";
 
 var builder = Host.CreateApplicationBuilder(args);
@@ -39,8 +38,6 @@
 });
 endpointConfiguration.EnableInstallers();
 
-
-
 Console.WriteLine("Press any key, the application is starting");
 Console.ReadKey();
 Console.WriteLine("Starting...");

samples/aws/dynamodb-transactions/DynamoDB_4/Client/Client.csproj

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFramework>net10.0</TargetFramework>
+    <OutputType>Exe</OutputType>
+    <LangVersion>preview</LangVersion>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\SharedMessages\SharedMessages.csproj" />
+  </ItemGroup>
+  <ItemGroup>
+    <PackageReference Include="NServiceBus" Version="10.0.0-alpha.3" />
+    <PackageReference Include="NServiceBus.Extensions.Hosting" Version="4.0.0-alpha.2" />
+  </ItemGroup>
+</Project>

samples/aws/dynamodb-transactions/DynamoDB_4/Client/OrderCompletedHandler.cs

@@ -0,0 +1,11 @@
+using Microsoft.Extensions.Logging;
+
+public class OrderCompletedHandler(ILogger<OrderCompletedHandler> logger) :
+    IHandleMessages<OrderCompleted>
+{
+    public Task Handle(OrderCompleted message, IMessageHandlerContext context)
+    {
+        logger.LogInformation("Received OrderCompleted for OrderId {OrderId}", message.OrderId);
+        return Task.CompletedTask;
+    }
+}

samples/aws/dynamodb-transactions/DynamoDB_4/Client/Program.cs

@@ -0,0 +1,51 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+
+Console.Title = "Client";
+
+var builder = Host.CreateApplicationBuilder(args);
+
+var endpointConfiguration = new EndpointConfiguration("Samples.DynamoDB.Transactions.Client");
+endpointConfiguration.UseSerialization<SystemJsonSerializer>();
+endpointConfiguration.UseTransport<LearningTransport>();
+
+builder.UseNServiceBus(endpointConfiguration);
+
+var host = builder.Build();
+
+// Get the required services
+var messageSession = host.Services.GetRequiredService<IMessageSession>();
+// Register a cancellation token to gracefully handle application shutdown
+var ct = host.Services.GetRequiredService<IHostApplicationLifetime>().ApplicationStopping;
+
+Console.WriteLine("Press 'S' to send a StartOrder message to the server endpoint.");
+Console.WriteLine("Press Ctrl+C to shut down.");
+
+// Wait for user input to publish messages
+while (!ct.IsCancellationRequested)
+{
+    if (!Console.KeyAvailable)
+    {
+        // If no key is pressed, wait for a short time before checking again
+        await Task.Delay(100, CancellationToken.None);
+        continue;
+    }
+
+    var key = Console.ReadKey();
+    Console.WriteLine();
+
+    if (key.Key == ConsoleKey.S)
+    {
+        var orderId = Guid.NewGuid();
+        var startOrder = new StartOrder
+        {
+            OrderId = orderId
+        };
+
+        await messageSession.Send("Samples.DynamoDB.Transactions.Server", startOrder);
+        Console.WriteLine($"StartOrder Message sent to Server with OrderId {orderId}");
+    }
+}
+
+// Wait for the host to stop gracefully
+await host.StopAsync();

samples/aws/dynamodb-transactions/DynamoDB_4/DynamoDB.StartupProjects.txt

@@ -0,0 +1,2 @@
+Client\Client.csproj
+Server\Server.csproj

samples/aws/dynamodb-transactions/DynamoDB_4/DynamoDB.sln

@@ -0,0 +1,34 @@
+Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 16
+VisualStudioVersion = 16.0.29728.190
+MinimumVisualStudioVersion = 15.0.26730.12
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Server", "Server\Server.csproj", "{48F718EE-6C45-41BA-80EC-81BF34D4A623}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SharedMessages", "SharedMessages\SharedMessages.csproj", "{DD438DB2-9C03-4BC0-BA52-BB7A35098458}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Client", "Client\Client.csproj", "{2FE71442-7F81-428E-B945-D564850D6564}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{4043293B-171D-4E7D-8E47-A9957327AD20}"
+	ProjectSection(SolutionItems) = preProject
+		docker-compose.yml = docker-compose.yml
+	EndProjectSection
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{48F718EE-6C45-41BA-80EC-81BF34D4A623}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{48F718EE-6C45-41BA-80EC-81BF34D4A623}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{DD438DB2-9C03-4BC0-BA52-BB7A35098458}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{DD438DB2-9C03-4BC0-BA52-BB7A35098458}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{2FE71442-7F81-428E-B945-D564850D6564}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{2FE71442-7F81-428E-B945-D564850D6564}.Debug|Any CPU.Build.0 = Debug|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {BB6580A8-3A40-4373-B226-A86402536658}
+	EndGlobalSection
+EndGlobal

samples/aws/dynamodb-transactions/DynamoDB_4/Server/CompleteOrder.cs

@@ -0,0 +1,5 @@
+public class CompleteOrder
+{
+    public Guid OrderId { get; set; }
+    public string OrderDescription { get; set; }
+}

samples/aws/dynamodb-transactions/DynamoDB_4/Server/OrderSaga.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using NServiceBus;
+
+#region thesaga
+
+public class OrderSaga(ILogger<OrderSaga> logger):
+    Saga<OrderSagaData>,
+    IAmStartedByMessages<StartOrder>,
+    IHandleMessages<OrderShipped>,
+    IHandleTimeouts<CompleteOrder>
+{
+    protected override void ConfigureHowToFindSaga(SagaPropertyMapper<OrderSagaData> mapper)
+    {
+        mapper.MapSaga(saga => saga.OrderId)
+            .ToMessage<StartOrder>(msg => msg.OrderId)
+            .ToMessage<OrderShipped>(msg => msg.OrderId);
+    }
+
+    public async Task Handle(StartOrder message, IMessageHandlerContext context)
+    {
+        var orderDescription = $"The saga for order {message.OrderId}";
+        Data.OrderDescription = orderDescription;
+
+        logger.LogInformation("Received StartOrder message {OrderId}. Starting Saga", Data.OrderId);
+
+        var shipOrder = new ShipOrder
+        {
+            OrderId = message.OrderId
+        };
+
+        logger.LogInformation("Order will complete in 5 seconds");
+        var timeoutData = new CompleteOrder
+        {
+            OrderDescription = orderDescription,
+            OrderId = Data.OrderId,
+        };
+
+        await Task.WhenAll(
+            context.SendLocal(shipOrder),
+            RequestTimeout(context, TimeSpan.FromSeconds(5), timeoutData)
+        );
+    }
+
+    public Task Handle(OrderShipped message, IMessageHandlerContext context)
+    {
+        logger.LogInformation("Order with OrderId {OrderId} shipped on {ShippingDate}", Data.OrderId, message.ShippingDate);
+        return Task.CompletedTask;
+    }
+
+    public async Task Timeout(CompleteOrder state, IMessageHandlerContext context)
+    {
+        logger.LogInformation("Saga with OrderId {OrderId} completed", Data.OrderId);
+        MarkAsComplete();
+        var orderCompleted = new OrderCompleted
+        {
+            OrderId = Data.OrderId
+        };
+        await context.Publish(orderCompleted);
+    }
+}
+
+#endregion

samples/aws/dynamodb-transactions/DynamoDB_4/Server/OrderSagaData.cs

@@ -0,0 +1,12 @@
+using System;
+using NServiceBus;
+
+#region sagadata
+
+public class OrderSagaData :
+    ContainSagaData
+{
+    public Guid OrderId { get; set; }
+    public string OrderDescription { get; set; }
+}
+#endregion

samples/aws/dynamodb-transactions/DynamoDB_4/Server/OrderShipped.cs

@@ -0,0 +1,5 @@
+public class OrderShipped : IEvent
+{
+    public Guid OrderId { get; set; }
+    public DateTimeOffset ShippingDate { get; set; }
+}

samples/aws/dynamodb-transactions/DynamoDB_4/Server/OrderShippingInformation.cs

@@ -0,0 +1,8 @@
+public class OrderShippingInformation
+{
+    public Guid Id { get; set; }
+
+    public Guid OrderId { get; set; }
+
+    public DateTimeOffset ShippedAt { get; set; }
+}

samples/aws/dynamodb-transactions/DynamoDB_4/Server/OrderShippingInformationExtensions.cs

@@ -0,0 +1,14 @@
+using Amazon.DynamoDBv2.Model;
+using NServiceBus.Persistence.DynamoDB;
+
+public static class OrderShippingInformationExtensions
+{
+    public static Dictionary<string, AttributeValue> ToMap(this OrderShippingInformation orderShippingInformation)
+    {
+        var attributeMap = Mapper.ToMap(orderShippingInformation);
+        var orderShippingInformationId = orderShippingInformation.Id.ToString();
+        attributeMap["PK"] = new AttributeValue(orderShippingInformationId);
+        attributeMap["SK"] = new AttributeValue(orderShippingInformationId);
+        return attributeMap;
+    }
+}

samples/aws/dynamodb-transactions/DynamoDB_4/Server/Program.cs

@@ -0,0 +1,40 @@
+using Amazon.DynamoDBv2;
+using Amazon.Runtime;
+using Microsoft.Extensions.Hosting;
+
+Console.Title = "Server";
+
+var builder = Host.CreateApplicationBuilder(args);
+
+#region DynamoDBConfig
+
+var amazonDynamoDbClient = new AmazonDynamoDBClient(
+    new BasicAWSCredentials("localdb", "localdb"),
+    new AmazonDynamoDBConfig
+    {
+        ServiceURL = "http://localhost:8000"
+    });
+
+
+var endpointConfiguration = new EndpointConfiguration("Samples.DynamoDB.Transactions.Server");
+endpointConfiguration.UseSerialization<SystemJsonSerializer>();
+
+var persistence = endpointConfiguration.UsePersistence<DynamoPersistence>();
+persistence.DynamoClient(amazonDynamoDbClient);
+persistence.UseSharedTable(new TableConfiguration
+{
+    TableName = "Samples.DynamoDB.Transactions"
+});
+
+endpointConfiguration.EnableOutbox();
+
+#endregion
+
+endpointConfiguration.UseTransport(new LearningTransport
+{
+    TransportTransactionMode = TransportTransactionMode.ReceiveOnly
+});
+endpointConfiguration.EnableInstallers();
+
+builder.UseNServiceBus(endpointConfiguration);
+await builder.Build().RunAsync();