If you'd like to use direct dotnet-dotnet communication to execute command-handlers install MicroPlumberd.DirectConnect
dotnet add package MicroPlumberd.DirectConnect
Configure plumber
/// change to your connection-string.
string connectionString = $"esdb://admin:changeit@localhost:2113?tls=false&tlsVerifyCert=false";
var settings = EventStoreClientSettings.Create(connectionString);
var plumber = Plumber.Create(settings);
Aggregates
Write an aggregate.
[Aggregate]
public partial class FooAggregate(Guid id) : AggregateBase<FooAggregate.FooState>(id)
{
internal new FooState State => base.State;
public record FooState { public string Name { get; set; } };
private static FooState Given(FooState state, FooCreated ev) => state with { Name = ev.Name };
private static FooState Given(FooState state, FooUpdated ev) => state with { Name =ev.Name };
public void Open(string msg) => AppendPendingChange(new FooCreated() { Name = msg });
public void Change(string msg) => AppendPendingChange(new FooUpdated() { Name = msg });
}
Comments:
State is encapsulated in nested class FooState.
Given methods, that are used when loading aggregate from the EventStoreDB are private and static. State is encouraged to be immutable.
[Aggregate] attribute is used by SourceGenerator that will generate dispatching code and handy metadata.
Consume an aggregate.
If you want to create a new aggregate and save it to EventStoreDB:
If you want to load aggregate from EventStoreDB, change it and save back to EventStoreDB
var aggregate = await plumber.Get<FooAggregate>("YOUR_ID");
aggregate.Change("World");
await plumber.SaveChanges(aggregate);
Write a read-model/processor
Read-Models
[EventHandler]
public partial class FooModel
{
private async Task Given(Metadata m, FooCreated ev)
{
// your code
}
private async Task Given(Metadata m, FooUpdated ev)
{
// your code
}
}
Comments:
ReadModels have private async Given methods. Since they are async, you can invoke SQL here, or othere APIs to store your model.
Metadata contains standard stuff (Created, CorrelationId, CausationId), but can be reconfigured.
var fooModel = new FooModel();
var sub= await plumber.SubscribeModel(fooModel);
// or if you want to persist progress of your subscription
var sub2= await plumber.SubscribeModelPersistently(fooModel);
With SubscribeModel you can subscribe from start, from certain moment or from the end of the stream.
Processors
[EventHandler]
public partial class FooProcessor(IPlumber plumber)
{
private async Task Given(Metadata m, FooUpdated ev)
{
var agg = FooAggregate.New(Guid.NewGuid());
agg.Open(ev.Name + " new");
await plumber.SaveNew(agg);
}
}
Implementing a processor is technically the same as implementing a read-model, but inside the Given method you would typically invoke a command or execute an aggregate.
Features
Conventions
SteamNameConvention - from aggregate type, and aggregate id
EventNameConvention - from aggregate? instance and event instance
MetadataConvention - to enrich event with metadata based on aggregate instance and event instance
EventIdConvention - from aggregate instance and event instance
OutputStreamModelConvention - for output stream name from model-type
GroupNameModelConvention - for group name from model-type
Ultra development cycle for Read-Models (EF example).
Imagine this:
You create a read-model that subscribes persistently.
You subscribe it with plumber.
You changed something in the event and want to see the new model.
Instead of re-creating old read-model, you can easily create new one. Just change MODEL_VER to reflect new version.
Please note that Sql schema create/drop auto-generation script will be covered in a different article. (For now we leave it for developers.)
Comments:
By creating a new read-model you can always compare the differences with the previous one.
You can leverage canary-deployment strategy and have 2 versions of your system running in parallel.
[OutputStream(FooModel.MODEL_NAME)]
[EventHandler]
public partial class FooModel : DbContext
{
internal const string MODEL_VER = "_v1";
internal const string MODEL_NAME = $"FooModel{MODEL_VER}";
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
modelBuilder
.Entity<FooEntity>()
.ToTable($"FooEntities{MODEL_VER}");
}
private async Task Given(Metadata m, FooCreated ev)
{
// your code
}
private async Task Given(Metadata m, FooUpdated ev)
{
// your code
}
}
Subscription Sets - Models ultra-composition
You can easily create a stream that joins events together by event-type, and subscribe many read-models at once. Here it is named 'MasterStream', which is created out of events used to create DimentionLookupModel and MasterModel.
In this way, you can easily manage the composition and decoupling of read-models. You can nicely composite your read-models. And if you don't wish to decouple read-models, you can reuse your existing one.
/// Given simple models, where master-model has foreign-key used to obtain value from dimentionLookupModel
var dimentionTable = new DimentionLookupModel();
var factTable = new MasterModel(dimentionTable);
await plumber.SubscribeSet()
.With(dimentionTable)
.With(factTable)
.SubscribeAsync("MasterStream", FromStream.Start);
GRPC Direct communication
/// Let's configure server:
services.AddCommandHandler<FooCommandHandler>().AddServerDirectConnect();
/// Add mapping to direct-connect service
app.MapDirectConnect();
Here is an example of a command handler code:
[CommandHandler]
public partial class FooCommandHandler(IPlumber plumber)
{
[ThrowsFaultException<BusinessFault>]
public async Task Handle(Guid id, CreateFoo cmd)
{
if (cmd.Name == "error")
throw new BusinessFaultException("Foo");
var agg = FooAggregate.New(id);
agg.Open(cmd.Name);
await plumber.SaveNew(agg);
}
[ThrowsFaultException<BusinessFault>]
public async Task<HandlerOperationStatus> Handle(Guid id, ChangeFoo cmd)
{
if (cmd.Name == "error")
throw new BusinessFaultException("Foo");
var agg = await plumber.Get<FooAggregate>(id);
agg.Change(cmd.Name);
await plumber.SaveChanges(agg);
return HandlerOperationStatus.Ok();
}
}
And how on the client side:
service.AddClientDirectConnect().AddCommandInvokers();
// And invocation
var clientPool = sp.GetRequiredService<IRequestInvokerPool>();
var invoker = clientPool.Get("YOUR_GRPC_URL");
await invoker.Execute(Guid.NewId(), new CreateFoo(){});
Aspects
You can easily inject aspects through decorator pattern.