Repository navigation
CrossDomainSaga : Unable to invoke IReadModeLocator #798
Description
I am in the process of creating a CrossdomainSaga through rabbitmq messaging, Application A . creates a domain event as "SagaConsultantCreateEvent" on RabbitMq , the other application Application B subscribe to a "SagaQueue".
Then I created a DomainEventPublisher, which invoke AllEventsSubscriber successfully. This subscriber add consultant in eventstore successfully but unable to invoke IReadModelLocator.
A) The AllEventsSubscriber is listed as below
`
public class AllEventsSubscriber : ISubscribeSynchronousToAll
{
private readonly ILogger _logger;
private readonly IBootstrapper _bootstrapper;
private readonly IConsultantManagementApplicationService _consultantManagementApplicationService;
public AllEventsSubscriber( ILogger logger, IBootstrapper bootstrapper,
IConsultantManagementApplicationService consultantManagementApplicationService)
{
_logger = logger;
_bootstrapper = bootstrapper;
_consultantManagementApplicationService = consultantManagementApplicationService;
_bootstrapper.StartAsync(CancellationToken.None).Wait();
}
public async Task HandleAsync(IReadOnlyCollection<IDomainEvent> domainEvents,
CancellationToken cancellationToken)
{
IDomainEvent domainEvent = domainEvents.FirstOrDefault();
if (!domainEvent.Metadata.EventName.StartsWith("Saga")) return;
IAggregateEvent aggregateEvent = domainEvent.GetAggregateEvent();
var eventName = aggregateEvent.ToString();
IIdentity identity = domainEvent.GetIdentity();
try
{
switch (eventName)
{
case "ConsultantAggregate/SagaConsultantCreatedEvent":
{
await _consultantManagementApplicationService.AddNewAsync(((SagaConsultantCreatedEvent)aggregateEvent).ConsultantRecord, cancellationToken);
_logger.LogInformation($"new consultant added id:{identity}");
break;
}
}
}
catch (Exception exp)
{
_logger.LogInformation($"Failed to add new consultant id:{identity} , Exception: {exp.Message}");
}
}
}
}`
B) In ConsultantManagementApplicationService, the add event is added as following
` public async Task<bool> AddNewAsync(Core.Model.ConsultantAggregate.Consultant model, CancellationToken token)
{
var consultantById = await GetById(model.Id);
if (consultantById != null) return false;
var cmd = new AddNewConsultantCommand(model.Id, model);
await _bus.PublishAsync(cmd, token);
return true;
}`
C) ConsultantLocator is defined as following
`public class ConsultantLocator : IReadModelLocator
{
public IEnumerable<string> GetReadModelIds(IDomainEvent domainEvent)
{
var aggregateEvent = domainEvent.GetAggregateEvent();
switch (aggregateEvent)
{
case ConsultantCreatedEvent consultantRecordCreatedEvent:
yield return consultantRecordCreatedEvent.ConsultantRecord.Id.Value;
break;
case ConsultantEditedEvent consultantEdited:
yield return consultantEdited.ConsultantRecord.Id.Value;
break;
case ConsultantDeletedEvent consultantDeleted:
yield return consultantDeleted.ConsultantRecord.Id.Value;
break;
}
}
}`
d) In Startup locator is tied to readModel.
.RegisterServices(sr => sr.RegisterType(typeof(ConsultantLocator))) .UseElasticsearchReadModel<ConsultantReadModel, ConsultantLocator>()
Activity