Repository: commanded/commanded
Pull request #448 (2021-06-07 01:49:34)

Get aggregate state
This PR addresses https://github.com/commanded/commanded/issues/336



Patch:
diff --git a/lib/application.ex b/lib/application.ex
index 61f01bc..0c61009 100644
--- a/lib/application.ex
+++ b/lib/application.ex
@@ -202,6 +202,15 @@ defmodule Commanded.Application do
         Supervisor.stop(pid, :normal, timeout)
       end
 
+      def aggregate_state(aggregate_module, aggregate_uuid, timeout \\ 5000) do
+        Commanded.Aggregates.Aggregate.aggregate_state(
+          __MODULE__,
+          aggregate_module,
+          aggregate_uuid,
+          timeout
+        )
+      end
+
       defp name(opts) do
         case Keyword.get(opts, :name) do
           nil ->
diff --git a/lib/commanded.ex b/lib/commanded.ex
index 84c536f..f8e2754 100644
--- a/lib/commanded.ex
+++ b/lib/commanded.ex
@@ -29,4 +29,20 @@ defmodule Commanded do
     opts = [strategy: :one_for_one, name: Commanded.Supervisor]
     Supervisor.start_link(children, opts)
   end
+
+  @doc """
+  Retrieve aggregate state of an aggregate.
+
+  Retrieving aggregate state is done by calling to the opened aggregate,
+  or querying the event store for an optional state snapshot
+  and then replaying the aggregate's event stream.
+  """
+  def aggregate_state(application, aggregate_module, aggregate_uuid, timeout \\ 5_000) do
+    Commanded.Aggregates.Aggregate.aggregate_state(
+      application,
+      aggregate_module,
+      aggregate_uuid,
+      timeout
+    )
+  end
 end
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index e9d5659..8c58152 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -99,19 +99,18 @@ defmodule Commanded.Aggregates.Aggregate do
 
   require Logger
 
+  alias Commanded.Application.Config
   alias Commanded.Aggregates.Aggregate
+  alias Commanded.Aggregates.AggregateStateBuilder
   alias Commanded.Aggregates.ExecutionContext
   alias Commanded.Event.Mapper
   alias Commanded.Event.Upcast
   alias Commanded.EventStore
   alias Commanded.EventStore.RecordedEvent
-  alias Commanded.EventStore.SnapshotData
   alias Commanded.Registration
   alias Commanded.Snapshotting
   alias Commanded.Telemetry
 
-  @read_event_batch_size 100
-
   defstruct [
     :application,
     :aggregate_module,
@@ -197,7 +196,38 @@ defmodule Commanded.Aggregates.Aggregate do
   @doc false
   def aggregate_state(application, aggregate_module, aggregate_uuid, timeout \\ 5_000) do
     name = via_name(application, aggregate_module, aggregate_uuid)
-    GenServer.call(name, :aggregate_state, timeout)
+
+    try do
+      GenServer.call(name, :aggregate_state, timeout)
+    catch
+      :exit, {reason, {GenServer, :call, [^name, :aggregate_state, ^timeout]}}
+      when reason in [:normal, :noproc] ->
+        task =
+          Task.async(fn ->
+            snapshot_options =
+              application
+              |> Config.get(:snapshotting)
+              |> Kernel.||(%{})
+              |> Map.get(aggregate_module, [])
+
+            %Aggregate{
+              application: application,
+              aggregate_module: aggregate_module,
+              aggregate_uuid: aggregate_uuid,
+              snapshotting: Snapshotting.new(application, aggregate_uuid, snapshot_options)
+            }
+            |> Commanded.Aggregates.AggregateStateBuilder.populate()
+            |> Map.fetch!(:aggregate_state)
+          end)
+
+        case Task.yield(task, timeout) || Task.shutdown(task) do
+          {:ok, result} ->
+            result
+
+          nil ->
+            exit({:timeout, {GenServer, :call, [name, :aggregate_state, timeout]}})
+        end
+    end
   end
 
   @doc false
@@ -231,7 +261,7 @@ defmodule Commanded.Aggregates.Aggregate do
   def handle_continue(:populate_aggregate_state, %Aggregate{} = state) do
     # Subscribe to aggregate's events to catch any events appended to its stream
     # by another process, such as directly appended to the event store.
-    {:noreply, populate_aggregate_state(state), {:continue, :subscribe_to_events}}
+    {:noreply, AggregateStateBuilder.populate(state), {:continue, :subscribe_to_events}}
   end
 
   @doc false
@@ -410,69 +440,6 @@ defmodule Commanded.Aggregates.Aggregate do
     end
   end
 
-  # Populate the aggregate's state from a snapshot, if present, and it's events.
-  #
-  # Attempt to fetch a snapshot for the aggregate to use as its initial state.
-  # If the snapshot exists, fetch any subsequent events to rebuild its state.
-  # Otherwise start with the aggregate struct and stream all existing events for
-  # the aggregate from the event store to rebuild its state from those events.
-  defp populate_aggregate_state(%Aggregate{} = state) do
-    %Aggregate{aggregate_module: aggregate_module, snapshotting: snapshotting} = state
-
-    aggregate =
-      case Snapshotting.read_snapshot(snapshotting) do
-        {:ok, %SnapshotData{source_version: source_version, data: data}} ->
-          %Aggregate{
-            state
-            | aggregate_version: source_version,
-              aggregate_state: data
-          }
-
-        {:error, _error} ->
-          # No snapshot present, or exists but for outdated state, so use intial empty state
-          %Aggregate{state | aggregate_version: 0, aggregate_state: struct(aggregate_module)}
-      end
-
-    rebuild_from_events(aggregate)
-  end
-
-  # Load events from the event store, in batches, to rebuild the aggregate state
-  defp rebuild_from_events(%Aggregate{} = state) do
-    %Aggregate{
-      application: application,
-      aggregate_uuid: aggregate_uuid,
-      aggregate_version: aggregate_version
-    } = state
-
-    case EventStore.stream_forward(
-           application,
-           aggregate_uuid,
-           aggregate_version + 1,
-           @read_event_batch_size
-         ) do
-      {:error, :stream_not_found} ->
-        # aggregate does not exist, return initial state
-        state
-
-      event_stream ->
-        rebuild_from_event_stream(event_stream, state)
-    end
-  end
-
-  # Rebuild aggregate state from a `Stream` of its events.
-  defp rebuild_from_event_stream(event_stream, %Aggregate{} = state) do
-    Enum.reduce(event_stream, state, fn event, state ->
-      %RecordedEvent{data: data, stream_version: stream_version} = event
-      %Aggregate{aggregate_module: aggregate_module, aggregate_state: aggregate_state} = state
-
-      %Aggregate{
-        state
-        | aggregate_version: stream_version,
-          aggregate_state: aggregate_module.apply(aggregate_state, data)
-      }
-    end)
-  end
-
   defp aggregate_lifespan_timeout(lifespan, function_name, args) do
     # Take the last event or the command or error
     args = args |> List.wrap() |> Enum.take(-1)
@@ -580,7 +547,7 @@ defmodule Commanded.Aggregates.Aggregate do
     else
       {:error, :wrong_expected_version} ->
         # Fetch missing events from event store
-        state = rebuild_from_events(state)
+        state = AggregateStateBuilder.rebuild_from_events(state)
 
         # Retry command if there are any attempts left
         case ExecutionContext.retry(context) do
diff --git a/lib/commanded/aggregates/aggregate_state_builder.ex b/lib/commanded/aggregates/aggregate_state_builder.ex
new file mode 100644
index 0000000..ec51dee
--- /dev/null
+++ b/lib/commanded/aggregates/aggregate_state_builder.ex
@@ -0,0 +1,77 @@
+defmodule Commanded.Aggregates.AggregateStateBuilder do
+  alias Commanded.Aggregates.Aggregate
+
+  alias Commanded.EventStore
+  alias Commanded.EventStore.RecordedEvent
+  alias Commanded.EventStore.SnapshotData
+  alias Commanded.Snapshotting
+
+  @read_event_batch_size 100
+
+  @doc """
+  Populate the aggregate's state from a snapshot, if present, and it's events.
+
+  Attempt to fetch a snapshot for the aggregate to use as its initial state.
+  If the snapshot exists, fetch any subsequent events to rebuild its state.
+  Otherwise start with the aggregate struct and stream all existing events for
+  the aggregate from the event store to rebuild its state from those events.
+  """
+  def populate(%Aggregate{} = state) do
+    %Aggregate{aggregate_module: aggregate_module, snapshotting: snapshotting} = state
+
+    aggregate =
+      case Snapshotting.read_snapshot(snapshotting) do
+        {:ok, %SnapshotData{source_version: source_version, data: data}} ->
+          %Aggregate{
+            state
+            | aggregate_version: source_version,
+              aggregate_state: data
+          }
+
+        {:error, _error} ->
+          # No snapshot present, or exists but for outdated state, so use intial empty state
+          %Aggregate{state | aggregate_version: 0, aggregate_state: struct(aggregate_module)}
+      end
+
+    rebuild_from_events(aggregate)
+  end
+
+  @doc """
+  Load events from the event store, in batches, to rebuild the aggregate state
+  """
+  def rebuild_from_events(%Aggregate{} = state) do
+    %Aggregate{
+      application: application,
+      aggregate_uuid: aggregate_uuid,
+      aggregate_version: aggregate_version
+    } = state
+
+    case EventStore.stream_forward(
+           application,
+           aggregate_uuid,
+           aggregate_version + 1,
+           @read_event_batch_size
+         ) do
+      {:error, :stream_not_found} ->
+        # aggregate does not exist, return initial state
+        state
+
+      event_stream ->
+        rebuild_from_event_stream(event_stream, state)
+    end
+  end
+
+  # Rebuild aggregate state from a `Stream` of its events.
+  defp rebuild_from_event_stream(event_stream, %Aggregate{} = state) do
+    Enum.reduce(event_stream, state, fn event, state ->
+      %RecordedEvent{data: data, stream_version: stream_version} = event
+      %Aggregate{aggregate_module: aggregate_module, aggregate_state: aggregate_state} = state
+
+      %Aggregate{
+        state
+        | aggregate_version: stream_version,
+          aggregate_state: aggregate_module.apply(aggregate_state, data)
+      }
+    end)
+  end
+end
diff --git a/mix.exs b/mix.exs
index ab6c82d..b08ba6c 100644
--- a/mix.exs
+++ b/mix.exs
@@ -136,6 +136,7 @@ defmodule Commanded.Mixfile do
         Aggregates: [
           Commanded.Aggregate.Multi,
           Commanded.Aggregates.Aggregate,
+          Commanded.Aggregates.AggregateStateBuilder,
           Commanded.Aggregates.AggregateLifespan,
           Commanded.Aggregates.DefaultLifespan,
           Commanded.Aggregates.ExecutionContext,


Test patch:
diff --git a/test/aggregates/aggregate_state_test.exs b/test/aggregates/aggregate_state_test.exs
new file mode 100644
index 0000000..fbcf9fd
--- /dev/null
+++ b/test/aggregates/aggregate_state_test.exs
@@ -0,0 +1,94 @@
+defmodule Commanded.Aggregates.AggregateStateTest do
+  use ExUnit.Case
+
+  alias Commanded.DefaultApp
+
+  alias Commanded.Aggregates.{
+    Aggregate,
+    AppendItemsHandler,
+    ExampleAggregate,
+    ExecutionContext,
+    Supervisor
+  }
+
+  alias Commanded.Aggregates.ExampleAggregate.Commands.AppendItems
+
+  @aggregate_module ExampleAggregate
+
+  setup do
+    start_supervised!({DefaultApp, snapshotting: %{ExampleAggregate => [snapshot_every: 10]}})
+
+    :ok
+  end
+
+  test "query aggregate_state from the running aggregate gen_server" do
+    aggregate_uuid = UUID.uuid4()
+    append_items(aggregate_uuid, 9)
+
+    assert_aggregate_version(aggregate_uuid, 9)
+    assert_aggregate_state(aggregate_uuid, 9)
+  end
+
+  test "get aggregate_state by querying the state snapshot" do
+    aggregate_uuid = UUID.uuid4()
+    append_items(aggregate_uuid, 10)
+
+    assert_aggregate_version(aggregate_uuid, 10)
+    stop_aggregate(aggregate_uuid)
+    assert_aggregate_state(aggregate_uuid, 10)
+  end
+
+  test "get aggregate_state by querying the state snapshot and rebuilding from events" do
+    aggregate_uuid = UUID.uuid4()
+    append_items(aggregate_uuid, 11)
+
+    assert_aggregate_version(aggregate_uuid, 11)
+    stop_aggregate(aggregate_uuid)
+
+    assert_aggregate_state(aggregate_uuid, 11)
+  end
+
+  test "get aggregate_state by rebuilding from events" do
+    aggregate_uuid = UUID.uuid4()
+    append_items(aggregate_uuid, 9)
+
+    assert_aggregate_version(aggregate_uuid, 9)
+    stop_aggregate(aggregate_uuid)
+
+    assert_aggregate_state(aggregate_uuid, 9)
+  end
+
+  # Assert aggregate's state equals the given expected state.
+  defp assert_aggregate_state(aggregate_uuid, last_index) do
+    assert Aggregate.aggregate_state(DefaultApp, @aggregate_module, aggregate_uuid) ==
+             %@aggregate_module{
+               items: Enum.to_list(1..last_index),
+               last_index: last_index
+             }
+  end
+
+  # Assert aggregate's version equals the given expected version.
+  defp assert_aggregate_version(aggregate_uuid, expected_version) do
+    assert Aggregate.aggregate_version(DefaultApp, @aggregate_module, aggregate_uuid) ==
+             expected_version
+  end
+
+  # Shutdown the aggregate process
+  defp stop_aggregate(aggregate_uuid) do
+    assert :ok = Aggregate.shutdown(DefaultApp, @aggregate_module, aggregate_uuid)
+  end
+
+  defp append_items(aggregate_uuid, count) do
+    execution_context = %ExecutionContext{
+      command: %AppendItems{count: count},
+      handler: AppendItemsHandler,
+      function: :handle
+    }
+
+    {:ok, ^aggregate_uuid} =
+      Supervisor.open_aggregate(DefaultApp, @aggregate_module, aggregate_uuid)
+
+    {:ok, _count, _events} =
+      Aggregate.execute(DefaultApp, @aggregate_module, aggregate_uuid, execution_context)
+  end
+end