Reputation: 13
I am using Request Response Pattern for making sure that the proper work flow done sequentially, I have 3 Projects
Order.API
OrderPayment.API
OrderConfirmation.API
It is properly executing each Consumers, but after that, after last consumer's last code, it give me loading on swagger and Every time it is throwing Request Time out error. , it is not returning response back to Controller wither successfull work done or not, it is going for wait and then response is Request Timeout Error
Code
Consumer In OrderPayment.API
public class PaymentConsumer : IConsumer<PaymentProcessed>
{
public async Task Consume(ConsumeContext<PaymentProcessed> context)
{
Console.WriteLine($"Payment processed for OrderId: {context.Message.OrderId}");
await context.Send(new OrderConfirmed { OrderId = context.Message.OrderId });
}
}
and in Program.cs
builder.Services.AddMassTransit(x =>
{
x.AddConsumer<PaymentConsumer>();
x.UsingRabbitMq((context, cfg) =>
{
cfg.Host("localhost", "/", h => { h.Username("guest"); h.Password("guest");});
cfg.ConfigureEndpoints(context);
});
});
In OrderConfirmation.API
public class ConfirmationConsumer : IConsumer<OrderConfirmed>
{
public async Task Consume(ConsumeContext<OrderConfirmed> context)
{
Console.WriteLine($"Order {context.Message.OrderId} has been confirmed.");
await Task.CompletedTask;
}
}
and in Program.cs
builder.Services.AddMassTransit(x =>
{
x.AddConsumer<ConfirmationConsumer>();
x.UsingRabbitMq((context, cfg) =>
{
cfg.Host("localhost", "/", h => { h.Username("guest"); h.Password("guest");});
cfg.ConfigureEndpoints(context);
});
});
Shared Models and States and Events
namespace SharedModels.OrderProcessModels
{
// Event when an order is created
public class OrderCreated
{
public Guid OrderId { get; set; }
}
// Response for Order creation
public class OrderCreatedResponse
{
public Guid OrderId { get; set; }
public string Status { get; set; } = "OrderCreated";
}
// Event when the payment is processed
public class PaymentProcessed
{
public Guid OrderId { get; set; }
}
// Response after Payment is processed
public class PaymentProcessedResponse
{
public Guid OrderId { get; set; }
public string Status { get; set; } = "PaymentProcessed";
}
// Event when the order is confirmed
public class OrderConfirmed
{
public Guid OrderId { get; set; }
}
// Response for order confirmation
public class OrderConfirmedResponse
{
public Guid OrderId { get; set; }
public string Status { get; set; } = "OrderConfirmed";
}
// State object that tracks the lifecycle of an order
public class OrderState : SagaStateMachineInstance
{
public Uri ResponseAddress { get; set; }
public Guid RequestId { get; set; } // Optionally store RequestId for correlation
public Guid CorrelationId { get; set; } // Unique identifier for Saga instance
public Guid OrderId { get; set; } // Order identifier
public State CurrentState { get; set; } // Current state of the order in saga
public DateTime? CreatedAt { get; set; } // When the order was created
public DateTime? PaymentAt { get; set; } // When the payment was processed
public DateTime? ConfirmedAt { get; set; } // When the order was confirmed
}
}
Now the Order.API I created OrderStateMachine.cs
public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
public State PaymentProcessed { get; private set; }
public State OrderConfirmed { get; private set; }
public Event<OrderCreated> OrderCreatedEvent { get; private set; }
public Event<PaymentProcessed> PaymentProcessedEvent { get; private set; }
public Event<OrderConfirmed> OrderConfirmedEvent { get; private set; }
public OrderStateMachine()
{
InstanceState(x => x.CurrentState);
// Correlate Events with Saga Instance by OrderId
Event(() => OrderCreatedEvent, x => x.CorrelateById(c => c.Message.OrderId));
Event(() => PaymentProcessedEvent, x => x.CorrelateById(c => c.Message.OrderId));
Event(() => OrderConfirmedEvent, x => x.CorrelateById(c => c.Message.OrderId));
// Initial State -> PaymentProcessed
Initially(
When(OrderCreatedEvent)
.Then(context =>
{
context.Saga.OrderId = context.Message.OrderId; // Save the order ID
context.Saga.CreatedAt = DateTime.UtcNow; // Set the creation timestamp
Console.WriteLine("Order created, transitioning to PaymentProcessed...");
})
.TransitionTo(PaymentProcessed) // Move to PaymentProcessed state
.Publish(context => new PaymentProcessed
{
OrderId = context.Saga.OrderId
})
);
// PaymentProcessed State -> OrderConfirmed
During(PaymentProcessed,
When(PaymentProcessedEvent)
.Then(context =>
{
context.Saga.PaymentAt = DateTime.UtcNow; // Set payment timestamp
Console.WriteLine("Payment processed, transitioning to OrderConfirmed...");
})
.TransitionTo(OrderConfirmed) // Move to OrderConfirmed state
.Publish(context => new OrderConfirmed
{
OrderId = context.Saga.OrderId
})
);
//Final State -> Complete Saga
During(OrderConfirmed,
When(OrderConfirmedEvent)
.Then(context =>
{
context.Saga.ConfirmedAt = DateTime.UtcNow; // Set confirmation timestamp
Console.WriteLine($"Order confirmed for OrderId: {context.Saga.OrderId}");
})
.Finalize() // Finalize saga after order is confirmed
);
SetCompletedWhenFinalized();
}
}
OrderController.cs in Order.API
namespace Order.API.Controllers
{
[ApiController]
[Route("api")]
public class OrderController : ControllerBase
{
private readonly IRequestClient<OrderCreated> _requestClient; //bcz wait for the first one, then execute further.
public OrderController(IRequestClient<OrderCreated> requestClient)
{
_requestClient = requestClient;
}
[HttpPost("PlaceOrder")]
public async Task<IActionResult> PlaceOrder([FromBody] CreateOrderModel model)
{
try
{
var status = await _requestClient.GetResponse<OrderConfirmed>(new OrderCreated { OrderId = orderId });
if (status != null)
{
return Ok(new { OrderId = orderId, Status = "Order confirmed" });
}
else
{
return StatusCode(500, "Order processing failed.");
}
}
catch (Exception e)
{
return StatusCode(504, "Timeout waiting for the order confirmation.");
}
}
}
public class CreateOrderModel
{
public string OrderId { get; set; }
public string CustomerName { get; set; }
public double Amount { get; set; }
}
}
Upvotes: 0
Views: 30