|
| 1 | +# EventDriven.CQRS |
| 2 | + |
| 3 | +An event-driven approach to Command Query Responsibility Segregation. |
| 4 | + |
| 5 | +### Prerequisites |
| 6 | +- [.NET Core SDK](https://dotnet.microsoft.com/download) (5.0 or greater) |
| 7 | +- [Docker Desktop](https://www.docker.com/products/docker-desktop) |
| 8 | +- MongoDB Docker: `docker run --name mongo -d -p 27017:27017 -v /tmp/mongo/data:/data/db mongo` |
| 9 | +- [MongoDB Client](https://robomongo.org/download): |
| 10 | + - Download Robo 3T only. |
| 11 | + - Add connection to localhost on port 27017. |
| 12 | +- [Dapr](https://dapr.io/) (Distributed Application Runtime) |
| 13 | + - [Install Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/) |
| 14 | + - [Initialize Dapr](https://docs.dapr.io/getting-started/install-dapr-selfhost/) |
| 15 | + |
| 16 | +## Introduction |
| 17 | + |
| 18 | +This project builds on the principles of [Domain Driven Design](https://en.wikipedia.org/wiki/Domain-driven_design) to provide a set of abstractions and reference architecture for implementing the [Command Query Responsibility Segregation](https://docs.microsoft.com/en-us/azure/architecture/patterns/cqrs) pattern, also known as as CQRS. Because entities process commands by emitting domain events, adding [event sourcing](https://microservices.io/patterns/data/event-sourcing.html) at a later time will be relatively straightforward. |
| 19 | + |
| 20 | +The **EventDriven.CQRS.Abstractions** library contains interfaces and abstract base classes to support these concepts: |
| 21 | +- **Entity**: A type that has an identity with behavior and state that can change over time. |
| 22 | +- **Command**: An object that is sent to the domain for a state change which is handled by a command handler. |
| 23 | +- **Event**: A statement of fact about what change has been made to the domain state. |
| 24 | + |
| 25 | +The **Reference Architecture** projects demonstrate how to apply these concepts to two microservices: `CustomerService` and `OrderService`. In addition, each service has *separate controllers for read and write operations*, thus segregating command and query responsibilities, with different sets of models, or Data Transfer Objects (DTO's). |
| 26 | +- **Query Controller**: Uses repository to retrieve entities and converts them to DTO's with AutoMapper. |
| 27 | +- **Command Controller**: Converts DTO's to domain entities using AutoMapper. Then hands control over to a command handler for executing business logic. |
| 28 | +- **Command Handler**: Uses a domain entity to process commands which generate one or more domain events, then requests entity to apply the domain events in order to mutate entity state. Persists entity state to a state store and optionally publishes an integration event which is handled by another microservice. |
| 29 | +- **Repository**: Persists entity state to a database. |
| 30 | +- **Event Bus**: Used to publish integration events, as well as subscribe to events using an event handler. Dapr is used to abstract away the underlying pub/sub implementation. The default is Redis (for local development), but Dapr can be configured to use other components, such as AWS SNS+SQS. |
| 31 | + |
| 32 | +> **Note**: This example illustrates a *simple* CQRS implementation with a **shared database** and **single service** for both read and write operations. A more sophisticated implementation might entail **separate services and databases** for read and write operations, using integration events to communicate between them. This simple example only uses integration events to communicate between the customer and order services. |
| 33 | +
|
| 34 | +<p align="center"> |
| 35 | + <img width="600" src="images/event-driven-cqrs-ref-arch.png"> |
| 36 | +</p> |
| 37 | + |
| 38 | +### Usage: Reference Architecture Projects |
| 39 | + |
| 40 | +1. Run Dapr Dashboard. |
| 41 | + - Then open http://localhost:8080 to view containers after executing `dapr run` commands. |
| 42 | + ``` |
| 43 | + dapr dashboard |
| 44 | + ``` |
| 45 | +2. Use Dapr to run the customer service. |
| 46 | + ``` |
| 47 | + dapr run --app-id customer-service --app-port 5000 -- dotnet run |
| 48 | + ``` |
| 49 | +3. Use Dapr to run the order service. |
| 50 | + ``` |
| 51 | + dapr run --app-id order-service --app-port 5050 -- dotnet run |
| 52 | + ``` |
| 53 | +4. Create some customers. |
| 54 | + - Open http://localhost:5000/swagger |
| 55 | + - Execute posts using contents of **customers.json**. |
| 56 | + - Copy post response, modify fields, then execute puts. |
| 57 | + - Make sure to copy `etag` value from last response, or you will get a concurrency error. |
| 58 | + - Copy `id` and `etag` values to execute deletes. |
| 59 | + - Execute gets to retrieve customers. |
| 60 | + - View customers database collections using Robo 3T. |
| 61 | +5. Create some orders. |
| 62 | + - Execute posts using contents of **orders.json**. |
| 63 | + - Copy post response, modify fields, then execute puts. |
| 64 | + - Make sure to copy `etag` value from last response, or you will get a concurrency error. |
| 65 | + - Copy `id` and `etag` values to execute deletes. |
| 66 | + - Execute gets to retrieve orders. |
| 67 | + - View orders database collections using Robo 3T. |
| 68 | +6. Update the address of a customer who has order. |
| 69 | + - Note the address is also updated for the customer's orders. |
| 70 | + - Observe log messages in terminal when integration events are published and handled. |
| 71 | +7. To **debug** services, you will need to use **Visual Studio Code** with the *Dapr extension*. |
| 72 | + - Open instances of VS Code at both CustomerService and OrderService. |
| 73 | + - If .vscode folder not present: |
| 74 | + - First create build and debug artifacts. |
| 75 | + - Then from the task palette run `Dapr: Scaffold Dapr Tasks`. |
| 76 | + - Enter values for launch, app id (customer-service) and port (5000). |
| 77 | + - Enter values for launch, app id (order-service) and port (5050). |
| 78 | + - Switch to the Debug tab and select "with Dapr" configuration. |
| 79 | + - Set breakpoints as needed and press F5 to start debugging. |
| 80 | +
|
| 81 | +### Usage: EventDriven.CQRS.Abstractions |
| 82 | +
|
| 83 | +> This section describes how to build the Customer and Order services from scratch using the **EventDriven.CQRS.Abstractions** package. For your own project substitute `Customer` and `Order` for your own aggregate entites and related classes. |
| 84 | +
|
| 85 | +1. Add **Domain** and **CustomerAggregate** folders to the project, then add a `Customer` class that extends `Entity`. |
| 86 | + - Add properties representing entity state. |
| 87 | + - Create commands that are C# records and extend a `Command` base class. |
| 88 | + ```csharp |
| 89 | + public record CreateCustomer(Customer Customer) : Command.Create(Customer.Id); |
| 90 | + ``` |
| 91 | + - Create domain events that extend `DomainEvent`. |
| 92 | + ```csharp |
| 93 | + public record CustomerCreated(Customer Customer) : DomainEvent(Customer.Id); |
| 94 | + ``` |
| 95 | + - Where you need to execute business logic, implement `ICommandProcessor` and `IEventApplier` interfaces to process commands by emitting domain events and to apply those events to mutate entity state. |
| 96 | + ```csharp |
| 97 | + public IEnumerable<IDomainEvent> Process(CreateCustomer command) |
| 98 | + // To process command, return one or more domain events |
| 99 | + => new List<IDomainEvent> |
| 100 | + { |
| 101 | + new CustomerCreated(command.Customer) |
| 102 | + }; |
| 103 | +
|
| 104 | + public void Apply(CustomerCreated domainEvent) => |
| 105 | + // Set Id |
| 106 | + Id = domainEvent.EntityId != default(Guid) ? domainEvent.EntityId : Guid.NewGuid(); |
| 107 | + ``` |
| 108 | +2. Add a `CustomerCommandHandler` class that implements `ICommandHandler` for create, update and remove commands. |
| 109 | + - Inject `ICustomerRepository`, `IEventBus` and `IMapper` into the ctor. |
| 110 | + - In the handler for `CreateCustomer`, write code to process the command, apply events, and persist the entity. |
| 111 | + ```csharp |
| 112 | + public async Task<CommandResult<Customer>> Handle(CreateCustomer command) |
| 113 | + { |
| 114 | + // Process command |
| 115 | + _logger.LogInformation("Handling command: {commandName}", nameof(CreateCustomer)); |
| 116 | + var events = command.Customer.Process(command); |
| 117 | + |
| 118 | + // Apply events |
| 119 | + var domainEvent = events.OfType<CustomerCreated>().SingleOrDefault(); |
| 120 | + if (domainEvent == null) return new CommandResult<Customer>(CommandOutcome.NotHandled); |
| 121 | + command.Customer.Apply(domainEvent); |
| 122 | + |
| 123 | + // Persist entity |
| 124 | + var entity = await _repository.Add(command.Customer); |
| 125 | + if (entity == null) return new CommandResult<Customer>(CommandOutcome.InvalidCommand); |
| 126 | + return new CommandResult<Customer>(CommandOutcome.Accepted, entity); |
| 127 | + } |
| 128 | + ``` |
| 129 | + - Create a **Common** class library project and add the package **EventDriven.CQRS.Abstractions**. |
| 130 | + - Reference the Common project from the CustomerService project. |
| 131 | + - Create a `CustomerAddressUpdated` record that extends `IntegrationEvent`. |
| 132 | + ```csharp |
| 133 | + public record CustomerAddressUpdated(Guid CustomerId, Address ShippingAddress) : IntegrationEvent; |
| 134 | + ``` |
| 135 | + - In the `UpdateCustomer` handler, see if the shipping address has changed, and if so, publish a `CustomerAddressUpdated` integration event, so that the order service can update the shipping address in the customer's orders. |
| 136 | + ```csharp |
| 137 | + public async Task<CommandResult<Customer>> Handle(UpdateCustomer command) |
| 138 | + { |
| 139 | + // Compare shipping addresses |
| 140 | + _logger.LogInformation("Handling command: {commandName}", nameof(UpdateCustomer)); |
| 141 | + var existing = await _repository.Get(command.EntityId); |
| 142 | + var addressChanged = command.Customer.ShippingAddress != existing.ShippingAddress; |
| 143 | + |
| 144 | + try |
| 145 | + { |
| 146 | + // Persist entity |
| 147 | + var entity = await _repository.Update(command.Customer); |
| 148 | + if (entity == null) return new CommandResult<Customer>(CommandOutcome.NotFound); |
| 149 | + |
| 150 | + // Publish events |
| 151 | + if (addressChanged) |
| 152 | + { |
| 153 | + var shippingAddress = _mapper.Map<Integration.Models.Address>(entity.ShippingAddress); |
| 154 | + _logger.LogInformation("Publishing event: {eventName}", $"v1.{nameof(CustomerAddressUpdated)}"); |
| 155 | + await _eventBus.PublishAsync( |
| 156 | + new CustomerAddressUpdated(entity.Id, shippingAddress), |
| 157 | + null, "v1"); |
| 158 | + } |
| 159 | + return new CommandResult<Customer>(CommandOutcome.Accepted, entity); |
| 160 | + } |
| 161 | + catch (ConcurrencyException) |
| 162 | + { |
| 163 | + return new CommandResult<Customer>(CommandOutcome.Conflict); |
| 164 | + } |
| 165 | + } |
| 166 | + ``` |
| 167 | +3. Add a `CustomerCommandController` to the project that injects `CustomerCommandHandler` into the ctor. |
| 168 | + - Add Post, Put and Delete actions which accept a `Customer` DTO, map it to a `Customer` entity and invoke the appropriate command handler. |
| 169 | +4. Add a `CustomerQueryController` to the project that injects a `ICustomerRepository` into the ctor. |
| 170 | + - Use the repository to retrieve entities, then map those to `Customer` DTO objects. |
| 171 | +5. Repeat these steps for the Order service. |
| 172 | + - Reference the Common project. |
| 173 | + - Add **Integration/EventHandlers** folders with a `CustomerAddressUpdatedEventHandler` class that extends `IntegrationEventHandler<CustomerAddressUpdated>`. |
| 174 | + - Override `HandleAsync` to update the order addresses for the customer. |
| 175 | + ```csharp |
| 176 | + public override async Task HandleAsync(CustomerAddressUpdated @event) |
| 177 | + { |
| 178 | + _logger.LogInformation("Handling CustomerAddressUpdated event."); |
| 179 | + var orders = await _orderRepository.GetCustomerOrders(@event.CustomerId); |
| 180 | + foreach (var order in orders) |
| 181 | + { |
| 182 | + var shippingAddress = _mapper.Map<Address>(@event.ShippingAddress); |
| 183 | + await _orderRepository.UpdateOrderAddress(order.Id, shippingAddress); |
| 184 | + } |
| 185 | + } |
| 186 | + ``` |
| 187 | + - In `Startup.ConfigureServices` register `CustomerAddressUpdatedEventHandler` then add the Dapr Event Bus. |
| 188 | + ```csharp |
| 189 | + services.AddSingleton<CustomerAddressUpdatedEventHandler>(); |
| 190 | + services.AddDaprEventBus(Constants.DaprPubSubName); |
| 191 | + ``` |
| 192 | + - In `Startup.Configure` use Cloud Events, map subscribe handlers, and map Dapr Event Bus endpoints, subscribing with the event handler. |
| 193 | + ```csharp |
| 194 | + // Use cloud events |
| 195 | + app.UseCloudEvents(); |
| 196 | + app.UseEndpoints(endpoints => |
| 197 | + { |
| 198 | + endpoints.MapControllers(); |
| 199 | + |
| 200 | + // Map subscribe handlers |
| 201 | + endpoints.MapSubscribeHandler(); |
| 202 | + endpoints.MapDaprEventBus(eventBus => |
| 203 | + { |
| 204 | + // Subscribe with event handler |
| 205 | + eventBus.Subscribe(customerAddressUpdatedEventHandler, null, "v1"); |
| 206 | + }); |
| 207 | + }); |
| 208 | + ``` |
0 commit comments