Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 11 additions & 7 deletions lib/commanded/aggregates/aggregate.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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

Expand Down
21 changes: 14 additions & 7 deletions lib/commanded/commands/router.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand All @@ -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()}
Expand Down
29 changes: 12 additions & 17 deletions lib/commanded/projections/ecto.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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

Expand Down
2 changes: 1 addition & 1 deletion test/aggregates/aggregate_concurrency_test.exs
Original file line number Diff line number Diff line change
@@ -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
Expand Down
6 changes: 4 additions & 2 deletions test/aggregates/aggregate_lifespan_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion test/aggregates/execute_command_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down
2 changes: 1 addition & 1 deletion test/application/telemetry_test.exs
Original file line number Diff line number Diff line change
@@ -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
Expand Down
2 changes: 1 addition & 1 deletion test/commands/correlation_causation_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion test/event/event_handler_batch_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion test/event/event_handler_telemetry_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
2 changes: 1 addition & 1 deletion test/event_store/adapters/in_memory/in_memory_test.exs
Original file line number Diff line number Diff line change
@@ -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
Expand Down
2 changes: 1 addition & 1 deletion test/event_store/support/adapter_test_data.ex
Original file line number Diff line number Diff line change
@@ -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
Expand Down
2 changes: 1 addition & 1 deletion test/event_store/telemetry_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion test/opentelemetry/aggregate_populate_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion test/opentelemetry/aggregate_snapshot_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
12 changes: 5 additions & 7 deletions test/opentelemetry/aggregate_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
10 changes: 4 additions & 6 deletions test/opentelemetry/event_store_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
51 changes: 27 additions & 24 deletions test/subscriptions/support/distributed_subscribers.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Loading