0

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?

  • 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. – Andy Jul 12 at 16:20
  • @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. – Mohsen Jul 12 at 18:01
  • 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? – Subhash Bhushan Jul 13 at 1:38
  • @Subhash Bhushan: Yes – Mohsen Jul 13 at 11:50
2

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.

  • 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? – Mohsen Jul 12 at 20:54
  • @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. – Bart van Ingen Schenau Jul 14 at 8:03
  • @Bart van Ingen Schenau: Would you introduce some resource or example to follow? Thank you. – Mohsen Jul 18 at 16:55
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.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.