gautema/cqrslite

A lightweight framework to help creating CQRS and Eventsourcing applications in C#

cqrslite's Issues

Adding new item generates an exception.

![Uploading CQRS Add.PNG…]()

Managing exceptions

I am not able to catch any exceptions that are thrown in the Handlers from the Command Sender. Is that normal ?
How to handle those in case of error ?

Read model handlers registration in Sample code

Hi,

I'm a little bit confusion. I will try to use read model form sample code in my application, but there seems to be a problem with read model handlers:

• InventoryItemDetailView.cs
• InventoryListView.cs

They are not registered as proper event handlers in service collection.
It's that true or maybe I do something wrong?

Best regards and thank you for help

Issue #50 is not fixed in .Net core 1.0

Thanks for looking into the issue #50 and fixing it but unfortunately the fix you posted is working only for .Net core 2.0 but it is not working in .Net core 1.0.
Please find the pull request #53 which contains the sample for .Net core 1.0 and also that shows it is not working.
To test in .Net core 1.0 I added a 'Name' property to ISession. This property is being set inside the command handler(CreateInventoryItemCommandHandler) when a new inventory is being created I am setting the Name property with the inventory name if it is not empty . When you look at subsequent requests the Name property value is the name of the first inventory item.
Also ServiceProvider class is not available in .Net core 1.0
Can you please look into it.

Async methods should accept a CancellationToken

I think it would be useful and make a great deal of sense if the async methods on the various interfaces and their implementations also took in an optional CancellationToken. This would include the following interfaces and their implementations:

• IEventStore
• IEventPublisher
• ISession
• IRepository
• ICommandSender

Get rid of the publish operation for the read model since it encourages to implement 2PC

This issue was found in Greg Young's sample app first,
and he said:

Submit a patch (its a relatively trivial one) just make the projections
read directly from the event store as opposed to the publish operation

Example of testing EventHandler

Hi, great project!,

In your Samples you have examples of how to test the WriteModel especially the commands.(With the Help of your Specification TestHelper class)
Can you give a example how to test the EventHandlers and the Events?

Regards
Michael

Pull Request for Apply/Emit pattern

I see that we use dynamic to call Apply method, it loose, we can remove apply method in somewhere without warning from compiler, so I want to implement Apply/Emit pattern into current code:

1. interface IEmit.cs:

/// <summary>
/// Marks an aggregate root as an emitter
/// </summary>
public interface IEmit<TEvent> where TEvent : IEvent
{
	/// <summary>
	/// Applies changes to the aggregate root instance
	/// </summary>
	void Apply(TEvent e);
}

And use it like this:

public partial class TestAggregate : AggregateRoot,
	IEmit<TestAggregateDidSomething>,
        IEmit<AnotherEvent>
{
	...
	
	public void DoSomething()
	{
		Emit(new TestAggregateDidSomething()); // We also change "ApplyChange" => "Emit"
	}
		
	public void Apply(TestAggregateDidSomething e)
	{
		...
	}

        public void Apply(AnotherEvent e)
	{
		...
	}
}

Use interface can help us make sure that we implemented event and we can know what event we implemented.

Can I create a pull request for this ?

IEventStore - concept of stream

Hi,

I am trying to plug Greg Young's EventStore (GES) to the CQRSlite framework with the provided IEventStore interface. Currently I cannot really decide how to proceed with the concept of streams from the GES. In the IEventStore we have:
Task Save(IEnumerable<IEvent> events, CancellationToken cancellationToken = default(CancellationToken));

However, for the GES implementation we also need to provide stream name and expected version, like so:
Task<WriteResult> AppendToStreamAsync(string stream, long expectedVersion, IEnumerable<EventData> events)

What course of actions would you recommend in this case?
Thanks in advance!

Handler with interface as generic type not working anymore

Hi,

I tried upgrading from v0.14.4 to v0.16.0 but the following event handler stopped working:

using System.Threading.Tasks;
using CQRSlite.Events;

namespace Televic.CoCon.Framework.Notification
{
    public class NotificationEventHandler : IEventHandler<IEvent>
    {
        private readonly Notifier _notifier;

        public NotificationEventHandler(Notifier notifier)
        {
            _notifier = notifier;
        }

        public Task Handle(IEvent message)
        {
            _notifier.Notify(message);

            return Task.CompletedTask;
        }
    }
}

All my other event handlers are still working, but this is a special case since I'm trying to handle all events here instead of events of a specific type.

Incoherency in SpecEventStorage --> Get<T>

public IEnumerable<IEvent> Get<T>(Guid aggregateId, int fromVersion)
        {
            return Events.Where(x => x.Version > fromVersion);
        }

Is it voluntary that the x.Id == aggregateId is omitted in the event storage test component / Get method ? It causes an EventsOutOfOrderException due to the fact that I initialize a test with 2 events in the Given() method, targeting 2 different aggregates.

Updating the method to the following corrected the issue:

public IEnumerable<IEvent> Get<T>(Guid aggregateId, int fromVersion)
        {
            return Events.Where(x => x.Version > fromVersion && x.Id == aggregateId);
        }

Purpose of generic argument on IEventStore.Save<T>

I had a question regarding IEventStore.Save<T>. The generic type argument is not used, therefore we have to rely on runtime type information in order to be able to serialize the events to a datastore. What is the purpose of this type argument?

Dependancy Resolver Creates new instance of Singleton class

the reason behind this change is the CQRSLite Dpandancy Resolver creates new instance of Singleton object like IEventStore and IRepository different than the asp.net core injected instances.
i mean there is new diffrent instance from a singleton Object in the same Application LifeTime.
you can check my theory by debugging the Id property i added in InMemoryEventStore.
inject the IEventStore in HomeController and Compare the Id in HomeController instance and the Instance in IRepository when Dependacy resolver inject the IEventStroe when you Add new inventory Item.
you will find them diffrent, but after my changes there will be one instance of singlton objects.

i couldnt add new pull so i will put my changes here:

1- UseCQrsLiteMiddleWare

 using System;
using System.Threading.Tasks;
using CQRSlite.Config;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;

namespace CQRSlite
{
    // You may need to install the Microsoft.AspNetCore.Http.Abstractions package into your project
    public class CqrsLite
    {
        private readonly RequestDelegate _next;
        private readonly IServiceProvider _serviceProvider;

        public CqrsLite(RequestDelegate next, IServiceProvider serviceProvider, Type type)
        {
            _next = next;
            _serviceProvider = serviceProvider;
           
           
            var registrar = (BusRegistrar)_serviceProvider.GetService(typeof(BusRegistrar));
            registrar.Register(type);
        }

        public Task Invoke(HttpContext httpContext)
        {
           
            return _next(httpContext);
        }
    }

    // Extension method used to add the middleware to the HTTP request pipeline.
    public static class UseCqrsLiteExtensions
    {
        public static IApplicationBuilder UseCqrsLite(this IApplicationBuilder builder, Type type)
        {
            return builder.UseMiddleware<CqrsLite>(type);
        }
    }
}

2- Startup Class in Web project

using System;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using CQRSlite.Bus;
using CQRSlite.Commands;
using CQRSlite.Events;
using CQRSlite.Domain;
using CQRSCode.WriteModel;
using CQRSlite.Cache;
using CQRSCode.ReadModel;
using CQRSlite.Config;
using CQRSCode.WriteModel.Handlers;
using Scrutor;
using System.Reflection;
using System.Linq;
using CQRSlite;

namespace CQRSWeb
{
    public class Startup
    {
        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddMemoryCache();

            //Add Cqrs services
            services.AddSingleton<InProcessBus>(new InProcessBus());
            services.AddSingleton<ICommandSender>(y => y.GetService<InProcessBus>());
            services.AddSingleton<IEventPublisher>(y => y.GetService<InProcessBus>());
            services.AddSingleton<IHandlerRegistrar>(y => y.GetService<InProcessBus>());
            services.AddScoped<ISession, Session>();
            services.AddSingleton<IEventStore, InMemoryEventStore>();
            services.AddScoped<ICache, CQRSlite.Cache.MemoryCache>();
            services.AddScoped<IRepository>(y => new CacheRepository(new Repository(y.GetService<IEventStore>()), 
                y.GetService<IEventStore>(), y.GetService<ICache>()));
            services.AddSingleton<IServiceLocator, DependencyResolver>();
            services.AddTransient<IReadModelFacade, ReadModelFacade>();
            services.AddScoped<BusRegistrar>(y => new BusRegistrar(y.GetService<IServiceLocator>()));
            //Scan for commandhandlers and eventhandlers
            services.Scan(scan => scan
                .FromAssemblies(typeof(InventoryCommandHandlers).GetTypeInfo().Assembly)
                    .AddClasses(classes => classes.Where(x => {
                        var allInterfaces = x.GetInterfaces();
                        return 
                            allInterfaces.Any(y => y.GetTypeInfo().IsGenericType && y.GetTypeInfo().GetGenericTypeDefinition() == typeof(ICommandHandler<>)) ||
                            allInterfaces.Any(y => y.GetTypeInfo().IsGenericType && y.GetTypeInfo().GetGenericTypeDefinition() == typeof(IEventHandler<>));
                    }))
                    .AsSelf()
                    .WithTransientLifetime()
            );

           
            // Add framework services.
            services.AddMvc();
        }

        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory loggerFactory)
        {
            app.UseDeveloperExceptionPage();
            app.UseStaticFiles();



            app.UseCqrsLite(typeof(InventoryCommandHandlers));

            app.UseMvc(routes =>
            {
                routes.MapRoute(
                    name: "default",
                    template: "{controller=Home}/{action=Index}/{id?}");
            });
        }
    }

 
}

Propagate information about handler implementation into RegisterHandler

It would be good to have information about handler implementation inside RegisterHandler method. In my case I'm implementing bus with RabbitMQ and when I register appropriate queues it would be good to give names to that queues as a full name of class that implements handler(s). But now I can't get access to that class. So,

void RegisterHandler<T, H>(Func<T, CancellationToken,Task> handler) where T : class, IMessage where H : class, IMessageHandler<T>;

would do the trick. I can submit new pull request if you think it would be useful. But for me it would be VERY useful!

Implementing a Process Manager

On a previous Issue I asked of the difference between Process Manager and Session and ended up creating a process manager.
One of the issue I am facing is that as a result several Aggregate are being tracked in the Session. When I commit for the first time the tracked aggregates are not cleared therefore still tracked. That's probably because
So during the second commit, it tries to save the aggregate previously save again but this create a Concurrency Issue because the version of the aggregate descriptor is not updated.

One solution I have found it to update the commit with this
await Task.WhenAll(_trackedAggregates.Values.Where(aggregate => aggregate.Aggregate.GetUncommittedChanges().Any()).Select(x => _repository.Save(x.Aggregate, x.Version, cancellationToken)));

In order to only save Aggregates that does not have any other changes to save.

The other option would be to not wait for the save of aggregate to be finished to actually clear the tracked aggregates of the session but I can't figure a way to make that work.

Does that make sense or am I going to break something I do not realize yet ?

Relate to another entity and get it

How to relate an entity to another entity? I have to use the AggregationId as Foreign Key or a Id Version data? When I need an specific version how I do to get it?

Is the "Infrastructure" stuff to call "Apply" method by reflection really needed?

I am taking my first steps into CQRS/ES and I am studying the code of this project.

Looking at the InventoryItem WriteModel supplied in CQRSCode project, we can notice the following methods:

        private void Apply(InventoryItemDeactivated e)
        {
            _activated = false;
        }

        public void Deactivate()
        {
            if(!_activated) throw new InvalidOperationException("already deactivated");
            ApplyChange(new InventoryItemDeactivated(Id));
        }

When we call the Deactivate method, it calls the ApplyChange method defined in the AggragteRoot base class which will finally call the Apply method at derived class by reflection.

We could merge those methods into:

        public void Deactivate()
        {
            if(!_activated) throw new InvalidOperationException("already deactivated");
            _activated = false;
            ApplyChange(new InventoryItemDeactivated(Id));
        }

thus eliminating the need of the following line in AggregateRoot's ApplyChange method:

this.AsDynamic().Apply(@event);

and the Infrastructure folder.

But when I tried applying those changes to the CQRSLite.Tests project, the Domain Test "When_getting_an_aggregate > Should_apply_events()" failed because it uses the TestEventStore which creates a "SavedEvents" list with two TestAggregateDidSomething that were supposed to increment the DidSomethingCount by reflection (using the Apply method).

The only changes I did were removing the Apply method from the TestAggregate and moving the "DidSomethingCount++;" to the DoSomething() method.

As I said, I am new to the CQRS/ES world and I might be wrong, but from my point of view, those events in the TestEventStore list would only be created as a result of the execution of either the DoSomething() or DoSomethingElse() methods, which would validate the test.

Please correct me if I am wrong and sorry for my english.

Command execution result

First of all I would like to say that this framework is awesome :-) I used CQRSlite in two projects and I'm very happy. One thing I was missing is information from handler that command was execution properly. I mean some boolean value or structure like CommandExecutionResult { Succeed = true/false, Error = "" } will be helpful. As for now I will throw an exception, but in my opinion it isn't a proper way. So if you have any ideas I can implement them with a pleasure on some fork.

Process manager / Saga vs Session

I am pretty new at CQRS, and I was wondering, as there is no way in CQRS Lite (or at least I don;t see any) to handle long running processes or saga, is this why the Session exists ?
If not, I feel like there is something I don't understand in CQRSLite. Should we create our process managers/sagas ?

Would it make sense to have generic IEventStore implementation

The problem is that I want to add metadata to events, for example correlationId, issuer. I could derive from
IEvent but I would need to create wrapper for IEventStore and maybe casts in other places.

So for example instead of:
Task<IEnumerable<IEvent>> Get(Guid aggregateId, int fromVersion, CancellationToken cancellationToken = default(CancellationToken));

Could it be:

Task<IEnumerable<T>> Get(Guid aggregateId, int fromVersion, CancellationToken cancellationToken = default(CancellationToken)) where T:IEvent;

Bug in DynamicInvoker.

IHandler<>.Handle is not called in explicitly implemented ICommandHandler<> classes.

CQRSlite.Infrastructure.DynamicInvoker.GetMember(Type, string, Type[]) incorrectly searches for a method by name.

Dirty fix is coming soon in a pull-request. The issue should not be closed until DynamicInvoker call stack is not fixed/refactored. Currently method name is provided simply as a string. This is bad. For example in VB.NET explicitly implemented interface methods may have names absolutely different from the names declared in interfaces, but they are still valid names, since interface names are used if object is used by an interface-typed variable.
Anyway, I had no time to check all the places where the DynamicInvoker.GetMember is used, but I believe that name as string should be replaced with something more sophisticated OR full name may be used as alternative (i.e. "CQRSlite.Messages.IHandler<SomeCommand>.Handle" instead of "Handle"), but latter approach is a crazy thing.

#63

Setup Sample without .NET Core DI?

Hi,

how would you set up the sample without .NET Core DI and the DepdencyResolver to simply use it on the console (for demonstration purposes without posting Commands via HTTP)?

InProcessBus cannot get the type of sent command if it is generated dynamically

I'm building an API with one endpoint for all commands and use the method described here to extract which command is being sent.

Here is the code I'm using.

[JsonConverter(typeof(JsonCommandConverter))]
public interface IBaseCommand : ICommand
{
    Guid Id { get; set; }
}

public class FooCommand : BaseCommand
{
}

class CommandHandler : ICommandHandler<FooCommand>
{
    public void Handle(FooCommand message)
    {
    }
}

public class CommandsController : ApiController
{
    private InProcessBus _bus;

    public CommandsController()
    {
        var handler = new CommandHandler();

        _bus = new InProcessBus();
        _bus.RegisterHandler<FooCommand>(c =>
        {
            handler.Handle(c);
        });
    }

    public void Post(IBaseCommand command)
    {
        _bus.Send(command);
    }
}

Running this will give me a FooCommand in the Post method but when I send it through the bus I get No handler registered.

This is due to this line in InProcessBus which uses typeof(T) which will result in IBaseCommand in my case. This could be fixed by using command.GetType() instead.

If this is an acceptable change I can create a pull request.

Nice projet, documentation missing

Your project looks promising.
The main problem is that it's not documented at all...

It would be really good to have a small documentation that explains the core principle, with the event mechanisms for example... The example you did is really good, but it's not sufficient.

Also, you could explain what is the goal of the Session, how one could interchange Autofact with another DI Framework, how one could bind Entity Framework, if it is possible to use the Window Service Bus instead of the InProcessBus class (or any other similar library), what is the snapshot strategy, and probably a lot more.
And what is the exact difference with Greg Young's project?

Concerning the code, I didn't go into the details, but I saw 2-3 small things:

• Use TimeSpan.FromXXX instead of new TimeSpan(x, x, x....)
• In the LoadAggreate method of the Repository, you could create the Aggregate just before the LoadFromHistory method
• The IEventHandler interface could be contravariant

Getting rid of the service locator

Hi Gaute!

I'm trying to use SaasKit.Multitenancy together with CQRSlite to add multi tenant support. The goal is to have a separate EventStore for each tenant.

SaasKit is using DI and a separate TenantResolver to resolve the current tenant based on the parameter of your choice. However, when used together with CQRSlite, it looks like the DependencyResolver is creating some problems.

I have added Tenant as a constructor param to InMemoryEventStore, but since the InMemoryEventStore is resolved via the custom CQRSlite DependencyResolver it's always null. It seems to me that DependencyResolver is only able to resolve stuff that was added to the container when the app starts and that this is a completely separate container than the build-in ASP.NET DI Container.

Would it be possible to get rid of the custom DependencyResolver or handle this in some other way?

I just created a fork where the issue can be seen.

Check the following line and see that tenant is null when InMemoryEventStore is created:
https://github.com/henningst/CQRSlite/blob/master/Sample/CQRSCode/WriteModel/InMemoryEventStore.cs#L18

Support for .Net Standard

Hi, can you tell me whether you already adhere to .Net Standard or is it on the roadmap (By the way, is there a roadmap)?

Using Autofac as DI container

Hi,

I am trying to integrate CQRSlite with a solution that uses NServiceBus. NServiceBus uses Autofac as its DI container. To integrate everything in Autofac, I tried the following :

builder.RegisterType<Session>().As<ISession>();
builder.RegisterType<SFEventStore>().As<IEventStore>();
builder.RegisterType<MemoryCache>().As<ICache>();
builder.Register(c =>
            {
                return new CacheRepository(new Repository(c.Resolve<IEventStore>()), c.Resolve<IEventStore>(), c.Resolve<ICache>());

            })
            .As<IRepository>();

This causes a circular dependency exception on the IRepository. When an ISession is resolved, Autofac resolves an IRepository to a CacheRepository, which in turn resolves the Repository to a CacheRepository because the constructor of a CacheRepository requires an IRepository:

public CacheRepository(IRepository repository, IEventStore eventStore, ICache cache);

Any idea how to solve this?

I really want to use CQRSlite because it contains great functionality for our solution!

Thanks,
Karl

AggregateRoot versioning causes ConcurrencyException

Completely open to the fact that this might just be my code here as I'm new to CqrsLite and the Cqrs concept in general.

I'm using event store (https://geteventstore.com/) to persist the change events to my aggregate root. My aggregate is that of a Booking, containing date, time, customer, etc. I have the approach of creating a stream in event store per aggregate instance, meaning a stream only includes events related to a single Booking. When a stream doesn't exist, event store reports the expected version as -1, so when I create my entity, I have to set version as -1 in the Apply method for my BookingCreatedEvent (ultimately called by the Booking constructor). As long as I specify -1 as the version, the creation event is persisted to event store.

I have an issue now where I am trying to persist a BookingDateTimeUpdated event, which should result in a version of 1. The steam already exists and the version of that stream is 0. The aggregate is loaded with version 0 as expected, this seems to be entered into the _trackedAggregates (of the Session object), with the version as it was when loaded.

So, all good til now. Now I want to apply the BookingDateTimeUpdated event, so I do that and when it comes to save the event, in the Repository, the save method first does this:

if (expectedVersion != null && (await _eventStore.Get(aggregate.Id, expectedVersion.Value)).Any())
{
    throw new ConcurrencyException(aggregate.Id);
}

My understanding of that bit of code is that; it's looking to see if there is already a version of that stream equal to that which we are trying to save, meaning that we didn't have the full state when our event was applied to the aggregate, so yeah, throw an exception to prevent corrupting the data.

The expected version comes down from the _trackedAggregates of Session as 0 (that's what was set at load time) and because version 0 already exists, a ConcurrencyException is thrown. I would expect the version that we are trying to save after a new event is applied to be version 1, not 0.

I can't work out if this is a bug in CqrsLite, that the expected version that Repository.Save uses should come from somewhere else or if I'm not setting things up correctly.

AspNetCore configuration extension

I've seen in your sample project the default configuration for an AspNetCore app.

I would like to have a configuration helper to configure CQRSLite.

services.AddCQRSLite() .AddEventStore<InMemoryEventStore>() .RegisterCommandHandler(typeof(InventoryCommandHandlers)) .RegisterEventHandler(typeof(InventoryCommandHandlers));

I propose a pull request with this behavior implemented

Please make use of ConfigureAwait(false)

Hi.

In my own code I follow the best practices Steven Cleary recommends.
As there is no UI thread context in CQRSlite, the use of .ConfigureAwait(false) IMHO makes sense and could improve performance.

Or is there a reason not making use of it in CQRSlite?

Kind regards
Bernd

Autofac integration

Hi @gautema,

Thanks for the wonderful project. I have more of a question than an issue...I'm trying to get the sample project to work with Autofac, I've gone about doing that by converting the .net core DI code to Autofac, here is my startup.cs:

public class Startup
{
    public IContainer ApplicationContainer { get; private set; }
    public IServiceProvider ConfigureServices(IServiceCollection services)
    {
        services.AddMemoryCache();
        var builder = new ContainerBuilder();
        services.AddMvc();
        // services.AddSingleton<InProcessBus>(new InProcessBus());
        builder.Register<InProcessBus>(x => new InProcessBus()).AsSelf().SingleInstance();
        builder.Register<ICommandSender>(c => c.Resolve<InProcessBus>()).SingleInstance();
        builder.Register<IEventPublisher>(c => c.Resolve<InProcessBus>()).SingleInstance();
        builder.RegisterType<Session>().As<ISession>().InstancePerLifetimeScope();
        builder.RegisterType<InMemoryEventStore>().As<IEventStore>().SingleInstance();
        builder.RegisterType<MemoryCache>().As<ICache>().InstancePerLifetimeScope();
        builder.Register<IRepository>(x =>
            new CacheRepository(new Repository(x.Resolve<IEventStore>()),
            x.Resolve<IEventStore>(), x.Resolve<ICache>())).InstancePerLifetimeScope();
        builder.RegisterType<ReadModelFacade>().As<IReadModelFacade>().InstancePerDependency();
        var targetAssembly = typeof(InventoryCommandHandlers).GetTypeInfo().Assembly;

        builder.RegisterAssemblyTypes(targetAssembly)
       .Where(type => type.GetInterfaces().Any(y => y.IsAssignableFrom(typeof(CQRSlite.Commands.ICommandHandler<>))))
       .AsImplementedInterfaces()
       .InstancePerDependency();
        

        builder.Populate(services);
        this.ApplicationContainer = builder.Build();

        var sp= new AutofacServiceProvider(ApplicationContainer);
        var registrar = new BusRegistrar(new DependencyResolver(sp));
        registrar.Register(typeof(InventoryCommandHandlers));

        return sp;
    }

}

However when i step through this, the GetService Method in DependecyResolver.cs can not resolve IHandlerRegistrar.

Any guidance would be appreciated.

Decorating handlers

I've been trying to figure out a nice way to decorate my handler implementations (e.g. with a logging decorater) without rolling my own version of the router and registrar, but it's harder than I imagined.

How do you usually go about applying decoraters to handlers, if at all?

Concurrency exception during back end processing

Hello,

Here is my situation... I'm currently trying to process some data on a back end and according to my workflow same object may be updated by different services simultaneously.

For example "Service A" and "Service B" should work with the same instance of the object and update the state. So, that's not a surprise that from time to time I'm getting ConcurrencyException.

Now I'm thinking how to resolve this situation. One of the possible options that I see is to implement kinda BackEndSession with almost the same functionality that already exists in the Session but with some changes on Commit. For example, if I catch some exception when try to save aggregator to repository I want to load fresh instance from repository, apply changes that were maid and try to save it again.

in other words, I'm trying to implement something like this

public void Commit()
{
    foreach (var descriptor in _trackedAggregates.Values)
    {
          var events = descriptor.Aggregate.GetUncommittedChanges();

          try
          {
              _repository.Save(descriptor.Aggregate, descriptor.Version);
          }
          catch (System.Exception)
          {
                //  pull fresh aggragator version from repository... 
                MethodInfo method = _repository.GetType().GetMethod("Get");
                MethodInfo generic = method.MakeGenericMethod(descriptor.Aggregate.GetType());
                var aggregator = generic.Invoke(_repository, new object[] { descriptor.Aggregate.Id }) as AggregateRoot;

                //  ... apply changes that were made...
                foreach (var e in events)
                {
                    aggregator.Apply(e);
                }

                //  ... and try to save it one more time
                _repository.Save(aggregator, aggregator.Version);
          }
    }
     _trackedAggregates.Clear();
}

the problem with this approach that aggregator.Apply(e); is private and I can't use it this way.

So, my question is there any other solution for my situation that I don't currently see?

Allow registration of specific message handlers

Hey @gautema,

Thanks for the effort you've put into CQRSlite. I've been using it for the last few months on a personal project and it's saved me a lot of time and effort. However, one thing that caused me some pain was how handlers are registered in RouteRegistrar. I have a suggestion for how this could be improved.

At the moment, the only option available is to register all the handlers in the same assembly as message type(s) I pass into RouteRegistrar.Register(...). This causes me problems because, in my use case, I don't want to register all the handlers in my code; I have a number of different handlers that might or might not be used, depending on the particular configuration of my code. I want to choose exactly which handlers I want to register. At the moment, the only option I have using is to split my handlers into multiple different assemblies.

To work around this limitation, I implemented my own code to allow me to explicitly pass in the handlers I want to register. So, now I can do this...

var handlers = new List<Type>(){
  // the message handlers I want to register
}

myRouteRegistrar.RegisterHandlers(handlers)

My implementation is much more complex than it ought to be; some of the registration code in CQRSlite is internal, so I need to use reflection and also copy and paste some of the code into my own project.

Implementing this directly in CQRSlite would be much simpler. I envisage that we'd make the following changes to RouteRegistrar:

1. Mark RouteRegistrar.Register(params Type[] typesFromAssemblyContainingMessages) as obsolete
2. Add a new method, RouteRegistrar.RegisterAllHandlersInSameAssemblyAs(params Type[] typesFromAssemblyContainingMessages) that behaves the same as the current RouteRegistrar.Register(...)
3. Add a new method, RouteRegistrar.RegisterHandlers(params Type[] messageHandlers) that registers the specific handlers passed in.

Is this something that you think would be useful to get into the CQRSlite code base? If so, I'll look at implementing this.

Recreate data model from events after application restart

Hi,

I'm stand in front of a problem. How to recreate data in data model from events after application restart?
Let's take an webapp from sample as an example and assume that we store events in real DB but our data model on the read part stays as a InMemoryDatabase. After app restart I would like to take all events from DB and force them to event handlers to recreate data model and keep up-to-date InMemoryDatabase. How do you menage that? Any suggestions, ideas? Could be done automatically via CQRS Lite?

Thank you in advance

Race condition in pesimistic concurrency management

I will always grow with new Guids, without any cleaning being performed in the concurrent dictionary of semaphores by Guid

Generic Constraints on IEventPublisher/IHandlerRegistrar

I'm trying to swap out the in-memory event implementations of the InProcessBus with RabbitMQ via the EasyNetQ library, and it's implementation makes sure that functions like Publish/Subscribe take classes instead of constraining on an interface. I have a workaround for Publish/Send that can use a base class; however, I can't seem to get a really non-hackish workaround for RegisterHandler. The error that I receive is: The type 'T' must be a reference type in order to use it as parameter 'T' in the generic type or method 'IMessageClient.SubscribeAsync<T>(string, Func<T, Task>)'

Would it be possible to add the class constraint to the Bus interfaces below?

//IHandlerRegistrar:
void RegisterHandler<T>(Func<T, Task> handler) where T : class, IMessage;

//IEventPublisher
Task Publish<T>(T @event) where T : class, IEvent;

//ICommandSender
Task Send<T>(T command) where T : class, ICommand;

Swapping out pieces

great project!
Is there any kind of docs or guidance if we want/need to swap out specific providers?

So like

1. What if we would like to use SQL Server as the backend for the event store?
2. What if we needed to use something like NServiceBus or Azure Service Bus?

Are these sort of things interchangeable now or would we need to fork and modify the code?

Cheers!
AJ

IEvent Interface Question

Just starting out with CQRS concepts and implementation so I am still learning how all this fits together. My current understanding of events is that they should be immutable. Doesn't having IEvent implement {set;} on each of the properties mean that an event is mutable by definition or does that come down to the implementation of the event itself.

public interface IEvent : IMessage
    {
        Guid Id { get; set; }
        int Version { get; set; }
        DateTimeOffset TimeStamp { get; set; }
    }

IEventPublisher Question

Hello, I am still learning event sourcing and trying examples out to get more deeper understanding.

Based on your Sample, I am confused why is Publish need to be called twice? On your Repository.cs

public void Save<T>(T aggregate, int? expectedVersion = null) where T : AggregateRoot
{
      if (expectedVersion != null && _eventStore.Get<T>(aggregate.Id, expectedVersion.Value).Any())
      {
            throw new ConcurrencyException(aggregate.Id);
      }

      var changes = aggregate.FlushUncommitedChanges();
       _eventStore.Save<T>(changes);

      if (_publisher != null)
      {
           foreach (var @event in changes)
           {
                _publisher.Publish(@event);
           }
      }
}

_eventStore.Save<T>(changes); Calls Publish from InMemoryEventStore class, then after that it calls Publish again on:

if (_publisher != null)
{
     foreach (var @event in changes)
     {
           _publisher.Publish(@event);
     }
}

Does InMemoryEventStore need Publish? Since Repository handles that?

Why do we need the cache if we don't read from it

after i debug the sample project i noticed that the CQRS flow is
Add Item -> Save to Repository -> Save to the Cache -> Save the event in the Event store

then the read event handler come into play which save to the read database so
why do we need to save items in the cache if we will not read from it?

Request scope objects are acting like singleton .Net core

When a new command is raised for each request, (e.g: CreateInventory request) commandhandler's corresponding dependent objects like DbContext, Repository are not being created newly even if they are registered in request scope using .Net core DI container. They are acting like singleton objects. For the first request they are getting created and for subsequent requests if they are being used outside the scope of handler like in controller methods they are getting created for each request but when it comes to the handler dependent objects are not being created newly and using the objects that are created when the first time handler is created.

Why the command needs to tell the expected version?

Race condition in Repository

Isn't there a possible race condition in the Save method of the Repository class? Furthermore, how to fix concurrency issues between multiple instances of a service (or should one then consider the actor model instead) ?

AggregateFactory is inaccessible due to its protection level

Hi,

Working on a repository (IRepository) with eventstore, I am now stuck on the implementation of the LoadAggregate() member, because the AggregateFactory is "internal" in the framework.

What am I missing here?

Exception when sending command

Hey guys

Just wanted to log that, with the latest build, I am getting an exception when I try to send a command via the Router:

System.NullReferenceException: Object reference not set to an instance of an object.
   at CQRSlite.Infrastructure.DynamicInvoker.GetTypeAndHash[T](T obj, String methodname, Object[] args, Type& type, Int32& hash)
   at CQRSlite.Infrastructure.DynamicInvoker.Invoke[T](T obj, String methodname, Object[] args)
   at CQRSlite.Routing.RouteRegistrar.<>c__DisplayClass3_0.<InvokeHandler>b__5(Object x, CancellationToken token)
   at CQRSlite.Routing.Router.<>c__DisplayClass1_0`1.<RegisterHandler>b__0(IMessage message, CancellationToken token)
   at CQRSlite.Routing.Router.Send[T](T command, CancellationToken cancellationToken)

Seems to be stemming from the latest change to PrivateReflectionDynamicObject, which was changed to DynamicInvoker with some changes to the hashing.

Cheers

Ben

Using database

Hi, please, how can i use a database on sample?

Optional expected version?

I'm a bit confused about the implementation of the expected version of an aggregate. You've said in #40 that expected version is optional on a command. I can see that ISession.Get<T>(...) defaults to a null expected version if no value is supplied; this will typically be called from my implementation of ICommandHandler when a ICommand is handled. However, I also see that ICommand does not allow a nullable expected version.

My particular use case is that I want to create a process manager to manage the interactions between two aggregates, AggregateA and AggregateB. The process manager subscribes to an event from AggregateA, and then sends a command to AggregateB. However, it cannot know what the expected version of AggregateB is, so it cannot use ICommand.

I'm not sure if my confusion is because I don't understand how process managers handle concurrency, or if I'm not understanding the code correctly, or if there is a bug in the code.

Any help is greatly appreciated!

Discussion of StreamId or AggregateId in Event

I'm looking into implementing an IEventStore for other event stores (such as Greg's EventStore, or StreamStone, et al), and one of the requirements for those storage backends is that the stream or aggregate id be used to persist the event.

Based on the current signature of IEvent, that is not possible. Would you be open to accepting a PR with AggregateId added to the IEvent signature? It may also be easy enough to just add the aggregate or stream id as an argument to the IEventStore Get/Save methods.

Any thoughts on this? Thanks!

Support for non-Guid Identity

Continuing my saga of implementing EventStore as an IEventStore implementation, I've found that limiting identity to Guid limits the flexibility in how we store and categorize our aggregates/events, with categorization being the primary driver personally.

I envision being able to use strings, such as Order-<someGuid>, or Order-1, which is especially important for event stores that use an identity prefix to categorize events and aggregates (such as EventStore).

Any thoughts or opinions on implementing something like this? It could be as simple as replacing Guid with string, or introducing some sort of IIdentity to use. I'd be more than happy to submit a PR for this change, but wanted to know what your thoughts were.