Command Query Responsibility Segregation Separates Reading From Writing... but How?

The first time this term appeared in Software Engineering lexicons was all the way back in 1997. It appeared in the book “Object-Oriented Software Construction” by Bertrand Meyer. At this time it was known as CQS (Command Query Separation). CQRS was finally introduced by Greg Young in 2010. This software architecture is designed to mitigate known caveats of Object-Oriented architecture. Specifically in this case, these caveats are: Writes and reads generate, almost all the time, a very different load against the system. Writing operations are often much more complicated than reading and they affect different scopes of the application. Write operations need to guarantee that correct, validated and consistent data reaches the store correctly. Security concerns are also different between write and read operations. The way data is normally stored in the database is in a 3NF or in another optimized form. When we read data, we often use it to build views to provide the user with readable data. The data presented to the user is hardly ever normalized. This is also referred to as denormalized data. In other words, we present the data to the user with the only concern that the user can read it. Through this very simple list, you can see that reading and writing operate in different ways, they have different concerns, they generate different performance concerns, and they generate different loads which can put a lot of strain on the system in very different ways. CQRS is also a form of DDD(Domain Driven Design). It wasn’t initially thought out to be an actual DDD. Every application has at least one bounded context. Bounded contexts are difficult to define but essentially, they isolate a responsibility of the application like for example, handing of debit cards, library book archiver, and patient data. The latter for example can be divided into multiple subdomains. There could be a separate domain to keep track of chronic illnesses like HIV and another different one to keep track of the common flu. Both of them have different data concerns. Being a chronic disease, HIV patients will need to keep track of a lot more data like T-Cell count, virus load, and other blood data for a lifetime. Patients with the flu don’t need so much monitoring. There are a lot more privacy concerns related to the first domain than the latter. Assessing domains is a bit of an art and it requires the analytical skills of the engineer to determine them.Once you have defined your domain, it’s time to start the design of CQRS for it. As you may have already seen, this design has as its main concern, the separation of reading operations and writing operations. Write operations cannot be read operations. In the same way, “read” operations also cannot be “write” operations. We will look into the source code I’m providing on GitHub So let's start! Commands Commands are defined as any operation that can mutate the data without returning a value. In essence, these all write operations. In CRUD terms, these are the Creation, Update, and Delete operations. You may also refer to them as CUD. The Read operation is implicitly not a part of the operations we need for the Command side of our application. Queries Queries are defined as any operation that will never mutate data and will always return values. Ultimately these are all read operations. Query operations are only read operations. They are only the R in CRUD terms. Models There are many ways to implement CQRS. The point is always to keep read operations apart from write operations as much as possible. In our implementation, we are also going to separate operations and use Event Sourcing. This will allow us to further separate the medium where we are going to keep our data. We will use two different databases. One database will be a part of the command flows and the other database will be part of the read flows. Implementation Let’s first have a look at how all the moving parts will work: In this example, I’m going to try to make this as simple as possible. There are much more elaborated options out there. There are more complicated, dynamic, and scalable options out there. One of these options would be to use RabbitMQ or any other kind of message queueing system to further decouple all components. However, may lead the attention out of the scope of this tutorial. The point here is to present a solution with all the fundamental points of CQRS at its core.We are going to need a lot of dependencies, but for our example, we will focus on the Axon dependencies. You can find a complete listing of all the dependencies used in this example in the root pom.xml: 4.0.0 ... ... org.axonframework axon-spring-boot-starter ${axon.version} org.axonframework axon-server-connector org.axonframework.extensions.mongo axon-mongo ${axon-mongo.version} org.projectlombok lombok ${lombok.version} true ... ... Notice that I am using the axon framework. This is one of the most popular frameworks to implement the CQRS design. Namely, we are going to see how EventHandlers and Aggregators work, how the EventBus works, and how the CommandBus works. We are also going to see how this works with MongoDB in order to, ultimately, get our database updated with new data. Core In this module, I’m going to consider everything that would be common to the application. I could also have named this module as common. For our application to run, we are going to need to consider a few important things. Given its complexity, I am only going to implement a read-all operation and a save operation. These will be my query and my command respectively. We’ll need a DTO to get our data into our system: public record VideoSeriesDto( String name, Integer volumes, BigDecimal cashValue, Genre genre ) { @Builder public VideoSeriesDto { } } Sending data via a writer is an operation that needs to be understood by the reader and also by the writer. Our only common command should be located here: @Builder public class AddSeriesEvent { String id; String name; Integer volumes; BigDecimal cashValue; Genre genre; ...( getters and setters ommitted) } Finally, we know from the schema, that both the “write service” and the “read service” will need to have access to the EventStore. This essentially is, at the end of the tail, our MongoDB database. Axon has very nice out-of-the-box libraries that allow us to easily implement this Event Sourcing mechanism. this is part of the reason why I chose this. It makes for a very simple form of implementation: @Configuration public class AxonConfig {

@Value("${spring.data.mongodb.host:127.0.0.1}") private String mongoHost;

@Value("${spring.data.mongodb.port:27017}") private int mongoPort;

@Value("${spring.data.mongodb.database:test}") private String mongoDatabase;

@Bean public TokenStore tokenStore(Serializer serializer) { return MongoTokenStore.builder().mongoTemplate(axonMongoTemplate()).serializer(serializer).build(); }

@Bean public EventStorageEngine eventStorageEngine(MongoClient client) { return MongoEventStorageEngine.builder().mongoTemplate(DefaultMongoTemplate.builder().mongoDatabase(client).build()).build(); }

@Bean public MongoTemplate axonMongoTemplate() { return DefaultMongoTemplate.builder().mongoDatabase(mongo(), mongoDatabase).build(); }

@Bean public MongoClient mongo() { return MongoClients.create( MongoClientSettings.builder() .applyToClusterSettings(builder -> builder.hosts(List.of( new ServerAddress(mongoHost, mongoPort)))) .build()); } } Command Service First, we need to implement a representation of our command. In the case of our command service, we only have a command to add further video series. Therefore, our command has the same properties as the actual series. Note the id field: @Data @Builder @EqualsAndHashCode @ToString public class AddVideoSeriesCommand {

@TargetAggregateIdentifier private String id;

private String name;

private Integer volumes;

private BigDecimal cashValue;

private Genre genre;

} The id field is indeed a String. This is essentially our operation ID. It can be implemented in several ways. We just need to make sure that it is always a unique string, number, or whatever we choose.Now it’s time to implement the aggregator which will send our command through the command bus and make it reach our command handler: @Slf4j @NoArgsConstructor @Aggregate @Data public class VideoSeriesAggregate {

@AggregateIdentifier private String id;

@CommandHandler public VideoSeriesAggregate(AddVideoSeriesCommand command) { apply(AddSeriesEvent.builder() .id(UUID.randomUUID().toString()) .cashValue(command.getCashValue()) .genre(command.getGenre()) .name(command.getName()) .volumes(command.getVolumes()).build() ); }

@EventSourcingHandler public void on(AddSeriesEvent event) { this.id = event.getId(); }

} Notice the EventSourcingHandler. It doesn’t seem to be doing much, but remember that in this code section you are looking at the contents of the Aggregate element. If you look at the mongo database you will find something like this: { "_id" : ObjectId("5df8ac587a0bba4960afce68"), "aggregateIdentifier" : "ed313d16-8d94-480a-85a0-b6897bcca4f5", "type" : "SeriesAggregate", "sequenceNumber" : NumberLong(0), "serializedPayload" : " ed313d16-8d94-480a-85a0-b6897bcca4f5 wosssda 10 123.2 woo ", "timestamp" : "2019-12-17T10:22:16.640261Z", "payloadType" : "org.jesperancinha.video.core.events.AddSeriesEvent", "payloadRevision" : null, "serializedMetaData" : " traceId 398a250f-8086-40e7-a767-1aa793231f62 correlationId 398a250f-8086-40e7-a767-1aa793231f62 ", "eventIdentifier" : "2ac1a49f-0124-4f6e-b13f-140c8f36979a" } Notice the aggregateIndentifier. This is our id. You need the EventSourcingHandler in order to complete the request and have your Event sourced to the EventStore.Now we only need to complete our application by implementing a Controller: @RestController @RequestMapping("/video-series") public class VideoSeriesController {

private final CommandGateway commandGateway;

public VideoSeriesController(CommandGateway commandGateway) { this.commandGateway = commandGateway; }

@PostMapping public void postNewVideoSeries( @RequestBody VideoSeriesDto videoSeriesDto) { commandGateway.send( AddVideoSeriesCommand.builder() .name(videoSeriesDto.name()) .volumes(videoSeriesDto.volumes()) .genre(videoSeriesDto.genre()) .cashValue(videoSeriesDto.cashValue()) .build()); } } Notice that we are injecting a CommandGateway. This is precisely the gateway that allows us to send commands into our system.Finally, the Spring Boot Launcher: @SpringBootApplication @Import(AxonConfig.class) public class VideoAppCommandLauncher { public static void main(String[] args) { SpringApplication.run(VideoAppCommandLauncher.class); } } To complete our application we still need to configure our Spring Boot Launcher: # spring server.port=8080 # mongodb spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 spring.data.mongodb.database=cqrs Query Service The query service is essentially a reader of the EventStore and will act upon it without the user intervention. The query service needs to perform queries. In this way, I implemented a command to do that just that:public class FindAllVideoSeriesQuery { } Notice that this command ended up being just an empty class. That is done on purpose. We do not need parameters to pass through a read all operation, but we do need its representation.Because we are accessing a database and storing records, we now need to implement the Entity responsible for this data: @Getter @Setter @ToString @AllArgsConstructor @NoArgsConstructor @Builder @Entity @Table(name = "VIDEO_SERIES") public class VideoSeries {

@Id @GeneratedValue(strategy = IDENTITY) @Column private Long id;

@Column private String name;

@Column private Integer volumes;

@Column private BigDecimal cashValue;

@Column @Enumerated(EnumType.STRING) private Genre genre;

@Override public boolean equals(Object o) { if (this == o) return true; if (o == null || Hibernate.getClass(this) != Hibernate.getClass(o)) return false; VideoSeries that = (VideoSeries) o;

return Objects.equals(id, that.id); }

@Override public int hashCode() { return Objects.hash(getId(), getName(), getVolumes(), getCashValue(), getGenre()); } } As you may already have guessed, in this implementation we are going to use JPA repositories:public interface VideoSeriesRepository extends JpaRepository { } On the query side, we have EventHandlers which are very similar in shape to the Aggregate. The difference, of course, is that they process immediately once they receive a command event: @Service @ProcessingGroup("video-series") public class VideoSeriesEventHandler {

private final VideoSeriesRepository videoSeriesRepository;

public VideoSeriesEventHandler(VideoSeriesRepository videoSeriesRepository) { this.videoSeriesRepository = videoSeriesRepository; }

@EventHandler public void on(AddSeriesEvent event) { videoSeriesRepository.save(VideoSeries .builder() .name(event.getName()) .volumes(event.getVolumes()) .genre(event.getGenre()) .cashValue(event.getCashValue()) .build()); }

@QueryHandler public List handle(FindAllVideoSeriesQuery query) { return videoSeriesRepository.findAll().stream().map( videoSeries -> VideoSeriesDto.builder() .name(videoSeries.getName()) .volumes(videoSeries.getVolumes()) .cashValue(videoSeries.getCashValue()) .genre(videoSeries.getGenre()) .build()).collect(Collectors.toList()); }

} Notice that instead of CommandHandler, we now have QueryHandler. Instead of EventSourcingHandler we now have EventHandler. There are annotations used to distinguish what happens in the command service and in the query service respectively. Also, the id isn’t there. The id isn’t important because no data will be going to the event store. All the data is handled directly with the JPA repositories. We can now focus our attention on the Controller for the query service controller: @RestController @RequestMapping("/video-series") public class VideoSeriesController {

@Autowired private QueryGateway queryGateway;

@GetMapping public List gertAllVideoSeries() { return queryGateway.query(new FindAllVideoSeriesQuery(), ResponseTypes.multipleInstancesOf(VideoSeriesDto.class)) .join(); } } And finally our Query Spring Boot Launcher: @SpringBootApplication @Import(AxonConfig.class) public class VideoAppQueryLauncher { public static void main(String[] args) { SpringApplication.run(VideoAppQueryLauncher.class); } } To complete our application we need to configure it: # spring server.port=8090 # h2 spring.h2.console.path=/spring-h2-video-series-query-console spring.h2.console.enabled=true # datasource spring.datasource.url=jdbc:h2:file:~/spring-datasource-video-series-query-file;auto_server=true spring.datasource.driver-class-name=org.h2.Driver spring.datasource.username=sa spring.datasource.password=sa spring.sql.init.schema-locations=classpath*:schema-h2.sql spring.sql.init.mode=always # hibernate spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.H2Dialect spring.jpa.hibernate.ddl-auto=none spring.jpa.show-sql=true # mongodb spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 spring.data.mongodb.database=cqrs Give it some structure: drop table if exists VIDEO_SERIES;

create table VIDEO_SERIES ( ID bigint auto_increment primary key not null, NAME varchar(100) not null, VOLUMES int not null, CASH_VALUE decimal not null, GENRE varchar(100) not null ); And finally some data: insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Modern Family', 12, 12.3, 'SITCOM'); insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Six Feet Under', 10, 34.3, 'DRAMA'); insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Queer as Folk', 24, 55.3, 'DRAMA'); We are almost ready to make some tests. We need now consider how are we going to launch our persistence databases. In our example, we mention 2 important ones: MongoDB and H2. In the code, you’ll find also Postgres in other configurations. There are multiple ways to start this application combo. In the default profile, there only needs to be a running MondDB database on port 27017. We can do this by installing mongo or by using containers. This is why you’ll find three profiles on both the command and the query SpringBoot applications: default — Runs with MongoDB and embedded H2 — run with make docker-mongo or with a local running MongoDB local — Runs with MongoDB and Postgres — run with make docker-databases or with local running MongoDB and Postgres databases prod — Runs with MongoDB and Postgres in a common network with docker-compose — Just run make build-docker and the system will be set up for you Check the Readme.md file for further details on how to run this on your machine.What I did for testing is very simple. First I performed a request to see all my current data: curl localhost:8090/video-series [ { "name": "Modern Family", "volumes": 12, "cashValue": 12.3, "genre": "SITCOM" }, { "name": "Six Feet Under", "volumes": 10, "cashValue": 34.3, "genre": "DRAMA" }, { "name": "Queer as Folk", "volumes": 24, "cashValue": 55.3, "genre": "DRAMA" } ] As you can see, we get three series. Let’s add a new one: curl -d '{ "name":"True Blood", "volumes":30, "cashValue": 1323.2, "genre": "Drama"}' -H "Content-Type: application/json" -X POST http://localhost:8080/video-series You should now see: curl localhost:8090/video-series [ { "name": "Modern Family", "volumes": 12, "cashValue": 12.3, "genre": "SITCOM" }, { "name": "Six Feet Under", "volumes": 10, "cashValue": 34.3, "genre": "DRAMA" }, { "name": "Queer as Folk", "volumes": 24, "cashValue": 55.3, "genre": "DRAMA" }, { "name": "True Blood", "volumes": 30, "cashValue": 1323.2, "genre": "Bloody" } ] Conclusion Note that although we can see that this works, it’s very important that you understand what happened behind the curtain for this application. The separation between the “write” and “read”, referred to as segregation of command and query is what makes the foundations of this architecture. The more decoupled the architecture is designed, the better it is. There are thousands of corner cases and special situations in the landscape of DDD and CQRS. Event sourcing is just one of the ways to get this implemented. In our example, we used Spring, SpringBoot, and Axon to get our commands and events across our network. We didn’t use any messaging queuing system. I do intend to write another article on that, but that will be for later. For the time being, I hope you have enjoyed this tutorial about this very simple example.You’ve probably noticed that I’m not using Java records everywhere. This is because in some instances I could not get the Axon converters to work with records, namely with the AddSeriesEvent. This issue is being resolved. Have a look at the issues document for more details on this.Have a look at the unit and integration tests. There I’ve used Kotlin, Kotest, Testcontainers, and Mockk, which are modern alternatives for Java, Jupiter 5, and Embedded databases. There is an example that goes through the whole architecture and tests the flow from issuing a command to handling the event, to storing and querying it back from the database. I have placed all the source code of this application on GitHubI hope that you have enjoyed this article as much as I enjoyed writing it.Thanks in advance for your help, and thank you for reading! References Working with Postgres Audit Triggers Apply simplified CQRS and DDD patterns in a microservice Platforms JVM Records Mongo Axon Reference Guide CQRS Microservice Sampler by Ben Wilcock on GitHub Patrick Gillespie's Text to Ascii Art Generator Microservices With Spring Boot, Axon CQRS/ES, and Docker by Ben Wilcock Advanced Message Queuing Protocol Install MongoDB on Mac OS Object-Oriented Software Construction by Bertrand Meyer CQRS Journey by Microsoft The first time this term appeared in Software Engineering lexicons was all the way back in 1997 . It appeared in the book “Object-Oriented Software Construction” by Bertrand Meyer . At this time it was known as CQS (Command Query Separation). CQRS was finally introduced by Greg Young in 2010 . This software architecture is designed to mitigate known caveats of Object-Oriented architecture . Specifically in this case, these caveats are: 1997 Bertrand Meyer CQS CQRS Greg Young 2010 Object-Oriented architecture Writes and reads generate, almost all the time, a very different load against the system. Writing operations are often much more complicated than reading and they affect different scopes of the application. Write operations need to guarantee that correct, validated and consistent data reaches the store correctly. Security concerns are also different between write and read operations. The way data is normally stored in the database is in a 3NF or in another optimized form. When we read data, we often use it to build views to provide the user with readable data. The data presented to the user is hardly ever normalized. This is also referred to as denormalized data. In other words, we present the data to the user with the only concern that the user can read it. Writes and reads generate, almost all the time, a very different load against the system. Writing operations are often much more complicated than reading and they affect different scopes of the application. Write operations need to guarantee that correct, validated and consistent data reaches the store correctly. Security concerns are also different between write and read operations. The way data is normally stored in the database is in a 3NF or in another optimized form. When we read data, we often use it to build views to provide the user with readable data. The data presented to the user is hardly ever normalized. This is also referred to as denormalized data. In other words, we present the data to the user with the only concern that the user can read it. Through this very simple list, you can see that reading and writing operate in different ways, they have different concerns, they generate different performance concerns, and they generate different loads which can put a lot of strain on the system in very different ways. CQRS is also a form of DDD (Domain Driven Design). It wasn’t initially thought out to be an actual DDD . Every application has at least one bounded context. Bounded contexts are difficult to define but essentially, they isolate a responsibility of the application like for example, handing of debit cards, library book archiver, and patient data. The latter for example can be divided into multiple subdomains. There could be a separate domain to keep track of chronic illnesses like HIV and another different one to keep track of the common flu. Both of them have different data concerns. Being a chronic disease, HIV patients will need to keep track of a lot more data like T-Cell count, virus load, and other blood data for a lifetime. Patients with the flu don’t need so much monitoring. There are a lot more privacy concerns related to the first domain than the latter. Assessing domains is a bit of an art and it requires the analytical skills of the engineer to determine them.Once you have defined your domain, it’s time to start the design of CQRS for it. As you may have already seen, this design has as its main concern, the separation of reading operations and writing operations. Write operations cannot be read operations. In the same way, “read” operations also cannot be “write” operations. CQRS DDD DDD We will look into the source code I’m providing on GitHub So let's start! GitHub Commands Commands are defined as any operation that can mutate the data without returning a value. In essence, these all write operations. In CRUD terms, these are the Creation, Update, and Delete operations. You may also refer to them as CUD. The Read operation is implicitly not a part of the operations we need for the Command side of our application. Queries Queries are defined as any operation that will never mutate data and will always return values. Ultimately these are all read operations. Query operations are only read operations. They are only the R in CRUD terms. Models There are many ways to implement CQRS. The point is always to keep read operations apart from write operations as much as possible. In our implementation, we are also going to separate operations and use Event Sourcing. This will allow us to further separate the medium where we are going to keep our data. We will use two different databases. One database will be a part of the command flows and the other database will be part of the read flows. Implementation Let’s first have a look at how all the moving parts will work: In this example, I’m going to try to make this as simple as possible. There are much more elaborated options out there. There are more complicated, dynamic, and scalable options out there. One of these options would be to use RabbitMQ or any other kind of message queueing system to further decouple all components. However, may lead the attention out of the scope of this tutorial. The point here is to present a solution with all the fundamental points of CQRS at its core.We are going to need a lot of dependencies, but for our example, we will focus on the Axon dependencies. You can find a complete listing of all the dependencies used in this example in the root pom.xml: 4.0.0 ... ... org.axonframework axon-spring-boot-starter ${axon.version} org.axonframework axon-server-connector org.axonframework.extensions.mongo axon-mongo ${axon-mongo.version} org.projectlombok lombok ${lombok.version} true ... ... 4.0.0 ... ... org.axonframework axon-spring-boot-starter ${axon.version} org.axonframework axon-server-connector org.axonframework.extensions.mongo axon-mongo ${axon-mongo.version} org.projectlombok lombok ${lombok.version} true ... ... Notice that I am using the axon framework. This is one of the most popular frameworks to implement the CQRS design. Namely, we are going to see how EventHandlers and Aggregators work, how the EventBus works, and how the CommandBus works. We are also going to see how this works with MongoDB in order to, ultimately, get our database updated with new data. Core In this module, I’m going to consider everything that would be common to the application. I could also have named this module as common. For our application to run, we are going to need to consider a few important things. Given its complexity, I am only going to implement a read-all operation and a save operation. These will be my query and my command respectively. We’ll need a DTO to get our data into our system: public record VideoSeriesDto( String name, Integer volumes, BigDecimal cashValue, Genre genre ) { @Builder public VideoSeriesDto { } } public record VideoSeriesDto( String name, Integer volumes, BigDecimal cashValue, Genre genre ) { @Builder public VideoSeriesDto { } } Sending data via a writer is an operation that needs to be understood by the reader and also by the writer. Our only common command should be located here: @Builder public class AddSeriesEvent { String id; String name; Integer volumes; BigDecimal cashValue; Genre genre; ...( getters and setters ommitted) } @Builder public class AddSeriesEvent { String id; String name; Integer volumes; BigDecimal cashValue; Genre genre; ...( getters and setters ommitted) } Finally, we know from the schema, that both the “write service” and the “read service” will need to have access to the EventStore. This essentially is, at the end of the tail, our MongoDB database. Axon has very nice out-of-the-box libraries that allow us to easily implement this Event Sourcing mechanism. this is part of the reason why I chose this. It makes for a very simple form of implementation: @Configuration public class AxonConfig {

@Value("${spring.data.mongodb.host:127.0.0.1}") private String mongoHost;

@Value("${spring.data.mongodb.port:27017}") private int mongoPort;

@Value("${spring.data.mongodb.database:test}") private String mongoDatabase;

@Bean public TokenStore tokenStore(Serializer serializer) { return MongoTokenStore.builder().mongoTemplate(axonMongoTemplate()).serializer(serializer).build(); }

@Bean public EventStorageEngine eventStorageEngine(MongoClient client) { return MongoEventStorageEngine.builder().mongoTemplate(DefaultMongoTemplate.builder().mongoDatabase(client).build()).build(); }

@Bean public MongoTemplate axonMongoTemplate() { return DefaultMongoTemplate.builder().mongoDatabase(mongo(), mongoDatabase).build(); }

@Bean public MongoClient mongo() { return MongoClients.create( MongoClientSettings.builder() .applyToClusterSettings(builder -> builder.hosts(List.of( new ServerAddress(mongoHost, mongoPort)))) .build()); } } @Configuration public class AxonConfig {

@Value("${spring.data.mongodb.host:127.0.0.1}") private String mongoHost;

@Value("${spring.data.mongodb.port:27017}") private int mongoPort;

@Value("${spring.data.mongodb.database:test}") private String mongoDatabase;

@Bean public TokenStore tokenStore(Serializer serializer) { return MongoTokenStore.builder().mongoTemplate(axonMongoTemplate()).serializer(serializer).build(); }

@Bean public EventStorageEngine eventStorageEngine(MongoClient client) { return MongoEventStorageEngine.builder().mongoTemplate(DefaultMongoTemplate.builder().mongoDatabase(client).build()).build(); }

@Bean public MongoTemplate axonMongoTemplate() { return DefaultMongoTemplate.builder().mongoDatabase(mongo(), mongoDatabase).build(); }

@Bean public MongoClient mongo() { return MongoClients.create( MongoClientSettings.builder() .applyToClusterSettings(builder -> builder.hosts(List.of( new ServerAddress(mongoHost, mongoPort)))) .build()); } } Command Service First, we need to implement a representation of our command. In the case of our command service, we only have a command to add further video series. Therefore, our command has the same properties as the actual series. Note the id field: @Data @Builder @EqualsAndHashCode @ToString public class AddVideoSeriesCommand {

@TargetAggregateIdentifier private String id;

private String name;

private Integer volumes;

private BigDecimal cashValue;

private Genre genre;

} @Data @Builder @EqualsAndHashCode @ToString public class AddVideoSeriesCommand {

@TargetAggregateIdentifier private String id;

private String name;

private Integer volumes;

private BigDecimal cashValue;

private Genre genre;

} The id field is indeed a String. This is essentially our operation ID. It can be implemented in several ways. We just need to make sure that it is always a unique string, number, or whatever we choose.Now it’s time to implement the aggregator which will send our command through the command bus and make it reach our command handler: @Slf4j @NoArgsConstructor @Aggregate @Data public class VideoSeriesAggregate {

@AggregateIdentifier private String id;

@CommandHandler public VideoSeriesAggregate(AddVideoSeriesCommand command) { apply(AddSeriesEvent.builder() .id(UUID.randomUUID().toString()) .cashValue(command.getCashValue()) .genre(command.getGenre()) .name(command.getName()) .volumes(command.getVolumes()).build() ); }

@EventSourcingHandler public void on(AddSeriesEvent event) { this.id = event.getId(); }

} @Slf4j @NoArgsConstructor @Aggregate @Data public class VideoSeriesAggregate {

@AggregateIdentifier private String id;

@CommandHandler public VideoSeriesAggregate(AddVideoSeriesCommand command) { apply(AddSeriesEvent.builder() .id(UUID.randomUUID().toString()) .cashValue(command.getCashValue()) .genre(command.getGenre()) .name(command.getName()) .volumes(command.getVolumes()).build() ); }

@EventSourcingHandler public void on(AddSeriesEvent event) { this.id = event.getId(); }

} Notice the EventSourcingHandler. It doesn’t seem to be doing much, but remember that in this code section you are looking at the contents of the Aggregate element. If you look at the mongo database you will find something like this: { "_id" : ObjectId("5df8ac587a0bba4960afce68"), "aggregateIdentifier" : "ed313d16-8d94-480a-85a0-b6897bcca4f5", "type" : "SeriesAggregate", "sequenceNumber" : NumberLong(0), "serializedPayload" : " ed313d16-8d94-480a-85a0-b6897bcca4f5 wosssda 10 123.2 woo ", "timestamp" : "2019-12-17T10:22:16.640261Z", "payloadType" : "org.jesperancinha.video.core.events.AddSeriesEvent", "payloadRevision" : null, "serializedMetaData" : " traceId 398a250f-8086-40e7-a767-1aa793231f62 correlationId 398a250f-8086-40e7-a767-1aa793231f62 ", "eventIdentifier" : "2ac1a49f-0124-4f6e-b13f-140c8f36979a" } { "_id" : ObjectId("5df8ac587a0bba4960afce68"), "aggregateIdentifier" : "ed313d16-8d94-480a-85a0-b6897bcca4f5", "type" : "SeriesAggregate", "sequenceNumber" : NumberLong(0), "serializedPayload" : " ed313d16-8d94-480a-85a0-b6897bcca4f5 wosssda 10 123.2 woo ", "timestamp" : "2019-12-17T10:22:16.640261Z", "payloadType" : "org.jesperancinha.video.core.events.AddSeriesEvent", "payloadRevision" : null, "serializedMetaData" : " traceId 398a250f-8086-40e7-a767-1aa793231f62 correlationId 398a250f-8086-40e7-a767-1aa793231f62 ", "eventIdentifier" : "2ac1a49f-0124-4f6e-b13f-140c8f36979a" } Notice the aggregateIndentifier. This is our id. You need the EventSourcingHandler in order to complete the request and have your Event sourced to the EventStore.Now we only need to complete our application by implementing a Controller: @RestController @RequestMapping("/video-series") public class VideoSeriesController {

private final CommandGateway commandGateway;

public VideoSeriesController(CommandGateway commandGateway) { this.commandGateway = commandGateway; }

@PostMapping public void postNewVideoSeries( @RequestBody VideoSeriesDto videoSeriesDto) { commandGateway.send( AddVideoSeriesCommand.builder() .name(videoSeriesDto.name()) .volumes(videoSeriesDto.volumes()) .genre(videoSeriesDto.genre()) .cashValue(videoSeriesDto.cashValue()) .build()); } } @RestController @RequestMapping("/video-series") public class VideoSeriesController {

private final CommandGateway commandGateway;

public VideoSeriesController(CommandGateway commandGateway) { this.commandGateway = commandGateway; }

@PostMapping public void postNewVideoSeries( @RequestBody VideoSeriesDto videoSeriesDto) { commandGateway.send( AddVideoSeriesCommand.builder() .name(videoSeriesDto.name()) .volumes(videoSeriesDto.volumes()) .genre(videoSeriesDto.genre()) .cashValue(videoSeriesDto.cashValue()) .build()); } } Notice that we are injecting a CommandGateway. This is precisely the gateway that allows us to send commands into our system.Finally, the Spring Boot Launcher: @SpringBootApplication @Import(AxonConfig.class) public class VideoAppCommandLauncher { public static void main(String[] args) { SpringApplication.run(VideoAppCommandLauncher.class); } } @SpringBootApplication @Import(AxonConfig.class) public class VideoAppCommandLauncher { public static void main(String[] args) { SpringApplication.run(VideoAppCommandLauncher.class); } } To complete our application we still need to configure our Spring Boot Launcher: # spring server.port=8080 # mongodb spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 spring.data.mongodb.database=cqrs # spring server.port=8080 # mongodb spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 spring.data.mongodb.database=cqrs Query Service The query service is essentially a reader of the EventStore and will act upon it without the user intervention. The query service needs to perform queries. In this way, I implemented a command to do that just that:public class FindAllVideoSeriesQuery { } Notice that this command ended up being just an empty class. That is done on purpose. We do not need parameters to pass through a read all operation, but we do need its representation.Because we are accessing a database and storing records, we now need to implement the Entity responsible for this data: @Getter @Setter @ToString @AllArgsConstructor @NoArgsConstructor @Builder @Entity @Table(name = "VIDEO_SERIES") public class VideoSeries {

@Id @GeneratedValue(strategy = IDENTITY) @Column private Long id;

@Column private String name;

@Column private Integer volumes;

@Column private BigDecimal cashValue;

@Column @Enumerated(EnumType.STRING) private Genre genre;

@Override public boolean equals(Object o) { if (this == o) return true; if (o == null || Hibernate.getClass(this) != Hibernate.getClass(o)) return false; VideoSeries that = (VideoSeries) o;

return Objects.equals(id, that.id); }

@Override public int hashCode() { return Objects.hash(getId(), getName(), getVolumes(), getCashValue(), getGenre()); } } @Getter @Setter @ToString @AllArgsConstructor @NoArgsConstructor @Builder @Entity @Table(name = "VIDEO_SERIES") public class VideoSeries {

@Id @GeneratedValue(strategy = IDENTITY) @Column private Long id;

@Column private String name;

@Column private Integer volumes;

@Column private BigDecimal cashValue;

@Column @Enumerated(EnumType.STRING) private Genre genre;

@Override public boolean equals(Object o) { if (this == o) return true; if (o == null || Hibernate.getClass(this) != Hibernate.getClass(o)) return false; VideoSeries that = (VideoSeries) o;

return Objects.equals(id, that.id); }

@Override public int hashCode() { return Objects.hash(getId(), getName(), getVolumes(), getCashValue(), getGenre()); } } As you may already have guessed, in this implementation we are going to use JPA repositories:public interface VideoSeriesRepository extends JpaRepository { } On the query side, we have EventHandlers which are very similar in shape to the Aggregate. The difference, of course, is that they process immediately once they receive a command event: @Service @ProcessingGroup("video-series") public class VideoSeriesEventHandler {

private final VideoSeriesRepository videoSeriesRepository;

public VideoSeriesEventHandler(VideoSeriesRepository videoSeriesRepository) { this.videoSeriesRepository = videoSeriesRepository; }

@EventHandler public void on(AddSeriesEvent event) { videoSeriesRepository.save(VideoSeries .builder() .name(event.getName()) .volumes(event.getVolumes()) .genre(event.getGenre()) .cashValue(event.getCashValue()) .build()); }

@QueryHandler public List handle(FindAllVideoSeriesQuery query) { return videoSeriesRepository.findAll().stream().map( videoSeries -> VideoSeriesDto.builder() .name(videoSeries.getName()) .volumes(videoSeries.getVolumes()) .cashValue(videoSeries.getCashValue()) .genre(videoSeries.getGenre()) .build()).collect(Collectors.toList()); }

} @Service @ProcessingGroup("video-series") public class VideoSeriesEventHandler {

private final VideoSeriesRepository videoSeriesRepository;

public VideoSeriesEventHandler(VideoSeriesRepository videoSeriesRepository) { this.videoSeriesRepository = videoSeriesRepository; }

@EventHandler public void on(AddSeriesEvent event) { videoSeriesRepository.save(VideoSeries .builder() .name(event.getName()) .volumes(event.getVolumes()) .genre(event.getGenre()) .cashValue(event.getCashValue()) .build()); }

@QueryHandler public List handle(FindAllVideoSeriesQuery query) { return videoSeriesRepository.findAll().stream().map( videoSeries -> VideoSeriesDto.builder() .name(videoSeries.getName()) .volumes(videoSeries.getVolumes()) .cashValue(videoSeries.getCashValue()) .genre(videoSeries.getGenre()) .build()).collect(Collectors.toList()); }

} Notice that instead of CommandHandler, we now have QueryHandler. Instead of EventSourcingHandler we now have EventHandler. There are annotations used to distinguish what happens in the command service and in the query service respectively. Also, the id isn’t there. The id isn’t important because no data will be going to the event store. All the data is handled directly with the JPA repositories. We can now focus our attention on the Controller for the query service controller: @RestController @RequestMapping("/video-series") public class VideoSeriesController {

@Autowired private QueryGateway queryGateway;

@GetMapping public List gertAllVideoSeries() { return queryGateway.query(new FindAllVideoSeriesQuery(), ResponseTypes.multipleInstancesOf(VideoSeriesDto.class)) .join(); } } @RestController @RequestMapping("/video-series") public class VideoSeriesController {

@Autowired private QueryGateway queryGateway;

@GetMapping public List gertAllVideoSeries() { return queryGateway.query(new FindAllVideoSeriesQuery(), ResponseTypes.multipleInstancesOf(VideoSeriesDto.class)) .join(); } } And finally our Query Spring Boot Launcher: @SpringBootApplication @Import(AxonConfig.class) public class VideoAppQueryLauncher { public static void main(String[] args) { SpringApplication.run(VideoAppQueryLauncher.class); } } @SpringBootApplication @Import(AxonConfig.class) public class VideoAppQueryLauncher { public static void main(String[] args) { SpringApplication.run(VideoAppQueryLauncher.class); } } To complete our application we need to configure it: # spring server.port=8090 # h2 spring.h2.console.path=/spring-h2-video-series-query-console spring.h2.console.enabled=true # datasource spring.datasource.url=jdbc:h2:file:~/spring-datasource-video-series-query-file;auto_server=true spring.datasource.driver-class-name=org.h2.Driver spring.datasource.username=sa spring.datasource.password=sa spring.sql.init.schema-locations=classpath*:schema-h2.sql spring.sql.init.mode=always # hibernate spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.H2Dialect spring.jpa.hibernate.ddl-auto=none spring.jpa.show-sql=true # mongodb spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 spring.data.mongodb.database=cqrs # spring server.port=8090 # h2 spring.h2.console.path=/spring-h2-video-series-query-console spring.h2.console.enabled=true # datasource spring.datasource.url=jdbc:h2:file:~/spring-datasource-video-series-query-file;auto_server=true spring.datasource.driver-class-name=org.h2.Driver spring.datasource.username=sa spring.datasource.password=sa spring.sql.init.schema-locations=classpath*:schema-h2.sql spring.sql.init.mode=always # hibernate spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.H2Dialect spring.jpa.hibernate.ddl-auto=none spring.jpa.show-sql=true # mongodb spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 spring.data.mongodb.database=cqrs Give it some structure: drop table if exists VIDEO_SERIES;

create table VIDEO_SERIES ( ID bigint auto_increment primary key not null, NAME varchar(100) not null, VOLUMES int not null, CASH_VALUE decimal not null, GENRE varchar(100) not null ); drop table if exists VIDEO_SERIES;

create table VIDEO_SERIES ( ID bigint auto_increment primary key not null, NAME varchar(100) not null, VOLUMES int not null, CASH_VALUE decimal not null, GENRE varchar(100) not null ); And finally some data: insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Modern Family', 12, 12.3, 'SITCOM'); insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Six Feet Under', 10, 34.3, 'DRAMA'); insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Queer as Folk', 24, 55.3, 'DRAMA'); insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Modern Family', 12, 12.3, 'SITCOM'); insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Six Feet Under', 10, 34.3, 'DRAMA'); insert into VIDEO_SERIES (NAME, VOLUMES, CASH_VALUE, GENRE) values ('Queer as Folk', 24, 55.3, 'DRAMA'); We are almost ready to make some tests. We need now consider how are we going to launch our persistence databases. In our example, we mention 2 important ones: MongoDB and H2. In the code, you’ll find also Postgres in other configurations. There are multiple ways to start this application combo. In the default profile, there only needs to be a running MondDB database on port 27017. We can do this by installing mongo or by using containers. This is why you’ll find three profiles on both the command and the query SpringBoot applications: default — Runs with MongoDB and embedded H2 — run with make docker-mongo or with a local running MongoDB local — Runs with MongoDB and Postgres — run with make docker-databases or with local running MongoDB and Postgres databases prod — Runs with MongoDB and Postgres in a common network with docker-compose — Just run make build-docker and the system will be set up for you default — Runs with MongoDB and embedded H2 — run with make docker-mongo or with a local running MongoDB local — Runs with MongoDB and Postgres — run with make docker-databases or with local running MongoDB and Postgres databases prod — Runs with MongoDB and Postgres in a common network with docker-compose — Just run make build-docker and the system will be set up for you Check the Readme.md file for further details on how to run this on your machine.What I did for testing is very simple. First I performed a request to see all my current data: curl localhost:8090/video-series curl localhost:8090/video-series [ { "name": "Modern Family", "volumes": 12, "cashValue": 12.3, "genre": "SITCOM" }, { "name": "Six Feet Under", "volumes": 10, "cashValue": 34.3, "genre": "DRAMA" }, { "name": "Queer as Folk", "volumes": 24, "cashValue": 55.3, "genre": "DRAMA" } ] [ { "name": "Modern Family", "volumes": 12, "cashValue": 12.3, "genre": "SITCOM" }, { "name": "Six Feet Under", "volumes": 10, "cashValue": 34.3, "genre": "DRAMA" }, { "name": "Queer as Folk", "volumes": 24, "cashValue": 55.3, "genre": "DRAMA" } ] As you can see, we get three series. Let’s add a new one: curl -d '{ "name":"True Blood", "volumes":30, "cashValue": 1323.2, "genre": "Drama"}' -H "Content-Type: application/json" -X POST http://localhost:8080/video-series curl -d '{ "name":"True Blood", "volumes":30, "cashValue": 1323.2, "genre": "Drama"}' -H "Content-Type: application/json" -X POST http://localhost:8080/video-series You should now see: curl localhost:8090/video-series curl localhost:8090/video-series [ { "name": "Modern Family", "volumes": 12, "cashValue": 12.3, "genre": "SITCOM" }, { "name": "Six Feet Under", "volumes": 10, "cashValue": 34.3, "genre": "DRAMA" }, { "name": "Queer as Folk", "volumes": 24, "cashValue": 55.3, "genre": "DRAMA" }, { "name": "True Blood", "volumes": 30, "cashValue": 1323.2, "genre": "Bloody" } ] [ { "name": "Modern Family", "volumes": 12, "cashValue": 12.3, "genre": "SITCOM" }, { "name": "Six Feet Under", "volumes": 10, "cashValue": 34.3, "genre": "DRAMA" }, { "name": "Queer as Folk", "volumes": 24, "cashValue": 55.3, "genre": "DRAMA" }, { "name": "True Blood", "volumes": 30, "cashValue": 1323.2, "genre": "Bloody" } ] Conclusion Note that although we can see that this works, it’s very important that you understand what happened behind the curtain for this application. The separation between the “write” and “read”, referred to as segregation of command and query is what makes the foundations of this architecture. The more decoupled the architecture is designed, the better it is. There are thousands of corner cases and special situations in the landscape of DDD and CQRS. Event sourcing is just one of the ways to get this implemented. In our example, we used Spring, SpringBoot, and Axon to get our commands and events across our network. We didn’t use any messaging queuing system. I do intend to write another article on that, but that will be for later. For the time being, I hope you have enjoyed this tutorial about this very simple example.You’ve probably noticed that I’m not using Java records everywhere. This is because in some instances I could not get the Axon converters to work with records, namely with the AddSeriesEvent. This issue is being resolved. Have a look at the issues document for more details on this.Have a look at the unit and integration tests. There I’ve used Kotlin, Kotest, Testcontainers, and Mockk, which are modern alternatives for Java, Jupiter 5, and Embedded databases. There is an example that goes through the whole architecture and tests the flow from issuing a command to handling the event, to storing and querying it back from the database. I have placed all the source code of this application on GitHub I hope that you have enjoyed this article as much as I enjoyed writing it.Thanks in advance for your help, and thank you for reading! GitHub References Working with Postgres Audit Triggers Apply simplified CQRS and DDD patterns in a microservice Platforms JVM Records Mongo Axon Reference Guide CQRS Microservice Sampler by Ben Wilcock on GitHub Patrick Gillespie's Text to Ascii Art Generator Microservices With Spring Boot, Axon CQRS/ES, and Docker by Ben Wilcock Advanced Message Queuing Protocol Install MongoDB on Mac OS Object-Oriented Software Construction by Bertrand Meyer CQRS Journey by Microsoft Working with Postgres Audit Triggers Working with Postgres Audit Triggers Apply simplified CQRS and DDD patterns in a microservice Apply simplified CQRS and DDD patterns in a microservice Platforms JVM Records Platforms JVM Records Mongo Axon Reference Guide Mongo Axon Reference Guide CQRS Microservice Sampler by Ben Wilcock on GitHub CQRS Microservice Sampler by Ben Wilcock on GitHub Patrick Gillespie's Text to Ascii Art Generator Patrick Gillespie's Text to Ascii Art Generator Microservices With Spring Boot, Axon CQRS/ES, and Docker by Ben Wilcock Microservices With Spring Boot, Axon CQRS/ES, and Docker by Ben Wilcock Advanced Message Queuing Protocol Advanced Message Queuing Protocol Install MongoDB on Mac OS Install MongoDB on Mac OS Object-Oriented Software Construction by Bertrand Meyer Object-Oriented Software Construction by Bertrand Meyer CQRS Journey by Microsoft CQRS Journey by Microsoft

Command Query Responsibility Segregation Separates Reading From Writing... but How?

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

A Strange Yucca Tribute Concert: Configuring Kong API Gateway With Micronaut Services in Kotlin

The Ethics Board of Axon Resigned Over Taser-Armed Drones - Then the Company Went Even Further

The Markup Wants to Investigate Companies Like Axon That Expand Their Reach With Public Dollars

ChatGPT Implements Work With Users Using the CQRS Pattern

Choreography — A Vital Tool to Strangle Monoliths

Commands & Events instead of CRUD — Part 1: Commands

A Strange Yucca Tribute Concert: Configuring Kong API Gateway With Micronaut Services in Kotlin

The Ethics Board of Axon Resigned Over Taser-Armed Drones - Then the Company Went Even Further

The Markup Wants to Investigate Companies Like Axon That Expand Their Reach With Public Dollars

ChatGPT Implements Work With Users Using the CQRS Pattern

Choreography — A Vital Tool to Strangle Monoliths

Commands & Events instead of CRUD — Part 1: Commands

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps