repro for a suspected issue with request/response messaging

src/Samples/OrderManagement/Messages/ApproveOrder.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record ApproveOrder(string OrderId, string CustomerId);

src/Samples/OrderManagement/Messages/CreditLimitExceeded.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record CreditLimitExceeded(string OrderId, string CustomerId);

src/Samples/OrderManagement/Messages/CreditReserved.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record CreditReserved(string OrderId, string CustomerId);

src/Samples/OrderManagement/Messages/Messages.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\Wolverine\Wolverine.csproj"/>
+    </ItemGroup>
+</Project>

src/Samples/OrderManagement/Messages/OrderApproved.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record OrderApproved(string OrderId);

src/Samples/OrderManagement/Messages/OrderCompleted.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record OrderCompleted(string OrderId);

src/Samples/OrderManagement/Messages/OrderCreated.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record OrderCreated(string OrderId, string CustomerId, string CustomerName);

src/Samples/OrderManagement/Messages/OrderPlaced.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record OrderPlaced(string OrderId, string CustomerId, decimal Amount);

src/Samples/OrderManagement/Messages/OrderRejected.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record OrderRejected(string OrderId);

src/Samples/OrderManagement/Messages/PlaceOrder.cs

@@ -0,0 +1,7 @@
+namespace Messages;
+
+public record PlaceOrder(
+    string OrderId,
+    string CustomerId,
+    decimal Amount
+);

src/Samples/OrderManagement/Messages/RejectOrder.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record RejectOrder(string OrderId);

src/Samples/OrderManagement/Messages/ReservceCredit.cs

@@ -0,0 +1,3 @@
+namespace Messages;
+
+public record ReserveCredit(string OrderId, string CustomerId, decimal Amount);

src/Samples/OrderManagement/Orders/Orders.csproj

@@ -0,0 +1,22 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\Persistence\Wolverine.Marten\Wolverine.Marten.csproj"/>
+        <ProjectReference Include="..\..\..\Transports\RabbitMQ\Wolverine.RabbitMQ\Wolverine.RabbitMQ.csproj"/>
+        <ProjectReference Include="..\Messages\Messages.csproj"/>
+    </ItemGroup>
+
+    <ItemGroup>
+      <Compile Include="..\..\..\Servers.cs">
+        <Link>Servers.cs</Link>
+      </Compile>
+    </ItemGroup>
+
+</Project>

src/Samples/OrderManagement/Orders/PlaceOrderHandler.cs

@@ -0,0 +1,59 @@
+using Marten;
+using Marten.Events;
+using Marten.Events.Aggregation;
+using Messages;
+using Microsoft.Extensions.Logging;
+
+namespace Orders;
+
+public record PurchaseOrder(string Id, string Status = "Placed");
+
+public class PurchaseOrderProjection : SingleStreamProjection<PurchaseOrder>
+{
+    public static PurchaseOrder Create(
+        IEvent<OrderPlaced> @event
+    )
+    {
+        return new PurchaseOrder(@event.Data.OrderId);
+    }
+
+    public static PurchaseOrder Apply(
+        IEvent<OrderRejected> @event,
+        PurchaseOrder current
+    )
+    {
+        return current with { Status = "Rejected" };
+    }
+
+    public static PurchaseOrder Apply(
+        IEvent<OrderApproved> @event,
+        PurchaseOrder current
+    )
+    {
+        return current with { Status = "Approved" };
+    }
+}
+
+public class OrderPlacedHandler
+{
+    public async Task<object[]> Handle(
+        PlaceOrder message,
+        IDocumentSession session,
+        ILogger logger
+    )
+    {
+        var (orderId, customerId, amount) = message;
+        logger.LogInformation("Received Order: {OrderId}", message.OrderId);
+        var orderPlaced = new OrderPlaced(
+            orderId,
+            customerId,
+            amount
+        );
+        session.Events.Append(orderId, orderPlaced);
+        await session.SaveChangesAsync();
+
+        var order = await session.LoadAsync<PurchaseOrder>(orderId);
+
+        return [order, orderPlaced];
+    }
+}

src/Samples/OrderManagement/Orders/Program.cs

@@ -0,0 +1,85 @@
+// See https://aka.ms/new-console-template for more information
+
+using IntegrationTests;
+using JasperFx.Core;
+using Marten;
+using Marten.Events;
+using Marten.Events.Projections;
+using Messages;
+using Microsoft.Extensions.Hosting;
+using Npgsql;
+using Orders;
+using Wolverine;
+using Wolverine.RabbitMQ;
+
+var hostBuilder = Host.CreateDefaultBuilder(args);
+hostBuilder.ConfigureServices(
+    services =>
+    {
+        services.AddMarten(
+                options =>
+                {
+                    options.Connection(Servers.PostgresConnectionString);
+                    options.Events.StreamIdentity = StreamIdentity.AsString;
+                    options.Projections.Add<PurchaseOrderProjection>(ProjectionLifecycle.Inline);
+                    options.DisableNpgsqlLogging = true;
+                }
+            )
+            .UseLightweightSessions();
+        // .IntegrateWithWolverine();
+    }
+);
+hostBuilder
+    .UseWolverine(
+        options =>
+        {
+            options
+                .ListenToRabbitQueue(
+                    "new-orders",
+                    queue => queue.TimeToLive(15.Seconds())
+                );
+
+            options
+                .ListenToRabbitQueue(
+                    "complete-orders"
+                );
+
+            var rabbitMq = options.UseRabbitMq(
+                configure => { configure.ClientProvidedName = "Orders"; }
+            );
+
+            rabbitMq
+                .DeclareExchange(
+                    "new-orders",
+                    exchange =>
+                    {
+                        exchange.ExchangeType = ExchangeType.Direct;
+                        exchange.BindQueue("new-orders");
+                    }
+                )
+                .AutoProvision();
+
+
+            options
+                .UseRabbitMq()
+                .DeclareExchange(
+                    "placed-orders",
+                    exchange =>
+                    {
+                        exchange.ExchangeType = ExchangeType.Fanout;
+                        exchange.BindQueue("placed-orders-orders");
+                    }
+                )
+                .AutoProvision();
+
+            options.PublishMessage<OrderPlaced>()
+                .ToRabbitExchange("placed-orders");
+
+
+            options.ListenToRabbitQueue("placed-orders-orders");
+        }
+    );
+var host = hostBuilder
+    .Build();
+
+await host.RunAsync();

src/Samples/OrderManagement/Orders/RejectOrderHandler.cs

@@ -0,0 +1,20 @@
+using Marten;
+using Messages;
+using Microsoft.Extensions.Logging;
+
+namespace Orders;
+
+public class RejectOrderHandler
+{
+    public async Task Handle(
+        RejectOrder command,
+        IDocumentSession session,
+        ILogger logger
+    )
+    {
+        var rejected = new OrderRejected(command.OrderId);
+        session.Events.Append(command.OrderId, rejected);
+
+        await session.SaveChangesAsync();
+    }
+}

src/Samples/OrderManagement/Orders/Sagas/Order.cs

@@ -0,0 +1,76 @@
+using Messages;
+using Microsoft.Extensions.Logging;
+using Wolverine;
+
+namespace Orders.Sagas;
+
+public enum OrderStatus
+{
+    Pending = 0,
+    CreditReserved = 1,
+    CreditLimitExceeded = 2,
+    Approved = 3,
+    Rejected = 4
+}
+
+public class Order : Saga
+{
+    public string? Id { get; set; }
+    public OrderStatus OrderStatus { get; set; } = OrderStatus.Pending;
+
+    public object[] Start(
+        OrderPlaced orderPlaced,
+        ILogger logger
+    )
+    {
+        Id = orderPlaced.OrderId;
+        logger.LogInformation("Order {OrderId} placed", Id);
+        OrderStatus = OrderStatus.Pending;
+        return
+        [
+            new ReserveCredit(
+                orderPlaced.OrderId,
+                orderPlaced.CustomerId,
+                orderPlaced.Amount
+            )
+        ];
+    }
+
+    public object[] Handle(
+        CreditReserved creditReserved,
+        ILogger logger
+    )
+    {
+        OrderStatus = OrderStatus.CreditReserved;
+        logger.LogInformation("Credit reserver for Order {OrderId}", Id);
+        return [new ApproveOrder(creditReserved.OrderId, creditReserved.CustomerId)];
+    }
+
+    public void Handle(
+        OrderApproved orderApproved,
+        ILogger logger
+    )
+    {
+        OrderStatus = OrderStatus.Approved;
+        logger.LogInformation("Order {OrderId} approved", Id);
+    }
+
+    public object[] Handle(
+        CreditLimitExceeded creditLimitExceeded,
+        ILogger logger
+    )
+    {
+        OrderStatus = OrderStatus.CreditLimitExceeded;
+        return [new RejectOrder(creditLimitExceeded.OrderId)];
+    }
+
+    public void Handle(
+        OrderRejected orderRejected,
+        ILogger logger
+    )
+    {
+        OrderStatus = OrderStatus.Rejected;
+        logger.LogInformation("Order {OrderId} rejected", Id);
+        MarkCompleted();
+    }
+}

src/Samples/OrderManagement/RetailClient/Customer.cs

@@ -0,0 +1,8 @@
+namespace RetailClient;
+
+public class Customer
+{
+    public string Id { get; set; }
+    public string Email { get; set; }
+    public string Name { get; set; }
+}