EventStore persistence for MassTransit
License: Apache License 2.0
Event Store has been updated to 20.6.0 now
https://eventstore.com/blog/event-store-20.6.0-release/
And they start promoting the new gRPC client, EventStore.Client.Grpc.Streams, and start disable TCP be default.
Will this package be updated to the new gRPC packages?
I wonder if it is possible to persist manually stream of events which will represent a saga instance in specified state?
I try to do some integration tests connected with sagas persisted in ES, here is my scenario:
During publishing of event2 I receive error that event2 is not expected in Initial saga state (like the saga would not exist). I suppose that I`m missing some MT configuration (using RabbitMQ).
Can you give me some advice if such scenario is possible?
Hello @alexeyzimarev !
I am using MassTransit in my .NET Core Service and came to the idea to use EventStore with it. As there is already an integration for it, it would be perfect to be able to use it in my .NET Core Application.
Do you plan to offer an .NET Core package?
Greetings,
OriginalDeveloper
Hi Alexey,
I'm working on a project which required saga features. The Sample-Courier provided by MassTransit describes everything I need, except the storing part.
So I'm writing a POC to be more familiar with Routing Slip pattern and so on...based on Sample-Courier and your MassTransit.EventStoreIntegration to store the saga state in EventStore.
When I start the SagaStateMachine host, I've got the following error:
"System.TypeLoadException: Method 'SendQuery' in type 'MassTransit.EventStoreIntegration.Saga.EventStoreSagaRepository`1' from assembly 'MassTransit.EventStoreIntegration, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null' does not have an implementation."
I noticed the Task SendQuery<T> method throw:
NotImplementedByDesignException("Redis saga repository does not support queries");
Can you explain me why this method is not implemented?
I hope it's clear enough...
Thanks for your help.
I get this exception, which could be ignored if I could set Json serializer settings:
invalid expiration '-68745': {value_negative,-68745}) (Self referencing loop detected for property 'Warehouse' with type 'REST.NWMS.Warehouse'. Path 'Warehouses.Value[0].SupervisorOrg[0]'.)
---> MassTransit.RabbitMqTransport.MessageNotConfirmedException: rabbitmq://localhost/import-microservice-nwms-service-warehouse_delay => The message was not confirmed: PRECONDITION_FAILED - invalid expiration '-68745': {value_negative,-68745}
at async Task MassTransit.RabbitMqTransport.Contexts.RabbitMqModelContext.MassTransit.RabbitMqTransport.ModelContext.BasicPublishAsync(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, byte[] body, bool awaitAck)
at async Task MassTransit.RabbitMqTransport.Transport.RabbitMqSendTransport+SendPipe<T>.Send(ModelContext modelContext)
at async Task MassTransit.RabbitMqTransport.Transport.RabbitMqSendTransport+SendPipe<T>.Send(ModelContext modelContext)
at async Task GreenPipes.Agents.PipeContextSupervisor<TContext>.GreenPipes.IPipeContextSource<TContext>.Send(?)
at async Task GreenPipes.Agents.PipeContextSupervisor<TContext>.GreenPipes.IPipeContextSource<TContext>.Send(?)
at async Task GreenPipes.Agents.PipeContextSupervisor<TContext>.GreenPipes.IPipeContextSource<TContext>.Send(?)
at async Task<ScheduledMessage<T>> MassTransit.RabbitMqTransport.Scheduling.DelayedExchangeScheduleMessageProvider.ScheduleSend<T>(Uri destinationAddress, DateTime scheduledTime, Task<T> message, IPipe<SendContext<T>> pipe, CancellationToken cancellationToken)
at async Task MassTransit.Pipeline.Filters.RedeliveryRetryFilter<TContext, TMessage>.Send(TContext context, IPipe<TContext> next)
--- End of inner exception stack trace ---
---> (Inner Exception #1) Newtonsoft.Json.JsonSerializationException: Self referencing loop detected for property 'Warehouse' with type 'IME.CustomerManagement.Models.Infrastructures.REST.NWMS.Warehouse'. Path 'Warehouses.Value[0].SupervisorOrg[0]'.
at bool Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.CheckForCircularReference(JsonWriter writer, object value, JsonProperty property, JsonContract contract, JsonContainerContract containerContract, JsonProperty containerProperty)
at bool Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.CalculatePropertyValues(JsonWriter writer, object value, JsonContainerContract contract, JsonProperty member, JsonProperty property, out JsonContract memberContract, out object memberValue)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeObject(JsonWriter writer, object value, JsonObjectContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeList(JsonWriter writer, IEnumerable values, JsonArrayContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeObject(JsonWriter writer, object value, JsonObjectContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeList(JsonWriter writer, IEnumerable values, JsonArrayContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeObject(JsonWriter writer, object value, JsonObjectContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeObject(JsonWriter writer, object value, JsonObjectContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
at void Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.Serialize(JsonWriter jsonWriter, object value, Type objectType)
at void Newtonsoft.Json.JsonSerializer.SerializeInternal(JsonWriter jsonWriter, object value, Type objectType)
at void Newtonsoft.Json.JsonSerializer.Serialize(TextWriter textWriter, object value)
at byte[] MassTransit.EventStoreIntegration.Audit.EventStoreMessageAudit.Serialise(object event)
at Task MassTransit.EventStoreIntegration.Audit.EventStoreMessageAudit.StoreMessage<T>(T message, MessageAuditMetadata metadata)
at Task MassTransit.Audit.Observers.AuditSendObserver.MassTransit.ISendObserver.PostSend<T>(SendContext<T> context)
at Task MassTransit.Pipeline.Observables.SendObservable.PostSend<T>(SendContext<T> context)+(ISendObserver x) => { }
at Task GreenPipes.Util.Connectable<T>.ForEachAsync(Func<T, Task> callback)
Hi Alexey,
Thanks for your sharing your work.
I'm willing to use it for my projet but from what I see, this is not compatible with the last version of MassTransit.
Any chance you can take a look?
Thanks in advance.
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
Django
The Web framework for perfectionists with deadlines.
Laravel
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft
Open source projects and samples from Microsoft.
Google
Google ❤️ Open Source for everyone.
Alibaba
Alibaba Open Source for everyone
D3
Data-Driven Documents codes.
Tencent
China tencent open source team.