Packages
commanded
0.16.0-rc.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/subscriptions/subscriptions_test.exs
defmodule Commanded.SubscriptionsTest do
use ExUnit.Case
alias Commanded.EventStore.RecordedEvent
alias Commanded.Subscriptions
setup do
Subscriptions.reset()
end
describe "register event handler" do
test "should be registered" do
:ok = Subscriptions.register("handler1", :strong)
:ok = Subscriptions.register("handler2", :eventual)
:ok = Subscriptions.register("handler3", :strong)
assert Subscriptions.all() |> Enum.sort() == [{"handler1", self()}, {"handler3", self()}]
end
test "should ack event" do
:ok = Subscriptions.register("handler1", :strong)
:ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
assert Subscriptions.handled?("stream1", 1)
assert Subscriptions.handled?("stream1", 2)
end
test "should require all subscriptions to ack event" do
:ok = Subscriptions.register("handler1", :strong)
:ok = Subscriptions.register("handler2", :strong)
:ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
refute Subscriptions.handled?("stream1", 1)
:ok = Subscriptions.ack_event("handler2", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 1})
assert Subscriptions.handled?("stream1", 1)
refute Subscriptions.handled?("stream1", 2)
:ok = Subscriptions.ack_event("handler2", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
assert Subscriptions.handled?("stream1", 1)
assert Subscriptions.handled?("stream1", 2)
end
test "should ignore current process as handler" do
:ok = Subscriptions.register("handler1", :strong)
# current process should not block handler
assert Subscriptions.handled?("stream1", 1, [self()])
end
end
describe "notify subscribers" do
test "should immediately succeed when no registered handlers" do
assert :ok == Subscriptions.wait_for("stream1", 2)
end
test "should immediately succeed when waited event has already been ack'd" do
:ok = Subscriptions.register("handler", :strong)
:ok = Subscriptions.ack_event("handler", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 1})
:ok = Subscriptions.ack_event("handler", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
assert :ok == Subscriptions.wait_for("stream1", 2)
end
test "should immediately succeed when excluding handler process" do
:ok = Subscriptions.register("handler", :strong)
assert :ok == Subscriptions.wait_for("stream1", 2, [self()])
end
test "should succeed when waited event is ack'd" do
:ok = Subscriptions.register("handler", :strong)
wait_task = Task.async(fn ->
Subscriptions.wait_for("stream1", 2, [], 1_000)
end)
:ok = Subscriptions.ack_event("handler", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 1})
:ok = Subscriptions.ack_event("handler", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
assert :ok == Task.await(wait_task, 1_000)
end
test "should ignore events before requested" do
:ok = Subscriptions.register("handler", :strong)
:ok = Subscriptions.ack_event("handler", :strong, %RecordedEvent{event_number: 1})
assert {:error, :timeout} == Subscriptions.wait_for(2, 100)
end
test "should wait for all subscriptions to ack event" do
:ok = Subscriptions.register("handler1", :strong)
:ok = Subscriptions.register("handler2", :strong)
:ok = Subscriptions.register("handler3", :eventual)
refute Subscriptions.handled?("stream1", 2)
:ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 1})
:ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
refute Subscriptions.handled?("stream1", 2)
:ok = Subscriptions.ack_event("handler2", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 1})
:ok = Subscriptions.ack_event("handler2", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
assert Subscriptions.handled?("stream1", 2)
end
test "should allow subscriptions to skip events when ack" do
:ok = Subscriptions.register("handler", :strong)
refute Subscriptions.handled?("stream1", 2)
:ok = Subscriptions.ack_event("handler", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 4})
assert Subscriptions.handled?("stream1", 2)
end
end
describe "expire stream acks" do
test "should expire stale acks" do
:ok = Subscriptions.register("handler1", :strong)
:ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 1})
assert Subscriptions.handled?("stream1", 1)
pid = Process.whereis(Subscriptions)
send(pid, {:purge_expired_streams, 0})
refute Subscriptions.handled?("stream1", 1)
end
test "should not expire fresh acks" do
:ok = Subscriptions.register("handler1", :strong)
:ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 1})
assert Subscriptions.handled?("stream1", 1)
pid = Process.whereis(Subscriptions)
send(pid, {:purge_expired_streams, 1_000})
assert Subscriptions.handled?("stream1", 1)
end
end
end