Repository: slashdotdash/commanded
Pull request #63 (2017-07-20 13:40:25)

Event handler and process manager macros
`Commanded.Event.Handler` and `Commanded.ProcessManagers.ProcessManager` macros to simplify defining, and starting, event handlers and process managers.

Note the previous approach to defining and starting may still be used, so this is *not* a breaking change.

Release 0.12.0


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 77a1202..4b015ab 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Changelog
 
+## v0.12.0
+
+### Enhancements
+
+- `Commanded.Event.Handler` and `Commanded.ProcessManagers.ProcessManager` macros to simplify defining, and starting, event handlers and process managers. Note the previous approach to defining and starting may still be used, so this is *not* a breaking change.
+
 ## v0.11.0
 
 ### Enhancements
diff --git a/README.md b/README.md
index 6914b33..ba22d69 100644
--- a/README.md
+++ b/README.md
@@ -81,7 +81,7 @@ You must decide which event store to use with Commanded. You have a choice betwe
     end
     ```
 
-2. Include `:eventstore` in the list of applications to start. 
+2. Include `:eventstore` in the list of applications to start.
 
     For **Elixir 1.4**, add `:eventstore` to the extra applications list in `mix.exs`:
 
@@ -96,7 +96,7 @@ You must decide which event store to use with Commanded. You have a choice betwe
       ]
     end
     ```
-    
+
     For **Elixir 1.3** and before, add `:eventstore` to the applications list in `mix.exs`:
 
     ```elixir
@@ -110,7 +110,7 @@ You must decide which event store to use with Commanded. You have a choice betwe
       ]
     end
     ```
-    
+
 3. Configure Commanded to use the event store adapter:
 
     ```elixir
@@ -416,62 +416,57 @@ end
 
 ### Event handlers
 
-Create an event handler module which implements the `Commanded.Event.Handler` behaviour.
-
-This consists of a single `handle/2` function that receives each published domain event and its metadata, including the event's unique id. It should return `:ok` on success or `{:error, :reason}` on failure.
+Use the `Commanded.Event.Handler` macro within your event handler module to implement the defined behaviour. This consists of a single `handle/2` function that receives each published domain event and its metadata, including the event's unique event number. It should return `:ok` on success or `{:error, :reason}` on failure. You can return `{:error, :already_seen_event}` to skip events that have already been handled, due to the at-least-once event delivery of the supported event stores.
 
-Use pattern matching to match on each type of event you are interested in. Add a catch-all `handle/2` function for all other events to be ignored.
+Use pattern matching to match on each type of event you are interested in. A catch-all `handle/2` function is included, so all other events will be ignored by default.
 
 ```elixir
 defmodule AccountBalanceHandler do
-  @behaviour Commanded.Event.Handler
-
-  def start_link do
-    Agent.start_link(fn -> 0 end, name: __MODULE__)
-  end
+  use Commanded.Event.Handler, name: "account_balance"
 
   def handle(%BankAccountOpened{initial_balance: initial_balance}, _metadata) do
-    Agent.update(__MODULE__, fn _ -> initial_balance end)
+    Agent.update(AccountBalance, fn _ -> initial_balance end)
   end
 
   def handle(%MoneyDeposited{balance: balance}, _metadata) do
-    Agent.update(__MODULE__, fn _ -> balance end)
+    Agent.update(AccountBalance, fn _ -> balance end)
   end
 
-  # ignore all other events
-  def handle(_event, _metadata), do: :ok
-
   def current_balance do
-    Agent.get(__MODULE__, fn balance -> balance end)
+    Agent.get(AccountBalance, fn balance -> balance end)
   end
 end
 ```
 
-Register the event handler with a given name. This is used when subscribing to the event store to track which events the handler has seen during restarts.
+The name given to the event handler **must be** unique and remain unchanged between releases. It is used when subscribing to the event store to track which events the handler has seen during restarts.
 
 ```elixir
-{:ok, _balance} = AccountBalanceHandler.start_link
-{:ok, _handler} = Commanded.Event.Handler.start_link("account_balance", AccountBalanceHandler)
+{:ok, _balance} = Agent.start_link(fn -> 0 end, name: AccountBalance)
+{:ok, _handler} = AccountBalanceHandler.start_link()
 ```
 
-You can choose to start the event handler's event store subscription from the `:origin`, `:current` position or an exact event id using the `start_from` option. The default is to use the origin so your handler will receive all events.
+You can choose to start the event handler's event store subscription from the `:origin`, `:current` position or an exact event number using the `start_from` option. The default is to use the origin so your handler will receive all events.
 
 ```elixir
-# start from :origin, :current, or an explicit event id
-{:ok, _handler} = Commanded.Event.Handler.start_link("account_balance", AccountBalanceHandler, start_from: :origin)
+# start from :origin, :current, or an explicit event number (e.g. 1234)
+defmodule AccountBalanceHandler do
+  use Commanded.Event.Handler, name: "account_balance", start_from: :origin
+
+  # ...
+end
+
+{:ok, _handler} = AccountBalanceHandler.start_link(start_from: :current)
 ```
 
 Use the `:current` position when you don't want newly created event handlers to go through all previous events. An example would be adding an event handler to send transactional emails to an already deployed system containing many historical events.
 
-You should use a [supervisor](#supervision) to start your event handlers to ensure they are restarted on error.
+You should start your event handlers using a [supervisor](#supervision) to ensure they are restarted on error.
 
 ### Process managers
 
-A process manager is responsible for coordinating one or more aggregate roots.
-
-It handles events and may dispatch commands in response. Process managers have state that can be used to track which aggregate roots are being orchestrated.
+A process manager is responsible for coordinating one or more aggregate roots. It handles events and dispatches commands in response. Process managers have state that can be used to track which aggregate roots are being orchestrated.
 
-A process manager must implement the `Commanded.ProcessManagers.ProcessManager` behaviour. It defines three callback functions: `interested?/1`, `handle/2`, and `apply/2`.
+Use the `Commanded.ProcessManagers.ProcessManager` macro in your process manager module and implement the three callback functions defined in the behaviour: `interested?/1`, `handle/2`, and `apply/2`.
 
 #### `interested?/1`
 
@@ -492,7 +487,9 @@ The `apply/2` function is used to mutate the process manager's state. It receive
 
 ```elixir
 defmodule TransferMoneyProcessManager do
-  @behaviour Commanded.ProcessManagers.ProcessManager
+  use Commanded.ProcessManagers.ProcessManager,
+    name: "transfer_money_process_manager",
+    router: BankRouter
 
   defstruct [
     transfer_uuid: nil,
@@ -543,13 +540,13 @@ defmodule TransferMoneyProcessManager do
 end
 ```
 
-Register the process manager router with a uniquely identified name. This is used when subscribing to events from the event store to track the last seen event and ensure they are only received once.
+The name given to the process manager *must* be unique. This is used when subscribing to events from the event store to track the last seen event and ensure they are only received once.
 
 ```elixir
-{:ok, _} = Commanded.ProcessManagers.Router.start_link("transfer_money_process_manager", TransferMoneyProcessManager, start_from: :current)
+{:ok, _} = TransferMoneyProcessManager.start_link(start_from: :current)
 ```
 
-You can choose to start the process router's event store subscription from the `:origin`, `:current` position or an exact event id using the `start_from` option. The default is to use the origin so it will receive all events. You typically use `:current` when adding a new process manager to an already deployed system containing historical events.
+You can choose to start the process router's event store subscription from the `:origin`, `:current` position or an exact event number using the `start_from` option. The default is to use the origin so it will receive all events. You typically use `:current` when adding a new process manager to an already deployed system containing historical events.
 
 Process manager instance state is persisted to storage after each handled event. This allows the process manager to resume should the host process terminate.
 
@@ -570,10 +567,10 @@ defmodule Bank.Supervisor do
       supervisor(Commanded.Supervisor, []),
 
       # process manager
-      worker(Commanded.ProcessManagers.ProcessRouter, ["TransferMoneyProcessManager", TransferMoneyProcessManager, BankRouter, [start_from: :current]], id: :transfer_money_process_manager),
+      worker(TransferMoneyProcessManager, [start_from: :current], id: :transfer_money_process_manager),
 
       # event handler
-      worker(Commanded.Event.Handler, ["AccountBalanceHandler", AccountBalanceHandler, [start_from: :origin]], id: :account_balance_handler)
+      worker(AccountBalanceHandler, [start_from: :origin], id: :account_balance_handler)
     ]
 
     supervise(children, strategy: :one_for_one)
@@ -629,7 +626,7 @@ You can implement the `EventStore.Serializer` behaviour to use an alternative se
 
 ## Read model projections
 
-Your read model can be built using a Commanded event handler and whatever storage provider you prefer. 
+Your read model can be built using a Commanded event handler and whatever storage provider you prefer.
 
 I typically use Ecto, and a PostgreSQL database, for read model projections. You can use the `project` macro from the [Commanded Ecto projections](https://github.com/slashdotdash/commanded-ecto-projections) library to build projectors, and have the at-least-once event delivery taken care of for you.
 
diff --git a/lib/commanded/commands/handler.ex b/lib/commanded/commands/handler.ex
index 8e397cf..f680d66 100644
--- a/lib/commanded/commands/handler.ex
+++ b/lib/commanded/commands/handler.ex
@@ -1,15 +1,16 @@
 defmodule Commanded.Commands.Handler do
   @type aggregate :: struct()
   @type command :: struct()
+  @type domain_event :: struct
   @type domain_events :: list(struct())
   @type reason :: term()
 
   @doc """
-  Apply the given command to the event-sourced aggregate root.
+  Apply the given command to the event sourced aggregate root.
 
   You must return a list containing the pending events, or `nil` / `[]` when no events produced.
 
   You should return `{:error, reason}` on failure.
   """
-  @callback handle(aggregate, command) :: domain_events | nil | {:error, reason}
+  @callback handle(aggregate, command) :: domain_event | domain_events | nil | {:error, reason}
 end
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index c7b129f..8e82dd0 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -13,9 +13,47 @@ defmodule Commanded.Event.Handler do
 
   @doc """
   Event handler behaviour to handle a domain event and its metadata
+
+  Return `:ok` on success, `{:error, :already_seen_event}` to ack and skip the event, or `{:error, reason}` on failure.
   """
   @callback handle(domain_event, metadata) :: :ok | {:error, reason :: atom}
 
+  @doc """
+  Macro as a convenience for defining an event handler
+
+    defmodule ExampleHandler do
+      use Commanded.Event.Handler, name: "example_handler"
+
+      def handle(%AnEvent{...}, _metadata) do
+        # ...
+      end
+    end
+
+    # start event handler process (or configure as a worker inside a supervisor)
+    {:ok, handler} = ExampleHandler.start_link()
+  """
+  defmacro __using__(opts) do
+    quote location: :keep do
+      @behaviour Commanded.Event.Handler
+
+      @opts unquote(opts) || []
+      @name @opts[:name] || raise "#{__MODULE__} expects :name to be given"
+
+      def start_link(opts \\ []) do
+        opts =
+          @opts
+          |> Keyword.take([:start_from])
+          |> Keyword.merge(opts)
+
+        Commanded.Event.Handler.start_link(@name, __MODULE__, opts)
+      end
+
+      def handle(_event, _metadata), do: :ok
+
+      defoverridable [handle: 2]
+    end
+  end
+
   defstruct [
     handler_name: nil,
     handler_module: nil,
diff --git a/lib/commanded/process_managers/process_manager.ex b/lib/commanded/process_managers/process_manager.ex
index 6943f9e..a50ae74 100644
--- a/lib/commanded/process_managers/process_manager.ex
+++ b/lib/commanded/process_managers/process_manager.ex
@@ -22,4 +22,53 @@ defmodule Commanded.ProcessManagers.ProcessManager do
   Mutate the process manager's state by applying the domain event
   """
   @callback apply(process_manager, domain_event) :: process_manager
+
+  @doc """
+  Macro as a convenience for defining a process manager
+
+    defmodule ExampleProcessManager do
+      use Commanded.ProcessManagers.ProcessManager,
+        name: "example_process_manager",
+        router: BankRouter
+
+      def interested?(%AnEvent{...}) do
+        # ...
+      end
+
+      def handle(%ExampleProcessManager{...}, %AnEvent{...}) do
+        # ...
+      end
+
+      def apply(%ExampleProcessManager{...}, %AnEvent{...}) do
+        # ...
+      end
+    end
+
+    # start process manager (or configure as a worker inside a supervisor)
+    {:ok, process_manager} = ExampleProcessManager.start_link()
+  """
+  defmacro __using__(opts) do
+    quote location: :keep do
+      @behaviour Commanded.ProcessManagers.ProcessManager
+
+      @opts unquote(opts) || []
+      @name @opts[:name] || raise "#{__MODULE__} expects :name to be given"
+      @router @opts[:router] || raise "#{__MODULE__} expects :router to be given"
+
+      def start_link(opts \\ []) do
+        opts =
+          @opts
+          |> Keyword.take([:start_from])
+          |> Keyword.merge(opts)
+
+        Commanded.ProcessManagers.ProcessRouter.start_link(@name, __MODULE__, @router, opts)
+      end
+
+      def interested?(_event), do: false
+      def handle(_process_manager, _event), do: []
+      def apply(process_manager, _event), do: process_manager
+
+      defoverridable [interested?: 1, handle: 2, apply: 2]
+    end
+  end
 end
diff --git a/mix.exs b/mix.exs
index a26f4fe..473ea4a 100644
--- a/mix.exs
+++ b/mix.exs
@@ -4,7 +4,7 @@ defmodule Commanded.Mixfile do
   def project do
     [
       app: :commanded,
-      version: "0.11.0",
+      version: "0.12.0",
       elixir: "~> 1.4",
       elixirc_paths: elixirc_paths(Mix.env),
       description: description(),


Test patch:
diff --git a/test/event/event_handler_macro_test.exs b/test/event/event_handler_macro_test.exs
new file mode 100644
index 0000000..d4e84bb
--- /dev/null
+++ b/test/event/event_handler_macro_test.exs
@@ -0,0 +1,47 @@
+defmodule Commanded.Event.EventHandlerMacroTest do
+  use Commanded.StorageCase
+  use Commanded.EventStore
+
+  defmodule AccountBalanceHandler do
+    use Commanded.Event.Handler, name: "account_balance_handler"
+
+    alias Commanded.ExampleDomain.BankAccount.Events.{
+      BankAccountOpened,
+      MoneyDeposited,
+    }
+
+    def handle(%BankAccountOpened{initial_balance: initial_balance}, _metadata) do
+      Agent.update(__MODULE__, fn _ -> initial_balance end)
+    end
+
+    def handle(%MoneyDeposited{balance: balance}, _metadata) do
+      Agent.update(__MODULE__, fn _ -> balance end)
+    end
+
+    def current_balance do
+      Agent.get(__MODULE__, fn balance -> balance end)
+    end
+  end
+
+  alias Commanded.Helpers.EventFactory
+  alias Commanded.Helpers.Wait
+  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
+
+  test "should handle published events" do
+    {:ok, _} = Agent.start_link(fn -> 0 end, name: AccountBalanceHandler)
+    {:ok, handler} = AccountBalanceHandler.start_link()
+
+    recorded_events =
+      [
+        %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
+        %MoneyDeposited{amount: 50, balance: 1_050}
+      ]
+      |> EventFactory.map_to_recorded_events()
+
+    send(handler, {:events, recorded_events})
+
+    Wait.until(fn ->
+      assert AccountBalanceHandler.current_balance == 1_050
+    end)
+  end
+end
diff --git a/test/example_domain/bank_router.ex b/test/example_domain/bank_router.ex
new file mode 100644
index 0000000..7f36bad
--- /dev/null
+++ b/test/example_domain/bank_router.ex
@@ -0,0 +1,14 @@
+defmodule Commanded.ExampleDomain.BankRouter do
+  use Commanded.Commands.Router
+
+  alias Commanded.ExampleDomain.{OpenAccountHandler,DepositMoneyHandler,TransferMoneyHandler,WithdrawMoneyHandler}
+  alias Commanded.ExampleDomain.{BankAccount,MoneyTransfer}
+  alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount,DepositMoney,WithdrawMoney}
+  alias Commanded.ExampleDomain.MoneyTransfer.Commands.{TransferMoney}
+
+  dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount, identity: :account_number
+  dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount, identity: :account_number
+  dispatch WithdrawMoney, to: WithdrawMoneyHandler, aggregate: BankAccount, identity: :account_number
+
+  dispatch TransferMoney, to: TransferMoneyHandler, aggregate: MoneyTransfer, identity: :transfer_uuid
+end
diff --git a/test/example_domain/money_transfer/transfer_money_process_manager.ex b/test/example_domain/money_transfer/transfer_money_process_manager.ex
index 0b5e935..a933e4c 100644
--- a/test/example_domain/money_transfer/transfer_money_process_manager.ex
+++ b/test/example_domain/money_transfer/transfer_money_process_manager.ex
@@ -1,5 +1,7 @@
 defmodule Commanded.ExampleDomain.TransferMoneyProcessManager do
-  @behaviour Commanded.ProcessManagers.ProcessManager
+  use Commanded.ProcessManagers.ProcessManager,
+    name: "transfer_money_process_manager",
+    router: Commanded.ExampleDomain.BankRouter
 
   defstruct [
     transfer_uuid: nil,
diff --git a/test/process_managers/process_manager_routing_test.exs b/test/process_managers/process_manager_routing_test.exs
index 4c899d5..46fc9e3 100644
--- a/test/process_managers/process_manager_routing_test.exs
+++ b/test/process_managers/process_manager_routing_test.exs
@@ -1,39 +1,32 @@
 defmodule Commanded.ProcessManager.ProcessManagerRoutingTest do
   use Commanded.StorageCase
-  doctest Commanded.ProcessManagers.ProcessRouter
 
-  alias Commanded.ProcessManagers.ProcessRouter
+  alias Commanded.ExampleDomain.BankRouter
   alias Commanded.ExampleDomain.TransferMoneyProcessManager
-  alias Commanded.ExampleDomain.{OpenAccountHandler,DepositMoneyHandler,TransferMoneyHandler,WithdrawMoneyHandler}
-  alias Commanded.ExampleDomain.{BankAccount,MoneyTransfer}
-  alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount,DepositMoney,WithdrawMoney}
+  alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount}
   alias Commanded.ExampleDomain.BankAccount.Events.{MoneyDeposited,MoneyWithdrawn}
   alias Commanded.ExampleDomain.MoneyTransfer.Commands.{TransferMoney}
   alias Commanded.ExampleDomain.MoneyTransfer.Events.{MoneyTransferRequested}
 
   import Commanded.Assertions.EventAssertions
 
-  defmodule BankRouter do
-    use Commanded.Commands.Router
-
-    dispatch OpenAccount, to: OpenAccountHandler, aggregate: BankAccount, identity: :account_number
-    dispatch DepositMoney, to: DepositMoneyHandler, aggregate: BankAccount, identity: :account_number
-    dispatch WithdrawMoney, to: WithdrawMoneyHandler, aggregate: BankAccount, identity: :account_number
-    dispatch TransferMoney, to: TransferMoneyHandler, aggregate: MoneyTransfer, identity: :transfer_uuid
-  end
-
   test "should start a process manager in response to an event" do
     account_number1 = UUID.uuid4
     account_number2 = UUID.uuid4
 
-    {:ok, _} = ProcessRouter.start_link("transfer_money_process_manager", TransferMoneyProcessManager, BankRouter)
+    {:ok, _} = TransferMoneyProcessManager.start_link()
 
     # create two bank accounts
     :ok = BankRouter.dispatch(%OpenAccount{account_number: account_number1, initial_balance: 1_000})
     :ok = BankRouter.dispatch(%OpenAccount{account_number: account_number2, initial_balance:  500})
 
     # transfer funds between account 1 and account 2
-    :ok = BankRouter.dispatch(%TransferMoney{transfer_uuid: UUID.uuid4, debit_account: account_number1, credit_account: account_number2, amount: 100})
+    :ok = BankRouter.dispatch(%TransferMoney{
+      transfer_uuid: UUID.uuid4,
+      debit_account: account_number1,
+      credit_account: account_number2,
+      amount: 100,
+    })
 
     assert_receive_event MoneyTransferRequested, fn event ->
       assert event.debit_account == account_number1
diff --git a/test/process_managers/process_router_process_pending_events_test.exs b/test/process_managers/process_router_process_pending_events_test.exs
index a007ad0..0363d3a 100644
--- a/test/process_managers/process_router_process_pending_events_test.exs
+++ b/test/process_managers/process_router_process_pending_events_test.exs
@@ -76,8 +76,16 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
     def handle(%ExampleAggregate{} = aggregate, %Stop{}), do: ExampleAggregate.stop(aggregate)
   end
 
+  defmodule Router do
+    use Commanded.Commands.Router
+
+    dispatch [Start,Publish,Stop], to: ExampleCommandHandler, aggregate: ExampleAggregate, identity: :aggregate_uuid
+  end
+
   defmodule ExampleProcessManager do
-    @behaviour Commanded.ProcessManagers.ProcessManager
+    use Commanded.ProcessManagers.ProcessManager,
+      name: "example_process_manager",
+      router: Router
 
     defstruct [
       status: nil,
@@ -111,16 +119,10 @@ defmodule Commanded.ProcessManager.ProcessRouterProcessPendingEventsTest do
     end
   end
 
-  defmodule Router do
-    use Commanded.Commands.Router
-
-    dispatch [Start,Publish,Stop], to: ExampleCommandHandler, aggregate: ExampleAggregate, identity: :aggregate_uuid
-  end
-
   test "should start process manager instance and successfully dispatch command" do
     aggregate_uuid = UUID.uuid4
 
-    {:ok, process_router} = ProcessRouter.start_link("example_process_manager", ExampleProcessManager, Router)
+    {:ok, process_router} = ExampleProcessManager.start_link()
 
     :ok = Router.dispatch(%Start{aggregate_uuid: aggregate_uuid})
 
diff --git a/test/process_managers/resume_process_manager_test.exs b/test/process_managers/resume_process_manager_test.exs
index 56f761f..05021ae 100644
--- a/test/process_managers/resume_process_manager_test.exs
+++ b/test/process_managers/resume_process_manager_test.exs
@@ -62,7 +62,9 @@ defmodule Commanded.ProcessManager.ResumeProcessManagerTest do
   end
 
   defmodule ExampleProcessManager do
-    @behaviour Commanded.ProcessManagers.ProcessManager
+    use Commanded.ProcessManagers.ProcessManager,
+      name: "example-process-manager",
+      router: ExampleRouter
 
     defstruct [
       status_history: []
@@ -95,7 +97,7 @@ defmodule Commanded.ProcessManager.ResumeProcessManagerTest do
   test "should resume a process manager with same state when process restarts" do
     process_uuid = UUID.uuid4
 
-    {:ok, process_router} = ProcessRouter.start_link("ExampleProcessManager", ExampleProcessManager, ExampleRouter)
+    {:ok, process_router} = ExampleProcessManager.start_link()
 
     # transfer funds between account 1 and account 2
     :ok = ExampleRouter.dispatch(%StartProcess{process_uuid: process_uuid, status: "start"})
@@ -118,7 +120,7 @@ defmodule Commanded.ProcessManager.ResumeProcessManagerTest do
     # wait for subscription to receive DOWN notification and remove subscription's PID
     :timer.sleep(1_000)
 
-    {:ok, process_router} = ProcessRouter.start_link("ExampleProcessManager", ExampleProcessManager, ExampleRouter)
+    {:ok, process_router} = ExampleProcessManager.start_link()
 
     :ok = ExampleRouter.dispatch(%ResumeProcess{process_uuid: process_uuid, status: "resume"})