Private static async Task RunPoc(IServiceProvider serviceProvider) Var busControl = serviceProvider.GetRequiredService() Var serviceProvider = services.BuildServiceProvider() R.ConcurrencyMode = ConcurrencyMode.Optimistic ĬonfigureMassTransit.SetEndpointNameFormatter(new DefaultEndpointNameFormatter(true)) ĬonfigureMassTransit.UsingActiveMq((context, config) =>Ĭonfig.Host("artemis", 61616, configureHost => Services.AddMassTransit(configureMassTransit =>ĬonfigureMassTransit.AddSagaStateMachine() M.MigrationsHistoryTable($"_EFMigrationsHistory_Sagas") M.MigrationsAssembly(typeof(MySagaDbContext).Assembly.GetName().Name) => builder.UseSqlServer(connectionString, m => Services.AddDbContext((provider, builder) Public static async Task Main(string args) R.Interval(5, TimeSpan.FromMilliseconds(100)) R.Handle(y => y.InnerException is SqlException e & e.Number = 2627)
#MASSTRANSIT IBUS VS IBUSCONTROL CODE#
This is the SQLServer error code for duplicate key Protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator consumerConfigurator) => Public class ExternalCheckRequestConsumerDefinition : ConsumerDefinition
Protected override void Configure(EntityTypeBuilder entity, ModelBuilder model)Įntity.Property(x => x.CurrentState).HasMaxLength(128) Įntity.Property(x => x.RowVersion).IsRowVersion() Public class MySagaClassMap : SagaClassMap Protected override IEnumerable Configurations
I am keeping this as a backup option in case I can't find any other solution.Public interface IInitialSagaEvent : CorrelatedBy However, taking this approach will impact large sections of code / tests that rely on the IBusControl interface. I have tried the following change and it seems to flow the Ids across the Consumers. With the above approach the 'CorrelationId' header is lost when the SecondConsumer's filters are run. Ppc.ConnectPublishPipeSpecificationObserver(new SimplePublishPipeSpecObserver()) Specification.AddPipeSpecification(new SimplePublishMessagePipeSpec()) Īdded to config via x.UsingRabbitMq((context, cfg) =>Ĭfg.ConnectConsumerConfigurationObserver(new SimpleConsumePipeSpecObserver()) Public void MessageSpecificationCreated(IMessagePublishPipeSpecification specification) Public class SimplePublishPipeSpecObserver : IPublishPipeSpecificationObserver Public class SimplePublishMessagePipeSpec : IPipeSpecification> where TMessage : classīuilder.AddFilter(new SimplePublishMessageFilter()) If (context.Message is BaseMessage baseEvent & !)Ĭontext.ConversationId = baseEvent.ConversationId ? Guid.NewGuid() Ĭ("ConversationId", ()) If (("ConversationId", out object conversationId = conversationId Public async Task Send(PublishContext context, IPipe> next) PublishFilter public class SimplePublishMessageFilter : IFilter> where TMessage : class Public void ConsumerMessageConfigured(IConsumerMessageConfigurator configurator)Ĭonfigurator.AddPipeSpecification(new SimpleConsumeMessagePipeSpec()) Public void ConsumerConfigured(IConsumerConfigurator configurator) Public class SimpleConsumePipeSpecObserver : IConsumerConfigurationObserver Public class SimpleConsumeMessagePipeSpec : IPipeSpecification>īuilder.AddFilter(new SimpleConsumeMessageFilter, TMessage>()) LogContext.PushProperty("InitiatorId", context.InitiatorId) LogContext.PushProperty("ConversationId", context.ConversationId) LogContext.PushProperty("CorrelationId", context.CorrelationId) Public async Task Send(TContext context, IPipe next) There is a message processing graph that looks likeįirstConsumer -> SecondConsumer -> ThirdConsumer.ĬonsumeFilter public class SimpleConsumeMessageFilter : IFilter
CorrelationId - Unique Id for tracing logs within a Consumer.ConversationId - Unique Id for tracing logs of all Consumers in a graph.I am trying to implement a solution for tracing messages end to end by using the Ids. The above code is an example and the actual code base has 30+ consumers and all have the same pattern, i.e there is a Service per Consumer and the message is passed to the Service for processing. Public Task Process(FirstMessage firstMessage) Public FirstService(IBusControl busControl, ILogger logger) Private readonly IBusControl _busControl NET Core application with RabbitMQ(for local development) and SQS(for deployment) where a single message processing can result in multiple new messages getting created and processed.Īll the messages share the same base type public class BaseMessage : CorrelatedBy
0 kommentar(er)
