How to check if the command is execuable when it delivers through asynchronous message queue?

Question

The aggregate root's state is the product of the commands it receives. So different order of receiving the commands produce different aggregate root states.

How can I check domain rules based on the state when the order of commands are not guaranteed? How to check if a command can be executed based on the last state of the aggregate root if the command is sent by an asynchronous message queue like Rabbitmq?

• Andy: It is quite simple, you either care about the order of commands and then need to implement something like the actor model, or you do not and you just accept the commands in any order they arrive.
• Mohsen: @Andy: The message queue of actor models are local in a process, but what I'm asking is about the integration between the actor model and asynchronous message queues. If you know an alternative sution share please. Thank you.
• Subhash: Do the messages from the message queue contain some state? Will this state be used to determine whether they can be "consumed" by the aggregate?
• Mohsen: @Subhash Bhushan: Yes

Answer 1

different order of receiving the commands produce different aggregate root states.

Yes, in general

How can I check domain rules based on the state when the order of commands are not guaranteed? How to check if a command can be executed based on the last state of the aggregate root if the command is sent by an asynchronous message queue like Rabbitmq?

So the basic rule, from which everything follows, is this: the domain model gets to decide what effect (if any) a message has based on its current state. I read a message, I update myself, I read the next message, I update myself, and so on.

The domain model is fundamentally a state machine - if we process message 1, then message 5, then messages 2 and 3, then message 2 again, then message 6, the the model will be in a deterministic state; repeating the same sequence of messages to a copy of the domain model with the same initial condition will produce the same results.

But those won't, generally, be the same results that you get if the messages were instead to arrive 1, 2, 3, 4, 5, 6....

The way that you check that that the domain model is implemented correctly is by feeding a controlled sequence of events directly to the model. Which is to say, you run the domain model inside of a process with guaranteed delivery order (for example, inside of a unit test).

Note that also allows to you measure what happens when messages are lost, or when messages are duplicated, or are passed in the "wrong" order. You make sure everything works in a localized environment before you try experimenting with it in a distributed environment.

• Mohsen: Do you mean that I should design the domain model in some way that delivery of the commands by any order affect the state of the aggregate root the same way?
• Bart van Ingen Schenau: @Mohsen: No. You should design the domain model such that receiving a command when it can't be processed is not a fatal error. Instead, you might, for example, let the inappropriate command have no effect.
• Mohsen: @Bart van Ingen Schenau: Would you introduce some resource or example to follow? Thank you.

Answer 2

I am going to propose a mechanism from the Functional Domain modeling world. And I am going to offer an example usecase, although it would have been easier if you had given a concrete example. I apologize in advance for putting the code in Python if you are not familiar with the language, but in my defense, it almost looks like pseudo-code :)

Assumptions:

• Aggregate root has a state
• Messages from RabbitMQ contain state and become commands
• Order of commands is not guaranteed
• Domain rules dictate whether a command should be accepted or not
• The reason to control order is to produce a consistent, repeatable state of the aggregate root (Otherwise, it will end up in a different state even with the same set of messages)

Let's consider a Loan Aggregate. Assume that a user applies for a loan, the loan might be enriched with additional documentation, and then either approved or rejected.

Applying your problem statement, a loan can be enriched only if it has been applied, approved only if it has been either applied or enriched, and so on.

Typically, you would model the state as an attribute of the loan, like so: state = STATUS.APPLIED. Then you would have to write rules like:

if command.state == COMMAND.ENRICH and loan.state == STATUS.APPLIED:
 execute_process(...)
else:
 raise(error)

Instead, you could model the state as a concrete type, like below:

class Loan(Object):
 _state_ = STATUS.NONE
 # Lot more attributes
 def apply(command):
 raise NotImplementedError()
 def enrich(command):
 raise NotImplementedError()
 def approve(command):
 raise NotImplementedError()
 def reject(command):
 raise NotImplementedError()
class NewLoan(Loan):
 _state_ = STATUS.NEW
 def apply(command):
 return Loan(...)
class AppliedLoan(Loan):
 _state_ = STATUS.APPLIED
 def enrich(command):
 return Loan(...)
 def approve(command):
 return Loan(...)
class EnrichedLoan(Loan):
 _state_ = STATUS.ENRICHED
 def approve(command):
 return Loan(...)
class ApprovedLoan(Loan):
 _state_ = STATUS.APPROVED
 def reject(command):
 return Loan(...)
class RejectedLoan(Loan):
 _state_ = STATUS.REJECTED

The _state_ class variable is redefined in each child class to the appropriate status. You would then have a factory method that reconstitutes the loan object from datastore:

def find_loan(application_number):
 return loan_repository.get(application_number)
def construct_loan(application_number):
 record = find_loan(application_number)
 if record['state'] == 'applied':
 return AppliedLoan(record)
 elif record['state'] == 'enriched':
 return EnrichedLoan(record)
 ...

The command is then passed to the appropriate process method for execution:

loan_object = construct_loan(command['application_number'])
if command['state'] == 'enrich':
 loan_object.enrich(command)
elif command['state'] == 'approve':
 loan_object.approve(command)
...

Now, system throws a runtime error "NotImplementedError" whenever a command cannot be processed because of state inconsistencies.

You can either ignore them, in which case, you will have a no-op if the wrong command is applied. Or you can bubble up an error, in case this is a situation that is not expected to happen.

You can even store the command in a data store for later processing, in case you expect to receive messages jumbled up, but coherent as a whole.