See the issues.
commanded / recipes Goto Github PK
View Code? Open in Web Editor NEWCommanded recipes
Commanded recipes
See the issues.
How to test event handler side effects.
defmodule CustomerPromotionHandler do
use Commanded.Event.Handler,
application: ExampleApp,
name: __MODULE__
def handle(%CustomerPromoted{} = event, _metadata) do
%CustomerPromoted{customer_id: customer_id} = event
Notification.send_promotion(customer_id)
end
end
Following the unit testing matrix outlined by Sandi Metz in her talk The Magic Tricks of Testing), to test side affects in an event handler you could use a mock to verify the expected external command is called. You don’t test the actual side effects when unit testing the handler, but you do verify they are called.
You can either call the event handler’s handle/2
function directly or cause the event to be produced (via command dispatch or appending the event to a stream) and afterwards verify the side-affect was called.
defmodule CustomerPromotionTest do
use ExUnit.Case
import Mox
setup :verify_on_exit!
test "`send_promotion/1` is called by promotion handler" do
reply_to = self()
NotificationMock.expect(:send_promotion, customer_id ->
send(reply_to, {:send_promotion, customer_id})
:ok
end)
# Start handler process and dispatch command to produce trigger event
start_supervised!(CustomerPromotionHandler)
:ok = MyApp.dispatch(command)
assert_receive {:send_promotion, ^expected_customer_id}
# Or call handler directly with an event (doesn’t require the handler process to be started)
:ok = CustomerPromotionHandler.handle(event, metadata)
end
end
With Mox you must define a behaviour which can then be mocked or stubbed at runtime.
defmodule NotificationBehaviour do
@callback send_promotion(non_neg_integer()) :: :ok | {:error, any()}
end
It's often useful to define a facade module for the behaviour which can substitute the mock at runtime or compile time. In this example the implementation module (real or mock) is determined at compile time from application config:
defmodule Notification do
@behaviour NotificationBehaviour
@implementation Application.get_env(:myapp, __MODULE__, Notification.DefaultImpl)
defdelegate send_promotion(customer_id), to: @implementation
end
defmodule Notification.DefaultImpl do
@behaviour NotificationBehaviour
def send_promotion(customer_id) do
# Actual implementation goes here ...
end
end
The mock must be defined in the test support:
# test/support/mocks.ex
Mox.defmock(NotificationMock, for: NotificationBehaviour)
Then configured for the test environment:
# config/test.exs
config :my_app, Notification, NotificationMock
With the above setup the Notification.DefaultImpl
module containing the actual implementation will be used in non-test environments and the mock will be used during tests. This allows the side effect call (send_promotion/1
) to be tested, without actually causing the side effect. It also allows you to easily test failures by returning a different response from the mock call (e.g. {:error, failed_to_send}
).
To test the event handler and its side effects you could use an integration style test and would replace the side effect module mock with the actual implementation. You could use Mox’s stub_with(NotificationMock, Notification.DefaultImpl)
which forwards all calls to the mock to the actual implementation. This ensures the contract is valid between the caller and the target module.
How to deal with streams which have thousands of events and/or are long running.
Use the approach taken by accountants where at the end of each financial year the current ledger is closed and the closing balances are taken as the starting balance into the next period. This can be used to limit the duration or length of event streams by closing the old stream and migrating to a new stream with any required state carried over as an initial event in the new stream.
How to deserialize events to their Elixir types.
Commanded uses JSON as the event data and metadata serialization format by default. This means that type information is lost when serializing Elixir structs to JSON and requires one to implement the Commanded.Serialization.JsonDecoder
to decode strings to their actual Elixir type, such as DateTime
.
One possible option for automatic deserialization would be to provide a typespec for the domain event structs and use the defined types to deserialize events. The Spec library provides structure decoding from a typespec and could be used via a custom event store serializer.
How to comply with GDPR regulations which require personally identifiable information (PII) to be removed upon request. This is problematic with immutable, persistent event streams.
There are three possible solutions:
PII data is stored encrypted in events and decrypted on read. The encryption key can later be "forgotten" to prevent decryption. Not being able to read the PII data is equivalent to having deleted the data.
Use a mutable data store, such as a SQL database, for all PII data and only store a reference to the external data store within your events. PII data must be read from the external store when events are read. PII data can be deleted from the external store. Attempting to read deleted PII data will return an appropriate error to indicate it has been removed.
Allow events or streams containing PII to be updated or deleted.
Note: EventStore has support for hard deleting streams.
How to enrich commands before or during dispatch with external data. This data may be looked up from a read model, another service, or an external third party API.
A router middleware defining an enrichment protocol is one possible solution. Here's an example gist with middleware for command enrichment.
Note processing by an aggregate instance during comand execution blocks it from handling any other commands. Whereas router middleware is executed by the dispatching process, before the aggregate instance is called (and is non-blocking to other processes).
How to write a unit test for an aggregate.
Use Commanded.AggregateCase
ExUnit case template to simplify aggregate unit testing.
How to enrich notification events to event-carried state transfer events.
Event enrichment allows "thin" domain events, but have handlers receive enriched "fat" events.
How to use a queue to handle incoming commands to limit how many commands are concurrently processed or to scale load amongst a cluster of nodes.
Record incoming commands in an event store stream and use one or more handlers to process the requests.
How do I use a different event type name provider than the default which uses the Elixir module name?
For example, use account_closed
instead of Elixir.MyApp.Account.Events.AccountClosed
How to define and run business focused acceptance tests.
How to deploy and run Commanded apps on a cluster of nodes.
Multi-node deployment may be using distributed Erlang (nodes connected together) or not (nodes not directly connected together). Commanded supports both approaches.
Use a separate event store for inter-service messaging.
Allows a context/service to keep its internal domain events private, but can publish enriched events for public consumption by other services.
Related to #1 (Event enrichment).
How to capture external events, such as those recorded by an IoT device.
Events can be directly appended to an event stream, using Commanded.EventStore
, without needing to be produced by an aggregate from a command. An event handler can be used to react to these external events. It may dispatch a command as a reaction to an event.
Define an integration event struct which can be used for any external event without needing to define a custom struct for each type.
defmodule MyApp.Integration.Event do
defstruct [:source, :type, :data]
end
The type of the external event will be indicated by the type
field. This allow loose coupling which is preferred when dealing with events from a third party, external system.
Append events to a stream (e.g. devices/device-1234
) using Commanded.EventStore.append_to_stream/4
and an integration Commanded Application instance.
source = "devices/device-1234"
events = [
%Commanded.EventStore.EventData{
event_type: "Elixir.MyApp.Integration.Event",
data: %MyApp.Integration.Event{
source: source,
type: "ThirdPartyService.SomeExternalEvent"
data: %{
"key1" => "value1",
"key2" => "value2"
}
},
metadata: %{}
}
]
:ok = Commanded.EventStore.append_to_stream(MyApp.Integration.App, source, :any_version, events)
As these are external events we are storing in raw format we can use :any_version
to skip concurrency control checking.
How to prevent duplicate commands from being executed by an aggregate.
Use a router middleware to record the identity of each successfully dispatched command and prevent duplicates. Record the causation_id
for each event applied to an aggregate and reject commands that have already be seen (caused an event to be recorded).
How to create a transient subscription to a stream, or all streams, and wait until an expected event is received.
Use Commanded.EventStore.subscribe/2
as shown in the following example.
alias Commanded.EventStore.RecordedEvent
defp receive_events(selector) do
receive do
{:events, events} ->
case reduce(events, selector) do
:cont -> receive_events(selector)
{:halt, result} -> result
end
end
end
defp reduce([], _selector), do: :cont
defp reduce([%RecordedEvent{data: data} | events], selector) do
case selector.(data) do
:cont -> reduce(events, selector)
{:halt, result} -> {:halt, result}
end
end
:ok = Commanded.EventStore.subscribe(MyApp.App, stream_uuid)
receive_events(fn
%AnEvent{} = event -> {:halt, event}
_event -> :cont
end)
Use :all
as the stream identity to subscribe to all events.
How to validate dispatched commands.
Use a data validation library such as Vex to validate the command (mandatory fields are present, correct data type, etc.).
How to test a process manager produces the correct commands and state?
Use the process manager's handle/2
and apply/2
functions directly by reducing a list of one or more domain events. The resultant list of commands and process manager state can then be asserted on.
events = [%Event1{}, %Event2{}]
initial_state = %MyProcessManager{}
{commands, state} =
Enum.reduce(events, {[], initial_state}, fn event, {commands, state} ->
new_commands = MyProcessManager.handle(state, event)
new_state = MyProcessManager.apply(state, event)
{commands ++ new_commands, new_state}
end)
assert match?([%Command1{}, %Command2{}], commands)
assert match?(%MyProcessManager{key: value}, state)
This type of test can be run using async: true
and will be very quick to execute as it does no IO.
How can I test a read model projector so that tests wait until events have been projected?
after_update/3
callback functionCommanded Ecto projections provides an after_update/3
callback function. This gets called after each event is projected. You can use the funtion to publish a notification whenever an event is projected, including the database changes.
Here’s an example using the Elixir/Erlang telemetry library.
In the projector:
def after_update(event, metadata, changes) do
:telemetry.execute(
[:projector],
%{system_time: System.system_time()},
%{event: event, metadata: metadata, changes: changes, projector: __MODULE__}
)
end
In a test:
setup do
reply_to = self()
:ok =
:telemetry.attach(
"test-handler",
[:projector],
fn event, measurements, metadata, reply_to ->
send(reply_to, {:telemetry, event, measurements, metadata})
end,
self()
)
end
Usage in test to wait:
assert_receive {:telemetry, [:projector], _measurements, %{event: event, metadata: metadata, changes: changes}}
This approach will ensure the test blocks until the projector has done its work and then you can run either use the changes from the Ecto.Multi or run a query to verify the database changes. You can also use pattern matching on the event from the telemetry metadata to wait for a particular event type.
The performance impact at runtime is negligible if there are no handlers attached to a telemetry event.
How to version events and upcast events between versions.
Only add or rename fields in an event. Implement the Commanded.Event.Upcaster
protocol for the old version of the event to migrate the structure to the new version.
How to link an existing event from one stream to another.
Use an event handler to link events of a given type into another stream.
The source event_id
is accessible via metadata:
def handle(%ProductSold{} = event, metadata) do
%ProductSold{user_id: user_id} = event
event_id = Map.get(metadata, :event_id)
MyApp.EventStore.link_to_stream(user_id, :any_version, [event_id])
end
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.