Packages
commanded
0.15.0
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.3
1.4.2
1.4.1
1.4.0
1.4.0-rc.0
1.3.1
1.3.0
1.2.0
1.1.1
1.1.0
1.0.1
1.0.0
1.0.0-rc.1
1.0.0-rc.0
0.19.1
0.19.0
0.18.1
0.18.0
0.17.5
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.0
0.16.0-rc.1
0.16.0-rc.0
0.15.1
0.15.0
0.14.0
0.14.0-rc.0
0.13.0
0.12.0
0.11.0
0.10.0
0.9.0
0.8.5
0.8.4
0.8.3
0.8.1
0.8.0
0.7.1
0.6.2
0.6.1
0.6.0
0.4.0
0.3.1
0.3.0
0.2.1
0.2.0
0.1.0
Use Commanded to build your own Elixir applications following the CQRS/ES pattern.
Current section
Files
Jump to
Current section
Files
test/process_managers/process_router_process_pending_events_test.exs
defmodule Commanded.ProcessManagers.ProcessRouterProcessPendingEventsTest do
use Commanded.StorageCase
import Commanded.Assertions.EventAssertions
import Commanded.Enumerable
alias Commanded.EventStore
alias Commanded.Helpers.Wait
alias Commanded.ProcessManagers.{ExampleRouter,ExampleProcessManager,ProcessRouter,ProcessManagerInstance}
alias Commanded.ProcessManagers.ExampleAggregate.Commands.{Error,Publish,Start}
alias Commanded.ProcessManagers.ExampleAggregate.Events.{Interested,Started,Stopped,Uninterested}
test "should start process manager instance and successfully dispatch command" do
aggregate_uuid = UUID.uuid4
{:ok, process_router} = ExampleProcessManager.start_link()
:ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
# dispatch command to publish multiple events and trigger dispatch of the stop command
:ok = ExampleRouter.dispatch(%Publish{aggregate_uuid: aggregate_uuid, interesting: 10, uninteresting: 1})
assert_receive_event Stopped, fn event ->
assert event.aggregate_uuid == aggregate_uuid
end
events = EventStore.stream_forward(aggregate_uuid) |> Enum.to_list()
assert pluck(events, :data) == [
%Started{aggregate_uuid: aggregate_uuid},
%Interested{aggregate_uuid: aggregate_uuid, index: 1},
%Interested{aggregate_uuid: aggregate_uuid, index: 2},
%Interested{aggregate_uuid: aggregate_uuid, index: 3},
%Interested{aggregate_uuid: aggregate_uuid, index: 4},
%Interested{aggregate_uuid: aggregate_uuid, index: 5},
%Interested{aggregate_uuid: aggregate_uuid, index: 6},
%Interested{aggregate_uuid: aggregate_uuid, index: 7},
%Interested{aggregate_uuid: aggregate_uuid, index: 8},
%Interested{aggregate_uuid: aggregate_uuid, index: 9},
%Interested{aggregate_uuid: aggregate_uuid, index: 10},
%Uninterested{aggregate_uuid: aggregate_uuid, index: 1},
%Stopped{aggregate_uuid: aggregate_uuid},
]
Wait.until fn ->
# process instance should be stopped
assert ProcessRouter.process_instance(process_router, aggregate_uuid) == {:error, :process_manager_not_found}
# process state snapshot should be deleted
assert EventStore.read_snapshot("example_process_manager-#{aggregate_uuid}") == {:error, :snapshot_not_found}
end
end
test "should ignore uninteresting events" do
aggregate_uuid = UUID.uuid4
{:ok, process_router} = ExampleProcessManager.start_link()
:ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
# dispatch commands to publish a mix of interesting and uninteresting events for the process router
:ok = ExampleRouter.dispatch(%Publish{aggregate_uuid: aggregate_uuid, interesting: 0, uninteresting: 2})
:ok = ExampleRouter.dispatch(%Publish{aggregate_uuid: aggregate_uuid, interesting: 0, uninteresting: 2})
:ok = ExampleRouter.dispatch(%Publish{aggregate_uuid: aggregate_uuid, interesting: 10, uninteresting: 0})
assert_receive_event Stopped, fn event ->
assert event.aggregate_uuid == aggregate_uuid
end
events =
aggregate_uuid
|> EventStore.stream_forward()
|> Enum.to_list()
assert pluck(events, :data) == [
%Started{aggregate_uuid: aggregate_uuid},
%Uninterested{aggregate_uuid: aggregate_uuid, index: 1},
%Uninterested{aggregate_uuid: aggregate_uuid, index: 2},
%Uninterested{aggregate_uuid: aggregate_uuid, index: 1},
%Uninterested{aggregate_uuid: aggregate_uuid, index: 2},
%Interested{aggregate_uuid: aggregate_uuid, index: 1},
%Interested{aggregate_uuid: aggregate_uuid, index: 2},
%Interested{aggregate_uuid: aggregate_uuid, index: 3},
%Interested{aggregate_uuid: aggregate_uuid, index: 4},
%Interested{aggregate_uuid: aggregate_uuid, index: 5},
%Interested{aggregate_uuid: aggregate_uuid, index: 6},
%Interested{aggregate_uuid: aggregate_uuid, index: 7},
%Interested{aggregate_uuid: aggregate_uuid, index: 8},
%Interested{aggregate_uuid: aggregate_uuid, index: 9},
%Interested{aggregate_uuid: aggregate_uuid, index: 10},
%Stopped{aggregate_uuid: aggregate_uuid},
]
Wait.until fn ->
# process instance should be stopped
assert ProcessRouter.process_instance(process_router, aggregate_uuid) == {:error, :process_manager_not_found}
end
end
test "should ignore past events when starting subscription from current" do
aggregate_uuid = UUID.uuid4
:ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
:ok = ExampleRouter.dispatch(%Publish{aggregate_uuid: aggregate_uuid, interesting: 4, uninteresting: 0})
wait_for_event Interested, fn event -> event.index == 4 end
{:ok, process_router} = ExampleProcessManager.start_link(start_from: :current)
assert ProcessRouter.process_instances(process_router) == []
:ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
:ok = ExampleRouter.dispatch(%Publish{aggregate_uuid: aggregate_uuid, interesting: 6, uninteresting: 0})
wait_for_event Interested, fn event -> event.index == 6 end
Wait.until(fn ->
process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
refute process_instance == {:error, :process_manager_not_found}
%{items: items} = ProcessManagerInstance.process_state(process_instance)
assert items == [1, 2, 3, 4, 5, 6]
end)
end
test "should stop process router when handling event errors" do
aggregate_uuid = UUID.uuid4
{:ok, process_router} = ExampleProcessManager.start_link()
Process.unlink(process_router)
ref = Process.monitor(process_router)
:ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
:ok = ExampleRouter.dispatch(%Error{aggregate_uuid: aggregate_uuid})
# should shutdown process
assert_receive {:DOWN, ^ref, _, _, _}
end
end