diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index 5e321873..41b4f399 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -604,13 +604,7 @@ defmodule Commanded.Aggregates.Aggregate do {{:ok, []}, state} %Multi{} = multi -> - case Multi.run(multi) do - {:error, _error} = reply -> - {reply, state} - - {aggregate_state, pending_events} -> - persist_events(pending_events, aggregate_state, context, state, from) - end + run_multi(multi, context, state, from) {:ok, pending_events} -> apply_and_persist_events(pending_events, context, state, from) @@ -630,6 +624,16 @@ defmodule Commanded.Aggregates.Aggregate do {{:error, error, stacktrace}, state} end + defp run_multi(%Multi{} = multi, context, %Aggregate{} = state, from) do + case Multi.run(multi) do + {:error, _error} = reply -> + {reply, state} + + {aggregate_state, pending_events} -> + persist_events(pending_events, aggregate_state, context, state, from) + end + end + defp apply_and_persist_events(pending_events, context, %Aggregate{} = state, from) do %Aggregate{aggregate_module: aggregate_module, aggregate_state: aggregate_state} = state diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex index d71e9276..76d5db63 100644 --- a/lib/commanded/commands/router.ex +++ b/lib/commanded/commands/router.ex @@ -363,13 +363,11 @@ defmodule Commanded.Commands.Router do for command_module <- List.wrap(command_module_or_modules) do quote do - if Enum.any?(@registered_commands, fn {command_module, _command_opts} -> - command_module == unquote(command_module) - end) do - raise ArgumentError, - message: - "Command `#{inspect(unquote(command_module))}` has already been registered in router `#{inspect(__MODULE__)}`" - end + Router.__check_command_not_registered__( + __MODULE__, + @registered_commands, + unquote(command_module) + ) @registered_commands { unquote(command_module), @@ -379,6 +377,15 @@ defmodule Commanded.Commands.Router do end end + @doc false + def __check_command_not_registered__(router, registered_commands, command_module) do + if List.keymember?(registered_commands, command_module, 0) do + raise ArgumentError, + message: + "Command `#{inspect(command_module)}` has already been registered in router `#{inspect(router)}`" + end + end + @type dispatch_resp :: :ok | {:ok, aggregate_state :: struct()} diff --git a/lib/commanded/projections/ecto.ex b/lib/commanded/projections/ecto.ex index 40328f24..3a227082 100644 --- a/lib/commanded/projections/ecto.ex +++ b/lib/commanded/projections/ecto.ex @@ -360,28 +360,12 @@ if Code.ensure_loaded?(Ecto) do defp execute_batch_projection(multi, multi_fn) do # This ensures EVERYTHING happens in ONE atomic transaction with %Ecto.Multi{} = multi <- - Ecto.Multi.run(multi, :prepare_user_multi, fn _repo, - %{ - lock_and_filter: - {_current, unseen_events, - _last} - } -> - user_multi = multi_fn.(unseen_events, Ecto.Multi.new()) - - case user_multi do - %Ecto.Multi{} -> - {:ok, {unseen_events, user_multi}} - - other -> - {:error, {:invalid_multi_return, other}} - end - end), + Ecto.Multi.run(multi, :prepare_user_multi, &prepare_user_multi(&1, &2, multi_fn)), %Ecto.Multi{} = multi <- Ecto.Multi.merge(multi, fn %{prepare_user_multi: {_unseen_events, user_multi}} -> user_multi end), {:ok, changes} <- transaction(multi) do - # Get the unseen events from our preparation step {unseen_events, _user_multi} = changes.prepare_user_multi after_update_batch(unseen_events, changes) @@ -396,6 +380,17 @@ if Code.ensure_loaded?(Ecto) do reply end end + + defp prepare_user_multi( + _repo, + %{lock_and_filter: {_current, unseen_events, _last}}, + multi_fn + ) do + case multi_fn.(unseen_events, Ecto.Multi.new()) do + %Ecto.Multi{} = user_multi -> {:ok, {unseen_events, user_multi}} + other -> {:error, {:invalid_multi_return, other}} + end + end end end diff --git a/test/aggregates/aggregate_concurrency_test.exs b/test/aggregates/aggregate_concurrency_test.exs index d57ae6ab..d5b1107d 100644 --- a/test/aggregates/aggregate_concurrency_test.exs +++ b/test/aggregates/aggregate_concurrency_test.exs @@ -1,11 +1,11 @@ defmodule Commanded.Aggregates.AggregateConcurrencyTest do use Commanded.MockEventStoreCase - alias Commanded.MockedApp alias Commanded.Aggregates.{Aggregate, ExecutionContext} alias Commanded.ExampleDomain.{BankAccount, DepositMoneyHandler, OpenAccountHandler} alias Commanded.ExampleDomain.BankAccount.Commands.{DepositMoney, OpenAccount} alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited + alias Commanded.MockedApp alias Commanded.UUID describe "concurrency error" do diff --git a/test/aggregates/aggregate_lifespan_test.exs b/test/aggregates/aggregate_lifespan_test.exs index d9af19c2..78183bcc 100644 --- a/test/aggregates/aggregate_lifespan_test.exs +++ b/test/aggregates/aggregate_lifespan_test.exs @@ -3,9 +3,11 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do alias Commanded.Aggregates.{Aggregate, DefaultLifespanRouter, LifespanAggregate, LifespanRouter} alias Commanded.Aggregates.LifespanAggregate.{Command, Event} + alias Commanded.DefaultApp + alias Commanded.EventStore + alias Commanded.Registration alias Commanded.TestSupport.Factory - alias Commanded.{DefaultApp, EventStore} - alias Commanded.{Registration, UUID} + alias Commanded.UUID describe "aggregate lifespan" do setup do diff --git a/test/aggregates/execute_command_test.exs b/test/aggregates/execute_command_test.exs index 889d9cf6..4e8d8dd1 100644 --- a/test/aggregates/execute_command_test.exs +++ b/test/aggregates/execute_command_test.exs @@ -5,13 +5,15 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do alias Commanded.Aggregates.{Aggregate, ExecutionContext} alias Commanded.Commands.{TimeoutAggregateRoot, TimeoutCommand, TimeoutCommandHandler} + alias Commanded.DefaultApp alias Commanded.ExampleDomain.{BankAccount, BankApp, OpenAccountHandler} alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened alias Commanded.Helpers.Wait + alias Commanded.Registration alias Commanded.TestSupport.RetryStopOnceAggregate alias Commanded.TestSupport.RetryStopOnceAggregate.Command, as: RetryStopOnceCommand - alias Commanded.{DefaultApp, Registration, UUID} + alias Commanded.UUID defmodule CrashCommand do defstruct [:uuid] diff --git a/test/application/telemetry_test.exs b/test/application/telemetry_test.exs index b1d685b4..37cbe807 100644 --- a/test/application/telemetry_test.exs +++ b/test/application/telemetry_test.exs @@ -1,8 +1,8 @@ defmodule Commanded.Application.TelemetryTest do use ExUnit.Case - alias Commanded.DefaultApp alias Commanded.Commands.{TimeoutCommand, TimeoutRouter} + alias Commanded.DefaultApp alias Commanded.Middleware.Commands.Fail alias Commanded.Middleware.Commands.IncrementCount alias Commanded.Middleware.Commands.RaiseError diff --git a/test/commands/correlation_causation_test.exs b/test/commands/correlation_causation_test.exs index f753f1da..922c1bf2 100644 --- a/test/commands/correlation_causation_test.exs +++ b/test/commands/correlation_causation_test.exs @@ -5,9 +5,9 @@ defmodule Commanded.Commands.CorrelationCasuationTest do alias Commanded.Commands.OpenAccountBonusHandler alias Commanded.EventStore - alias Commanded.ExampleDomain.BankApp alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount, WithdrawMoney} alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited + alias Commanded.ExampleDomain.BankApp alias Commanded.ExampleDomain.BankRouter alias Commanded.Helpers.CommandAuditMiddleware alias Commanded.UUID diff --git a/test/event/event_handler_batch_test.exs b/test/event/event_handler_batch_test.exs index 13465cba..9ff96ac2 100644 --- a/test/event/event_handler_batch_test.exs +++ b/test/event/event_handler_batch_test.exs @@ -6,11 +6,11 @@ defmodule Commanded.Event.EventHandlerBatchTest do alias Commanded.EventStore.Subscription # Test support code - alias Commanded.Helpers.EventFactory alias Commanded.Event.{BatchHandler, ErrorHandlingBatchHandler} alias Commanded.Event.ErrorAggregate.Events.ErrorEvent alias Commanded.Event.EventHandlerBatchTest.MockAdapter alias Commanded.Event.ReplyEvent + alias Commanded.Helpers.EventFactory describe "batch handling" do test "should receive events in batches" do diff --git a/test/event/event_handler_telemetry_test.exs b/test/event/event_handler_telemetry_test.exs index bcd085ba..a5568fa1 100644 --- a/test/event/event_handler_telemetry_test.exs +++ b/test/event/event_handler_telemetry_test.exs @@ -5,10 +5,10 @@ defmodule Commanded.Event.EventHandlerTelemetryTest do alias Commanded.Event.Handler alias Commanded.EventStore.Subscription - alias Commanded.Helpers.EventFactory alias Commanded.Event.{BatchHandler, EchoHandler, ErrorHandlingBatchHandler} alias Commanded.Event.EventHandlerTelemetryTest.MockAdapter alias Commanded.Event.ReplyEvent + alias Commanded.Helpers.EventFactory setup do attach_telemetry() diff --git a/test/event_store/adapters/in_memory/in_memory_test.exs b/test/event_store/adapters/in_memory/in_memory_test.exs index 9aaa7b9e..3b8775c8 100644 --- a/test/event_store/adapters/in_memory/in_memory_test.exs +++ b/test/event_store/adapters/in_memory/in_memory_test.exs @@ -1,8 +1,8 @@ defmodule Commanded.EventStore.Adapters.InMemoryTest do use Commanded.EventStore.InMemoryTestCase - alias Commanded.EventStore.AdapterTestData alias Commanded.EventStore.Adapters.InMemory + alias Commanded.EventStore.AdapterTestData describe "reset!/0" do test "wipes all data from memory", %{event_store_meta: event_store_meta} do diff --git a/test/event_store/support/adapter_test_data.ex b/test/event_store/support/adapter_test_data.ex index 7fbf846c..34f2f577 100644 --- a/test/event_store/support/adapter_test_data.ex +++ b/test/event_store/support/adapter_test_data.ex @@ -1,10 +1,10 @@ defmodule Commanded.EventStore.AdapterTestData do alias Commanded.Event.Mapper - alias Commanded.Helpers.EventFactory alias Commanded.EventStore.SnapshotData alias Commanded.EventStore.TypeProvider alias Commanded.ExampleDomain.BankAccount alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened, MoneyDeposited} + alias Commanded.Helpers.EventFactory alias Commanded.UUID def build_opened_event(opts \\ []) do diff --git a/test/event_store/telemetry_test.exs b/test/event_store/telemetry_test.exs index ad6d8911..fd513853 100644 --- a/test/event_store/telemetry_test.exs +++ b/test/event_store/telemetry_test.exs @@ -5,8 +5,8 @@ defmodule Commanded.EventStore.TelemetryTest do alias Commanded.DefaultApp alias Commanded.EventStore - alias Commanded.EventStore.AdapterTestData alias Commanded.EventStore.Adapters.Mock, as: MockEventStore + alias Commanded.EventStore.AdapterTestData alias Commanded.EventStore.RecordedEvent alias Commanded.Middleware.Commands.IncrementCount alias Commanded.Middleware.Commands.RaiseError diff --git a/test/opentelemetry/aggregate_populate_test.exs b/test/opentelemetry/aggregate_populate_test.exs index 75606c4e..113081d9 100644 --- a/test/opentelemetry/aggregate_populate_test.exs +++ b/test/opentelemetry/aggregate_populate_test.exs @@ -34,7 +34,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do ] do handlers = :telemetry.list_handlers(event) - assert length(handlers) >= 1, + assert handlers != [], "Expected handler for event #{inspect(event)}" end end diff --git a/test/opentelemetry/aggregate_snapshot_test.exs b/test/opentelemetry/aggregate_snapshot_test.exs index 395a8015..95bd09a7 100644 --- a/test/opentelemetry/aggregate_snapshot_test.exs +++ b/test/opentelemetry/aggregate_snapshot_test.exs @@ -29,7 +29,7 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshotTest do ] do handlers = :telemetry.list_handlers(event) - assert length(handlers) >= 1, + assert handlers != [], "Expected handler for event #{inspect(event)}" end end diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs index ae5c4b5a..8e8f19a3 100644 --- a/test/opentelemetry/aggregate_test.exs +++ b/test/opentelemetry/aggregate_test.exs @@ -2,8 +2,8 @@ defmodule Commanded.OpenTelemetry.AggregateTest do use Commanded.OpenTelemetryCase, async: false use Commanded.MockEventStoreCase - alias Commanded.Aggregates.AggregateTelemetrySupport alias Commanded.Aggregates.{Aggregate, ExecutionContext} + alias Commanded.Aggregates.AggregateTelemetrySupport alias Commanded.DefaultApp alias Commanded.Middleware.Commands.IncrementCount alias Commanded.Middleware.Commands.RaiseError @@ -872,12 +872,10 @@ defmodule Commanded.OpenTelemetry.AggregateTest do unsubscribe )a - for event <- event_store_events do - for suffix <- [:start, :stop, :exception] do - for handler <- :telemetry.list_handlers([:commanded, :event_store, event, suffix]) do - :telemetry.detach(handler.id) - end - end + for event <- event_store_events, + suffix <- [:start, :stop, :exception], + handler <- :telemetry.list_handlers([:commanded, :event_store, event, suffix]) do + :telemetry.detach(handler.id) end end end diff --git a/test/opentelemetry/event_store_test.exs b/test/opentelemetry/event_store_test.exs index e262f4c8..b99a804d 100644 --- a/test/opentelemetry/event_store_test.exs +++ b/test/opentelemetry/event_store_test.exs @@ -563,12 +563,10 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do end defp detach_handlers do - for event <- @events do - for suffix <- [:start, :stop, :exception] do - for handler <- :telemetry.list_handlers([:commanded, :event_store, event, suffix]) do - :telemetry.detach(handler.id) - end - end + for event <- @events, + suffix <- [:start, :stop, :exception], + handler <- :telemetry.list_handlers([:commanded, :event_store, event, suffix]) do + :telemetry.detach(handler.id) end end diff --git a/test/subscriptions/support/distributed_subscribers.ex b/test/subscriptions/support/distributed_subscribers.ex index 57979318..0f3ed29f 100644 --- a/test/subscriptions/support/distributed_subscribers.ex +++ b/test/subscriptions/support/distributed_subscribers.ex @@ -21,22 +21,7 @@ defmodule Commanded.Subscriptions.DistributedSubscribers do reply_to = self() for node <- nodes do - Node.spawn_link(node, fn -> - Logger.configure(level: :error) - - {:ok, _pid} = DistributedApp.start_link() - - for subscriber <- Enum.shuffle(all()) do - {:ok, _pid} = subscriber.start_link() - - # Sleep to allow subscribers to be distributed amongst all nodes - :timer.sleep(100) - end - - send(reply_to, {:started, node}) - - :timer.sleep(:infinity) - end) + Node.spawn_link(node, fn -> start_subscribers_on_node(node, reply_to) end) end for node <- nodes do @@ -48,14 +33,32 @@ defmodule Commanded.Subscriptions.DistributedSubscribers do reply_to = self() for node <- nodes do - Node.spawn_link(node, fn -> - subscriptions = - Subscriptions.all(DistributedApp) - |> Enum.map(fn {name, _module, _pid} -> name end) - |> Enum.sort() - - send(reply_to, {:subscriptions, node, subscriptions}) - end) + Node.spawn_link(node, fn -> query_subscriptions_on_node(node, reply_to) end) end end + + defp start_subscribers_on_node(node, reply_to) do + Logger.configure(level: :error) + + {:ok, _pid} = DistributedApp.start_link() + + for subscriber <- Enum.shuffle(all()) do + {:ok, _pid} = subscriber.start_link() + :timer.sleep(100) + end + + send(reply_to, {:started, node}) + + :timer.sleep(:infinity) + end + + defp query_subscriptions_on_node(node, reply_to) do + subscriptions = + DistributedApp + |> Subscriptions.all() + |> Enum.map(fn {name, _module, _pid} -> name end) + |> Enum.sort() + + send(reply_to, {:subscriptions, node, subscriptions}) + end end