Liking cljdoc? Tell your friends :D
Clojure only.

rill.wheel

Aggregates and Events

Synopsis

(require '[rill.wheel :as aggregate
           :refer [defaggregate defevent]])

(defaggregate user
  "a user is identified by a single `email` property"
  [email])

(defevent registered ::user
  "user was correctly registered"
  [user]
  (assoc user :registered? true))

(defevent unregistered ::user
  "user has unregistered"
  [user]
  (dissoc user :registered?))

(-> (user "user@example.com") registered :registered?)
  => true

(registered-event (user "user@example.com"))
  => {:rill.message/type :user/registered,
      :email "user@example.com",
      :rill.wheel/type :user/user}

(wheel/new-events some-aggreate)
  => seq-of-events

Store and retrieve aggregates in a repository

(-> (get-user repo "user@example.com)
    (registered)
    (command/commit!))
;; ...
(get-user some-repository "user@example.com")

Full example of defaggregate

(defaggregate turnstile
  "An aggregate with docstring"
  [turnstile-id]
  {:pre [(instance? java.util.UUID turnstile-id)]}
  ((installed
    "A turnstile was installed"
    [turnstile]
    (assoc turnstile
           :installed? true
           :locked? true
           :coins 0
           :turns 0
           :pushes 0))

   (coin-inserted
    "A Coin was inserted into the turnstile"
    [turnstile]
    (-> turnstile
        (update :coins inc)
        (assoc :locked? false)))

   (arm-turned
    "The turnstile's arm was turned"
    [turnstile]
    (-> turnstile
        (update :pushes inc)
        (update :turns inc)
        (assoc :locked? true)))

   (arm-pushed-ineffectively
    "The arm was pushed but did not move"
    [turnstile]
    (-> turnstile
        (update :pushes inc))))

  ((install-turnstile
    [repo turnstile-id]
    (let [turnstile (get-turnstile repo turnstile-id)]
      (if (wheel/exists turnstile)
        (rejection turnstile "Already exists")
        (installed turnstile))))

   (insert-coin
    "Insert coin into turnstile, will unlock"
    [repo turnstile-id]
    (let [turnstile (get-turnstile repo turnstile-id)]
      (if (:installed? turnstile)
        (coin-inserted turnstile)
        (rejection turnstile "Turnstile not installed"))))

   (push-arm
    "Push the arm, might turn or be ineffective"
    {::wheel/events [::arm-pushed-ineffectively ::arm-turned]}
    [repo turnstile-id]
    (let [turnstile (get-turnstile repo turnstile-id)]
      (cond
        (not (:installed? turnstile))
        (rejection turnstile "Not installed")
        (:locked? turnstile)
        (arm-pushed-ineffectively turnstile)
        :else
         (arm-turned turnstile))))))

Commands

Commands are functions that apply new events to aggregates.

Command flow

 (-> (get-some-aggregate repository id) ; 1.
     (cmd-call additional-argument)     ; 2.
     (commit!))                         ; 3.

1. Fetch aggregate

Before calling the command, the aggregate it applies to should get fetched from the repository. In rill/wheel, this will always work and must be done even for aggregates that have no events applied to them - this will result in an rill.wheel/empty? aggregate that can be committed later.

2. Calling the command

A command can have any number of arguments, and it's idiomatic for commands to take the aggregate-to-change as the first argument.

As a matter of style, it's suggested that commands do not fetch other objects from the repository but are explicitly passed any necessary aggregates.

Success

A successful command returns an uncommitted? aggregate.

Rejection

A command may be rejected, in which case the command returns a rejection - meaning the request was denied for business reasons. Rejections are explicitly constructed in the defcommand body by the application writer.

It's typically useless to retry a rejected command.

3. Committing results

The result of a command can be persisted back to the repository by calling commit!. If commit! is passed a rejection it will return it. Otherwise the argument should be an aggregate that will be persisted.

ok

A successful commit returns an ok? object describing the committed events and aggregate.

conflict

Committing an updated aggregate can return a conflict, meaning there were changes to the aggregate in the repository in the time between fetching the aggregate and calling commit!.

Depending on the use case, it may be useful to update the aggregate and retry a conflicted command.

Defining commands

 (defaggregate x                                 ; 1.
   [id])

 (defevent x-happened                            ; 2.
    "X happened to obj"
    [obj arg1]
    (assoc obj :a arg1))

 (defcommand do-x ::x                            ; 3.
    "Make X happen to obj"
    [obj arg1]
    (if (= (:a obj) arg1))                       ; 4.
        (rejection obj "Arg already applied")
        (x-happened obj)))                       ; 5.

1. Define the aggregate type

2. Define events to apply

Commands can only affect aggregates by applying events to them. Here we define an event with defevent. When the x-happened event is applied it will set key :a of aggregate obj.

It's idiomatic to give events a past-tense name, to indicate that the event happened and cannot be rejected.

3. Define command

Commands are defined by calling defcommand, specifying a name, aggregate type, optional docstring, argument vector and a command body.

4. Test state and reject command

Aggregate state is typically only used to keep track of information that must be used to validate commands. When a command must not proceed, the command body can return a rejection with a reason.

5. Apply new event(s)

When the command is acceptable, it should apply the necessary events to the aggregate and return the result (the updated aggregate).

Alternative command invocations

defcommand installs a number of functions and multi-methods that can be used to invoke the defined command.

In the following examples, we assume the following definitions:

 (ns user
    (:require [rill.wheel :as aggregate
                          :refer [defaggregate
                                  defcommand
                                  defevent
                                  transact!
                                  ok?]]))

 (defaggregate user
    [user-id])

 (defevent registered ::user
    [user name])

 (defcommand register ::user
    [user name]
    (registered user name))

Commands as data

If you need to, you can describe and execute any command defined with defcommand as a message. The commands are implemented as maps with a rill.message/type key indicating the command type with a qualified keyword.

Command constructor

For every (defcommand cmd-name ...) definition, a constructor function named ->cmd-name is created that will create a valid rill.wheel command message:

(->register "my-id" "Some Name")

=> {:rill.message/type :user/register,
    :user-id "my-id",
    :name "Some Name"}

Note that the ->register function also takes the identifying properties of the user aggregate. This is required so that the correct user aggregate can be fetched from the repository.

transact!

You can run the given command message directly against the repository using transact!. This will fetch the aggregate using the fetch-aggregate function, calls apply-command and commit! the result.

(ok? (transact! repository
                (->register "my-id"
                            "Some Name")))

=> true

If your commands are invoked from some remote source (like a single-page application - see the mpare-net/weir project), these are the semantics you probably want (excluding authentication etc).

apply-command

The apply-command function is used by transact! and takes the aggregate to update and the command message and executes the defcommand body:

 (-> (get-user repo my-id)
     (apply-command (->register my-id my-name)))

fetch-aggregate

Also used by transact!, this multi-method takes the repository and the command message and returns the aggregate the command should be applied to.

Commands as functions

It's convenient to be able to call commands directly as regular names functions. For this purpose there are two flavors:

command-name!

A generated function named after the command with an exclamation mark added. Takes the repository and all properties to identify the aggregate plus additional command properties, and commit! the result.

 (ok? (register! repository "user-id" "Some Name"))

 => true

command-name

Another generated function that takes the aggregate as the first argument and the additional command arguments and applies the command to the aggregate. This does not commit but returns the updated aggregate or a rejection. You can chain successful calls to named command functions and commit! the result.

 (ok? (-> repository
        (get-user "user-id")
        (register "Some Name")
        (commit!)))

 => true

See also

  • rill.event-store
# Aggregates and Events

### Synopsis

    (require '[rill.wheel :as aggregate
               :refer [defaggregate defevent]])

    (defaggregate user
      "a user is identified by a single `email` property"
      [email])

    (defevent registered ::user
      "user was correctly registered"
      [user]
      (assoc user :registered? true))

    (defevent unregistered ::user
      "user has unregistered"
      [user]
      (dissoc user :registered?))

    (-> (user "user@example.com") registered :registered?)
      => true

    (registered-event (user "user@example.com"))
      => {:rill.message/type :user/registered,
          :email "user@example.com",
          :rill.wheel/type :user/user}

    (wheel/new-events some-aggreate)
      => seq-of-events

### Store and retrieve aggregates in a repository

    (-> (get-user repo "user@example.com)
        (registered)
        (command/commit!))
    ;; ...
    (get-user some-repository "user@example.com")

### Full example of defaggregate

    (defaggregate turnstile
      "An aggregate with docstring"
      [turnstile-id]
      {:pre [(instance? java.util.UUID turnstile-id)]}
      ((installed
        "A turnstile was installed"
        [turnstile]
        (assoc turnstile
               :installed? true
               :locked? true
               :coins 0
               :turns 0
               :pushes 0))

       (coin-inserted
        "A Coin was inserted into the turnstile"
        [turnstile]
        (-> turnstile
            (update :coins inc)
            (assoc :locked? false)))

       (arm-turned
        "The turnstile's arm was turned"
        [turnstile]
        (-> turnstile
            (update :pushes inc)
            (update :turns inc)
            (assoc :locked? true)))

       (arm-pushed-ineffectively
        "The arm was pushed but did not move"
        [turnstile]
        (-> turnstile
            (update :pushes inc))))

      ((install-turnstile
        [repo turnstile-id]
        (let [turnstile (get-turnstile repo turnstile-id)]
          (if (wheel/exists turnstile)
            (rejection turnstile "Already exists")
            (installed turnstile))))

       (insert-coin
        "Insert coin into turnstile, will unlock"
        [repo turnstile-id]
        (let [turnstile (get-turnstile repo turnstile-id)]
          (if (:installed? turnstile)
            (coin-inserted turnstile)
            (rejection turnstile "Turnstile not installed"))))

       (push-arm
        "Push the arm, might turn or be ineffective"
        {::wheel/events [::arm-pushed-ineffectively ::arm-turned]}
        [repo turnstile-id]
        (let [turnstile (get-turnstile repo turnstile-id)]
          (cond
            (not (:installed? turnstile))
            (rejection turnstile "Not installed")
            (:locked? turnstile)
            (arm-pushed-ineffectively turnstile)
            :else
             (arm-turned turnstile))))))


# Commands

Commands are functions that apply new events to aggregates.

## Command flow

     (-> (get-some-aggregate repository id) ; 1.
         (cmd-call additional-argument)     ; 2.
         (commit!))                         ; 3.

### 1. Fetch aggregate

Before calling the command, the aggregate it applies to should get
fetched from the `repository`. In rill/wheel, this will always work
and must be done even for aggregates that have no events applied to
them - this will result in an `rill.wheel/empty?`
aggregate that can be committed later.

### 2. Calling the command

A command can have any number of arguments, and it's idiomatic for
commands to take the aggregate-to-change as the first argument.

As a matter of style, it's suggested that commands do not fetch
other objects from the repository but are explicitly passed any
necessary aggregates.

#### Success

A successful command returns an `uncommitted?` aggregate.

#### Rejection

A command may be rejected, in which case the command returns a
`rejection` - meaning the request was denied for business
reasons. Rejections are explicitly constructed in the `defcommand`
body by the application writer.

It's typically useless to retry a rejected command.

### 3. Committing results

The result of a command can be persisted back to the repository by
calling `commit!`. If `commit!` is passed a `rejection` it will
return it. Otherwise the argument should be an aggregate that will
be persisted.

### ok

A successful commit returns an `ok?` object describing the committed
events and aggregate.

### conflict

Committing an updated aggregate can return a `conflict`, meaning
there were changes to the aggregate in the repository in the time
between fetching the aggregate and calling `commit!`.

Depending on the use case, it may be useful to update the aggregate
and retry a conflicted command.

## Defining commands

     (defaggregate x                                 ; 1.
       [id])

     (defevent x-happened                            ; 2.
        "X happened to obj"
        [obj arg1]
        (assoc obj :a arg1))

     (defcommand do-x ::x                            ; 3.
        "Make X happen to obj"
        [obj arg1]
        (if (= (:a obj) arg1))                       ; 4.
            (rejection obj "Arg already applied")
            (x-happened obj)))                       ; 5.


### 1. Define the aggregate type

### 2. Define events to apply

Commands can only affect aggregates by applying events to them. Here
we define an event with `defevent`. When the `x-happened` event is
applied it will set key `:a` of aggregate `obj`.

It's idiomatic to give events a past-tense name, to indicate that
the event happened and cannot be rejected.

### 3. Define command

Commands are defined by calling `defcommand`, specifying a name,
aggregate type, optional docstring, argument vector and a command
body.

### 4. Test state and reject command

Aggregate state is typically only used to keep track of information
that must be used to validate commands. When a command must not
proceed, the command body can return a `rejection` with a reason.

### 5. Apply new event(s)

When the command is acceptable, it should apply the necessary events
to the aggregate and return the result (the updated aggregate).

## Alternative command invocations

`defcommand` installs a number of functions and multi-methods that
can be used to invoke the defined command.

In the following examples, we assume the following definitions:


     (ns user
        (:require [rill.wheel :as aggregate
                              :refer [defaggregate
                                      defcommand
                                      defevent
                                      transact!
                                      ok?]]))

     (defaggregate user
        [user-id])

     (defevent registered ::user
        [user name])

     (defcommand register ::user
        [user name]
        (registered user name))

### Commands as data

If you need to, you can describe and execute any command defined
with `defcommand` as a message. The commands are implemented as maps
with a `rill.message/type` key indicating the command type with a
qualified keyword.

#### Command constructor

For every `(defcommand cmd-name ...)` definition, a constructor
function named `->cmd-name` is created that will create a valid
rill.wheel command message:

    (->register "my-id" "Some Name")

    => {:rill.message/type :user/register,
        :user-id "my-id",
        :name "Some Name"}

Note that the ->register function also takes the identifying
properties of the `user` aggregate. This is required so that the
correct `user` aggregate can be fetched from the repository.

#### `transact!`

You can run the given command message directly against the
repository using `transact!`. This will fetch the aggregate using
the `fetch-aggregate` function, calls `apply-command` and `commit!`
the result.

    (ok? (transact! repository
                    (->register "my-id"
                                "Some Name")))

    => true

If your commands are invoked from some remote source (like a
single-page application - see the `mpare-net/weir` project), these
are the semantics you probably want (excluding authentication etc).

#### `apply-command`

The apply-command function is used by `transact!` and takes the
aggregate to update and the command message and executes the
`defcommand` `body`:

     (-> (get-user repo my-id)
         (apply-command (->register my-id my-name)))

#### `fetch-aggregate`

Also used by `transact!`, this multi-method takes the repository and
the command message and returns the aggregate the command should be
applied to.

### Commands as functions

It's convenient to be able to call commands directly as regular
names functions. For this purpose there are two flavors:

#### `command-name!`

A generated function named after the command with an exclamation
mark added. Takes the `repository` and all properties to identify
the `aggregate` plus additional command properties, and `commit!`
the result.

     (ok? (register! repository "user-id" "Some Name"))

     => true

#### `command-name`

Another generated function that takes the `aggregate` as the first
argument and the additional command arguments and applies the
command to the `aggregate`. This does not commit but returns the
updated `aggregate` or a `rejection`. You can chain successful calls
to named command functions and `commit!` the result.

     (ok? (-> repository
            (get-user "user-id")
            (register "Some Name")
            (commit!)))

     => true

## See also

- `rill.event-store`
raw docstring

aggregateclj

(aggregate result)

Return the aggregate from a result. If the result is an aggregate, returns it as is.

Return the aggregate from a result. If the result *is* an
aggregate, returns it as is.
sourceraw docstring

aggregate?clj

(aggregate? obj)

Test that obj is an aggregate.

Test that `obj` is an aggregate.
sourceraw docstring

apply-commandclj

(apply-command aggregate command)

Given a command and aggregate, apply the command to the aggregate. Should return an updated aggregate or a rejection.

Given a command and aggregate, apply the command to the
aggregate. Should return an updated aggregate or a rejection.
sourceraw docstring

apply-command*cljmultimethod

source

apply-eventclj

(apply-event aggregate event)

Update the properties of aggregate given event. Implementation for different event types will be given by defevent.

Update the properties of `aggregate` given `event`. Implementation
for different event types will be given by `defevent`.
sourceraw docstring

apply-event*cljmultimethod

source

apply-new-eventclj

(apply-new-event aggregate event)

Apply a new event to the aggregate. The new events will be committed when the aggregate is committed to a repository.

Apply a new event to the aggregate. The new events will be committed
when the aggregate is committed to a repository.
sourceraw docstring

apply-stored-eventclj

(apply-stored-event aggregate event)

Apply a previously committed event to the aggregate. This increments the version of the aggregate.

Apply a previously committed event to the aggregate. This
increments the version of the aggregate.
sourceraw docstring

command-result?clj

(command-result? r)
source

commit!clj

(commit! aggregate-or-rejection)

Commit the result of a command execution. If the command returned a rejection nothing is committed and the rejection is returned. If the result is an aggregate it is committed to the repository. If that succeeds an ok is returned. Otherwise a conflict is returned.

Commit the result of a command execution. If the command returned a
`rejection` nothing is committed and the rejection is returned. If the
result is an aggregate it is committed to the repository. If that
succeeds an `ok` is returned. Otherwise a `conflict` is returned.
sourceraw docstring

conflictclj

(conflict aggregate)

Creates a conflict. Use conflict when there were changes to the aggregate in the repository in the time between fetching the aggregate and calling commit!.

Creates a `conflict`. Use `conflict` when there were changes to
the aggregate in the repository in the time between fetching the
aggregate and calling `commit!`.
sourceraw docstring

conflict?clj

(conflict? result)

Was there a conflict between fetching the aggregate and calling commit?

Was there a conflict between fetching the aggregate and calling
`commit`?
sourceraw docstring

defaggregatecljmacro

(defaggregate name
              doc-string?
              attr-map?
              [properties*]
              pre-post-map?
              events?
              commands?)

Defines an aggregate type, and aggregate-id function. The aggregate's id key is a map with a key for every property in properties*, plus the aggregate type, a qualified keyword from name.

Also defines a function get-{name}, which takes an additional first repository argument and retrieves the aggregate.

events? and commands? are sequences of event specs and command specs and passed to defevent and rill.wheel/defcommand respectively.

Defines an aggregate type, and aggregate-id function. The
aggregate's id key is a map with a key for every property in
`properties*`, plus the aggregate type, a qualified keyword from
`name`.

Also defines a function `get-{name}`, which takes an additional
first repository argument and retrieves the aggregate.

events? and commands? are sequences of event specs and command specs
and passed to `defevent` and `rill.wheel/defcommand`
respectively.
sourceraw docstring

defcommandcljmacro

(defcommand name
            aggregate-type
            doc-string?
            attr-map?
            [repository properties*]
            pre-post-map?
            body)

Defines a command as a named function that takes any arguments and returns an aggregate or rejection that can be passed to commit!.

The metadata of the command may contain a :rill.wheel/events key, which will specify the types of the events that may be generated. As a convenience, the corresponding event functions are declared automatically so the defevent statements can be written after the command. This usually reads a bit nicer.

 (defcommand answer-question ::question
   "Try to answer a question"
   [question user-id answer]
   (if (some-check question answer)
     (answered-correctly question user-id answer)
     (answered-incorrectly question user-id answer)))

Defines a command as a named function that takes any arguments and
returns an `aggregate` or `rejection` that can be passed to
`commit!`.

The metadata of the command may contain a
`:rill.wheel/events` key, which will specify the types of
the events that may be generated. As a convenience, the
corresponding event functions are `declare`d automatically so the
`defevent` statements can be written after the command. This usually
reads a bit nicer.

     (defcommand answer-question ::question
       "Try to answer a question"
       [question user-id answer]
       (if (some-check question answer)
         (answered-correctly question user-id answer)
         (answered-incorrectly question user-id answer)))
sourceraw docstring

defeventcljmacro

(defevent name
          aggregate-type
          doc-string?
          attr-map?
          [aggregate properties*]
          pre-post-map?
          body*)

Defines function that takes aggregate + properties, constructs an event and applies the event as a new event to aggregate. Properties defined on the aggregate-type definition will be merged with the event; do not define properties with defevent that are already defined in the corresponding defaggregate.

For cases where you only need the event and can ignore the aggregate, the function "{name}-event" is defined with the same signature. This function is used by the "{name}" function to generate the event before calling apply-event (see below).

The given prepost-map, if supplied gets included in the definition of the "{name}-event" function.

The given body, if supplied, defines an apply-event multimethod that applies the event to the aggregate. If no body is supplied, the default apply-event will be used, which will return the aggregate as is.

Defines function that takes aggregate + properties, constructs an
event and applies the event as a new event to aggregate. Properties
defined on the `aggregate-type` definition will be merged with the
event; do not define properties with `defevent` that are already
defined in the corresponding `defaggregate`.

For cases where you only need the event and can ignore the
aggregate, the function "{name}-event" is defined with the same
signature. This function is used by the "{name}" function to
generate the event before calling `apply-event` (see below).

The given `prepost-map`, if supplied gets included in the definition
of the "{name}-event" function.

The given `body`, if supplied, defines an `apply-event` multimethod
that applies the event to the aggregate. If no `body` is supplied,
the default `apply-event` will be used, which will return the
aggregate as is.
sourceraw docstring

emptyclj

(empty aggregate-id)

Create a new aggregate with id aggregate-id and no events. Aggregate version will be -1. Note that empty aggregates cannot be stored.

Create a new aggregate with id `aggregate-id` and no
events. Aggregate version will be -1. Note that empty aggregates
cannot be stored.
sourceraw docstring

empty?clj

(empty? aggregate)

Test that the aggregate is new and has no uncommitted events.

Test that the aggregate is new and has no uncommitted events.
sourceraw docstring

existsclj

(exists aggregate)

If aggregate is not new, return aggregate, otherwise nil.

If aggregate is not new, return aggregate, otherwise nil.
sourceraw docstring

fetch-aggregatecljmultimethod

Given a command and repository, fetch the target aggregate.

Given a command and repository, fetch the target aggregate.
sourceraw docstring

merge-aggregate-propsclj

(merge-aggregate-props aggregate partial-event)
source

message?clj

(message? m)
source

new-eventsclj

(new-events aggregate)

The events that will be committed when this aggregate is committed.

The events that will be committed when this aggregate is committed.
sourceraw docstring

new?clj

(new? aggregate)

Test that the aggregate has no committed events.

Test that the aggregate has no committed events.
sourceraw docstring

okclj

(ok aggregate)

Creates an ok? object describing the committed events and aggregate.

Creates an `ok?` object describing the committed events and
aggregate.
sourceraw docstring

ok?clj

(ok? result)

Is result an ok? object?

Is result an `ok?` object?
sourceraw docstring

reasonclj

(reason rejection)

Return the reason for a rejection. Returns :rill.wheel/conflict for a conflict.

Return the reason for a `rejection`. Returns `:rill.wheel/conflict`
for a conflict.
sourceraw docstring

rejectionclj

(rejection aggregate reason)

Create a rejection for aggregate with reason.

Create a rejection for aggregate with reason.
sourceraw docstring

rejection?clj

(rejection? result)

Checks if result is rejected.

Checks if result is rejected.
sourceraw docstring

repositoryclj

(repository aggregate)

Return the repository of aggregate.

Return the repository of `aggregate`.
sourceraw docstring

transact!clj

(transact! repo command)

Run and commit the given command against the repository.

Run and commit the given command against the repository.
sourceraw docstring

typeclj

(type aggregate)

Return the type of this aggregate.

Return the type of this aggregate.
sourceraw docstring

type-propertiesclj

(type-properties t)

The properties of the identifier of aggregate type t.

The properties of the identifier of aggregate type `t`.
sourceraw docstring

uncommitted?clj

(uncommitted? aggregate)

aggregate has events applied that can be committed.

`aggregate` has events applied that can be committed.
sourceraw docstring

cljdoc builds & hosts documentation for Clojure/Script libraries

Keyboard shortcuts
Ctrl+kJump to recent docs
←Move to previous article
→Move to next article
Ctrl+/Jump to the search field
Raise an issueBrowse cljdoc sourceChat on Slack
× close