MicroPlumberd.Services.Grpc.DirectConnect 1.0.10.40

There is a newer version of this package available.
See the version list below for details.
dotnet add package MicroPlumberd.Services.Grpc.DirectConnect --version 1.0.10.40                
NuGet\Install-Package MicroPlumberd.Services.Grpc.DirectConnect -Version 1.0.10.40                
This command is intended to be used within the Package Manager Console in Visual Studio, as it uses the NuGet module's version of Install-Package.
<PackageReference Include="MicroPlumberd.Services.Grpc.DirectConnect" Version="1.0.10.40" />                
For projects that support PackageReference, copy this XML node into the project file to reference the package.
paket add MicroPlumberd.Services.Grpc.DirectConnect --version 1.0.10.40                
#r "nuget: MicroPlumberd.Services.Grpc.DirectConnect, 1.0.10.40"                
#r directive can be used in F# Interactive and Polyglot Notebooks. Copy this into the interactive tool or source code of the script to reference the package.
// Install MicroPlumberd.Services.Grpc.DirectConnect as a Cake Addin
#addin nuget:?package=MicroPlumberd.Services.Grpc.DirectConnect&version=1.0.10.40

// Install MicroPlumberd.Services.Grpc.DirectConnect as a Cake Tool
#tool nuget:?package=MicroPlumberd.Services.Grpc.DirectConnect&version=1.0.10.40                

micro-plumberd

Micro library for EventStore, CQRS and EventSourcing Just eXtreamly simple.

Documentation can be found here: MicroPlumberd Documentation

Getting started

Install nugets:

# For your domain
dotnet add package MicroPlumberd
dotnet add package MicroPlumberd.SourceGenerators

If you'd like to use direct dotnet-dotnet communication to execute command-handlers install MicroPlumberd.DirectConnect

# For application-layer GRPC integration. (dotnet-2-dotnet)
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

  1. 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 });
}
// And events:
public record FooCreated { public string? Name { get; set; } }
public record FooUpdated { public string? Name { get; set; } }

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.
  1. Consume an aggregate.

If you want to create a new aggregate and save it to EventStoreDB:


FooAggregate aggregate = FooAggregate.New(Guid.NewGuid());
aggregate.Open("Hello");

await plumber.SaveNew(aggregate);

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

  1. 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.

  1. 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:

  1. You create a read-model that subscribes persistently.
  2. You subscribe it with plumber.
  3. You changed something in the event and want to see the new model.
  4. 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.

Product Compatible and additional computed target framework versions.
.NET net8.0 is compatible.  net8.0-android was computed.  net8.0-browser was computed.  net8.0-ios was computed.  net8.0-maccatalyst was computed.  net8.0-macos was computed.  net8.0-tvos was computed.  net8.0-windows was computed. 
Compatible target framework(s)
Included target framework(s) (in package)
Learn more about Target Frameworks and .NET Standard.

NuGet packages

This package is not used by any NuGet packages.

GitHub repositories

This package is not used by any popular GitHub repositories.

Version Downloads Last updated
1.0.77.125 37 11/26/2024
1.0.76.125 79 10/12/2024
1.0.75.125 76 10/12/2024
1.0.74.125 74 10/12/2024
1.0.73.124 77 10/12/2024
1.0.72.122 81 6/9/2024
1.0.71.121 91 6/7/2024
1.0.70.121 104 6/6/2024
1.0.69.119 104 5/15/2024
1.0.68.118 103 5/15/2024
1.0.67.118 102 5/15/2024
1.0.66.118 92 5/15/2024
1.0.65.117 94 5/15/2024
1.0.64.116 94 5/14/2024
1.0.63.115 91 5/11/2024
1.0.62.114 92 5/11/2024
1.0.61.113 89 5/11/2024
1.0.60.112 115 5/8/2024
1.0.58.112 96 5/8/2024
1.0.57.111 101 4/26/2024
1.0.55.111 105 4/23/2024
1.0.54.110 117 4/23/2024
1.0.53.110 110 4/23/2024
1.0.51.109 104 4/22/2024
1.0.50.109 106 4/22/2024
1.0.49.108 106 4/22/2024
1.0.48.108 119 4/20/2024
1.0.46.107 102 4/20/2024
1.0.45.106 106 4/20/2024
1.0.44.106 109 4/20/2024
1.0.43.100 108 4/17/2024
1.0.41.97 102 4/11/2024
1.0.40.95 105 4/11/2024
1.0.39.94 103 4/11/2024
1.0.37.94 86 4/9/2024
1.0.36.93 109 4/9/2024
1.0.35.90 115 4/8/2024
1.0.34.87 112 4/8/2024
1.0.29.85 96 4/7/2024
1.0.26.83 108 4/7/2024
1.0.20.72 117 3/24/2024
1.0.19.71 116 3/24/2024
1.0.17.71 121 3/24/2024
1.0.16.71 112 3/23/2024
1.0.15.70 96 3/23/2024
1.0.14.57 123 3/21/2024
1.0.13.56 96 3/21/2024
1.0.11.54 111 3/21/2024
1.0.10.40 120 3/17/2024
1.0.9.40 112 3/17/2024