commanded / commanded-extreme-adapter

Extreme event store adapter for Commanded

Commanded

Use Commanded to build your own Elixir applications following the CQRS/ES pattern.

Provides support for:

• Command registration and dispatch.
• Hosting and delegation to aggregates.
• Event handling.
• Long running process managers.

Commanded provides a solid technical foundation for you to build on. It allows you to focus on modelling your domain, the most important part of your app, creating a better application at a faster pace.

You can use Commanded with one of the following event stores for persistence:

• EventStore - Elixir library using Postgres for persistence.
• EventStoreDB - a stream database built for Event Sourcing.
• In-memory event store - included for test use only.

Please refer to the CHANGELOG for features, bug fixes, and any upgrade advice included for each release.

Requires Erlang/OTP v21.0 and Elixir v1.11 or later.

---

Sponsors

• View sponsors & backers

---

• Changelog
• Wiki
• What is CQRS/ES?
• Frequently asked questions
• Getting help
• Latest published Hex package & documentation

MIT License

---

This README and the following guides follow the master branch which may not be the currently published version.

Read the documentation for the latest published version of Commanded on Hex.

Overview

• Getting started
• Choosing an event store
  • PostgreSQL-based EventStore
  • Greg Young's Event Store
• Using Commanded
  • Aggregates
    • Example aggregate
    • Commanded.Aggregate.Multi
    • Aggregate state snapshots
  • Commands
    • Command handlers
    • Command dispatch and routing
      • Define aggregate identity
      • Multi-command registration
      • Dispatch timeouts
      • Dispatch consistency guarantee
      • Dispatch returning execution result
      • Aggregate lifespan
      • Composite command routers
    • Middleware
      • Example middleware
    • Composite command routers
  • Events
    • Domain events
    • Event handlers
      • Consistency guarantee
    • Upcasting events
  • Process managers
    • Example process manager
  • Supervision
  • Serialization
    • Default JSON serializer
    • Configuring JSON serialization
    • Decoding event structs
    • Using an alternative serialization format
    • Customising serialization
  • Read model projections
• Deployment
  • Single node deployment
  • Multi node cluster deployment
  • Multi node, but not clustered deployment
• Testing with Commanded
• Used in production?
• Example application
• Learn Commanded in 20 minutes
• Choosing an event store provider
• Tooling
• Contributing
• Need help?

---

Used in production?

Yes, see the companies using Commanded.

Example application

Conduit is an open source, example Phoenix 1.3 web application implementing the CQRS/ES pattern in Elixir. It was built to demonstrate the implementation of Commanded in an Elixir application for the Building Conduit book.

Learn Commanded in 20 minutes

Watch Bernardo Amorim introduce CQRS and event sourcing at Code Beam SF 2018. Including a tutorial on how to implement an Elixir application using these concepts with Commanded.

Contributing

Pull requests to contribute new or improved features, and extend documentation are most welcome.

Please follow the existing coding conventions, or refer to the Elixir style guide.

You should include unit tests to cover any changes. Run mix test to execute the test suite.

Contributors

Commanded exists thanks to the following people who have contributed.

• Adil Yarulin
• Alexandre de Souza
• Andrey Akulov
• Andrzej Sliwa
• Ben Smith
• Benjamin Moss
• Bernardo Amorim
• Brenton Annan
• Chris Brodt
• Chris Martin
• Christophe Juniet
• Danilo Silva
• Dave Lucia
• David Carlin
• Damir Vandic
• Danni Friedland
• Dilaksun Bavarajan
• Ernesto
• Fernando Mendes
• Florian Ebeling
• Henry Hazan
• JC
• Joan Zapata
• Joao Gilberto Moura
• João Thallis
• John Wilger
• Joseph Lozano
• Kian-Meng Ang
• Kok J Sam
• Leif Gensert
• Luís Ferreira
• Marcelo Dominguez
• Matt Doughty
• Matthew Boehlig
• Michael Herold
• Miguel Palhas
• Nigel Thorne
• Olafur Arason
• Paolo Laurenti
• Patrick Detlefsen
• Phil Chen
• Raphaël Lustin
• Štefan Ľupták
• Tobiasz Małecki
• Vladimir Drobyshevskiy
• Willy Wombat
• Yordis Prieto
• Yuri de Figueiredo
• Zven

Need help?

Please open an issue if you encounter a problem, or need assistance. You can also seek help in the #commanded channel in the official Elixir Slack.

commanded-extreme-adapter's Issues

Make work with latest extreme/eventstore

Can we make sure this works with the latest version of extreme/commanded and eventstore?

I think there are some issues with compilation.

`:stream_not_found` is not handled properly

It seems like {:error, :stream_not_found} is wrapped within a stream, causing the first pattern of https://github.com/slashdotdash/commanded/blob/e96628253b70bec7ded3200e5efb6c2e0a6ebf19/lib/commanded/aggregates/aggregate.ex#L104 to not match

iex(1)> Cqrs.Router.dispatch(%Cqrs.Commands.OpenAccount{account_number: "abc", initial_balance: 123})
[debug] starting aggregate process for `Cqrs.Domain.BankAccount` with UUID "abc"
{:error, :aggregate_execution_failed}
iex(2)> [error] GenServer {:aggregate_registry, "abc"} terminating
** (FunctionClauseError) no function clause matching in Commanded.Event.Mapper.map_from_recorded_event/1
    (commanded) lib/commanded/event/mapper.ex:33: Commanded.Event.Mapper.map_from_recorded_event({:error, :stream_not_found})
    (commanded) lib/commanded/aggregates/aggregate.ex:112: anonymous fn/1 in Commanded.Aggregates.Aggregate.rebuild_from_events/1
    (elixir) lib/stream.ex:495: anonymous fn/4 in Stream.map/2
    (elixir) lib/stream.ex:1259: Stream.do_resource/5
    (elixir) lib/stream.ex:1405: Enumerable.Stream.do_each/4
    (elixir) lib/stream.ex:785: Stream.do_transform/8
    (elixir) lib/enum.ex:1776: Enum.take/2
    (elixir) lib/stream.ex:642: anonymous fn/4 in Stream.take/2
Last message: {:"$gen_cast", {:populate_aggregate_state}}
State: %Commanded.Aggregates.Aggregate{aggregate_module: Cqrs.Domain.BankAccount, aggregate_state: nil, aggregate_uuid: "abc", aggregate_version: 0}
[error] Task #PID<0.415.0> started from #PID<0.408.0> terminating
** (stop) exited in: GenServer.call({:via, Registry, {:aggregate_registry, "abc"}}, {:execute_command, Cqrs.Handlers.OpenAccountHandler, :handle, %Cqrs.Commands.OpenAccount{account_number: "abc", initial_balance: 123}}, 5000)
    ** (EXIT) an exception was raised:
        ** (FunctionClauseError) no function clause matching in Commanded.Event.Mapper.map_from_recorded_event/1
            (commanded) lib/commanded/event/mapper.ex:33: Commanded.Event.Mapper.map_from_recorded_event({:error, :stream_not_found})
            (commanded) lib/commanded/aggregates/aggregate.ex:112: anonymous fn/1 in Commanded.Aggregates.Aggregate.rebuild_from_events/1
            (elixir) lib/stream.ex:495: anonymous fn/4 in Stream.map/2
            (elixir) lib/stream.ex:1259: Stream.do_resource/5
            (elixir) lib/stream.ex:1405: Enumerable.Stream.do_each/4
            (elixir) lib/stream.ex:785: Stream.do_transform/8
            (elixir) lib/enum.ex:1776: Enum.take/2
            (elixir) lib/stream.ex:642: anonymous fn/4 in Stream.take/2
    (elixir) lib/gen_server.ex:737: GenServer.call/3
    (elixir) lib/task/supervised.ex:85: Task.Supervised.do_apply/2
    (elixir) lib/task/supervised.ex:36: Task.Supervised.reply/5
    (stdlib) proc_lib.erl:247: :proc_lib.init_p_do_apply/3
Function: &Commanded.Aggregates.Aggregate.execute/5
    Args: ["abc", %Cqrs.Commands.OpenAccount{account_number: "abc", initial_balance: 123}, Cqrs.Handlers.OpenAccountHandler, :handle, 5000]

Old httpoison dependency

Getting this error:

Failed to use "httpoison" (versions 1.2.0 and 1.3.0) because
  commanded_extreme_adapter (version 0.5.0) requires ~> 0.11.1
  extreme (versions 0.13.1 and 0.13.2) requires ~> 1.2

More stringent dependency version on extreme

Instead of:

      {:extreme, "~> 0.11"},

we need to do:

      {:extreme, "~> 0.11.0"},

This is because with the first option, mix pulls down extreme 0.13 automatically, which depends on new elixir_uuid renamed library. Fixing to 0.11.0 makes it download extreme 0.11.

I fixed this in my fork. But you might want to fix it here if you don't want to accept the full PR for other stuff: #17

Store event data and metadata as JSON

Issue originally raised in Commanded by @drozzy (#81).

Need to investigate whether it is possible to persist event data & metadata as JSON when using a JSON event serializer.

Example

Persisted events from Event Store:

{
    "correlationId": "ccf88746-06ba-459f-b93a-9ccce1cebdf1",
    "readerPosition": {
        "$s": {
            "$ce-account": 6
        }
    },
    "events": [
        {
            "eventStreamId": "account-333-121-568-3245",
            "eventType": "Elixir.Bank.Events.AccountOpened",
            "data": "{\"timestamp_utc\":1506096118,\"initial_balance\":0,\"client_id\":\"3324-john.oliver\",\"account_id\":\"333-121-568-3245\"}",
            "metadata": "{\"$correlationId\":\"7e083e9e-bcf6-4cc5-8cc8-47d84f14ec50\"}",
            "readerPosition": {
                "$s": {
                    "$ce-account": 0
                }
            }
        },
   ]
}

Possible solution

Set the data_content_type and metadata_content_type to 1 for Extreme's ExMsg.NewEvent message (source).

Event handler not receiving event

I have an umbrella app that contains 2 apps. I'm using Elixir 1.8.1, which I understand isn't officially supported by extreme and would be happy to switch back if I need to.

The first app receives HTTP input and generates a command. The aggregate accepts the command and generates an event that is published to the event store. The aggregate then processes the event to update its internal state.

I am attempting to configure an event handler in the second app to process the same event and perform some other work (just log for now). I can see that the event handler is starting up and subscribing to the event store, but the events don't come in.

My event handler:

defmodule Driven.Worker do
  @moduledoc """
  Listens for events and acts on them
  """

  use Commanded.Event.Handler, name: "DrivenHandler"
  alias Api.Events.UserRegistered
  require Logger

  def handle(%UserRegistered{}, _metadata) do
    Logger.info("Handling Driven event")
    :ok
  end
end

I've tracked the event down to Extreme.Subscription.def handle_cast/2 when the status is :subscribed. I can't tell where the message goes from there.

I'm going to keep looking into it, but I'd appreciate any pointers anyone might have.

Stream prefix not working?

I'm using commanded_extreme_adapter ~> 0.4 and for some reason configuring stream_prefix doesn't work:

config :commanded_extreme_adapter,
  serializer: Commanded.Serialization.JsonSerializer,
  stream_prefix: "blah"

Still all my streams are created with prefix "commanded".

Allow configuration of the subscriber max count for persistent subscriptions

A persistent subscription to Event Store sets the maximum number of allowed TCP subscribers (subscriber_max_count) value to 1 by default. Instead it should be made configurable.

Commanded.EventStore.Adapters.Extreme.EventStore missing from process registry

I am trying to run a negligible variation of the example-domain (https://github.com/slashdotdash/commanded/tree/e96628253b70bec7ded3200e5efb6c2e0a6ebf19/test/example_domain) test from commanded using the extreme adapter, but am getting the following error:

iex(1)> Cqrs.Router.dispatch(%Cqrs.Commands.OpenAccount{account_number: "abc", initial_balance: 123})
[debug] starting aggregate process for `Cqrs.Domain.BankAccount` with UUID "abc"
{:error, :aggregate_execution_failed}
iex(2)> [error] GenServer {:aggregate_registry, "abc"} terminating
** (stop) exited in: GenServer.call(Commanded.EventStore.Adapters.Extreme.EventStore, {:execute, %Extreme.Msg.ReadStreamEvents{event_stream_id: "commandeddev-abc", from_event_number: 0, max_count: 100, require_master: false, resolve_link_tos: true}}, 5000)
    ** (EXIT) no process: the process is not alive or there's no process currently associated with the given name, possibly because its application isn't started
    (elixir) lib/gen_server.ex:729: GenServer.call/3
    lib/extreme.ex:265: Commanded.EventStore.Adapters.Extreme.execute_read/5
    lib/extreme.ex:58: anonymous fn/3 in Commanded.EventStore.Adapters.Extreme.stream_forward/3
    (elixir) lib/stream.ex:1257: Stream.do_resource/5
    (elixir) lib/stream.ex:1405: Enumerable.Stream.do_each/4
    (elixir) lib/stream.ex:785: Stream.do_transform/8
    (elixir) lib/enum.ex:1776: Enum.take/2
    (elixir) lib/stream.ex:642: anonymous fn/4 in Stream.take/2
Last message: {:"$gen_cast", {:populate_aggregate_state}}
State: %Commanded.Aggregates.Aggregate{aggregate_module: Cqrs.Domain.BankAccount, aggregate_state: nil, aggregate_uuid: "abc", aggregate_version: 0}
[error] Task #PID<0.324.0> started from #PID<0.321.0> terminating
** (stop) exited in: GenServer.call({:via, Registry, {:aggregate_registry, "abc"}}, {:execute_command, Cqrs.Handlers.OpenAccountHandler, :handle, %Cqrs.Commands.OpenAccount{account_number: "abc", initial_balance: 123}}, 5000)
    ** (EXIT) exited in: GenServer.call(Commanded.EventStore.Adapters.Extreme.EventStore, {:execute, %Extreme.Msg.ReadStreamEvents{event_stream_id: "commandeddev-abc", from_event_number: 0, max_count: 100, require_master: false, resolve_link_tos: true}}, 5000)
        ** (EXIT) no process: the process is not alive or there's no process currently associated with the given name, possibly because its application isn't started
    (elixir) lib/gen_server.ex:737: GenServer.call/3
    (elixir) lib/task/supervised.ex:85: Task.Supervised.do_apply/2
    (elixir) lib/task/supervised.ex:36: Task.Supervised.reply/5
    (stdlib) proc_lib.erl:247: :proc_lib.init_p_do_apply/3
Function: &Commanded.Aggregates.Aggregate.execute/5
    Args: ["abc", %Cqrs.Commands.OpenAccount{account_number: "abc", initial_balance: 123}, Cqrs.Handlers.OpenAccountHandler, :handle, 5000]

[Subscription] Event handlers receive no events

As it stands commanded event handlers do not seem to receive any events.

Test event handler

I've implemented a test handler which listens for any event and logs it to the console:

defmodule TestEventHandler do
  @behaviour Commanded.Event.Handler

  require Logger

  def start_link do
    Commanded.Event.Handler.start_link("test_handler", __MODULE__)
  end

  def handle(any_event, _metadata) do
    Logger.info "EVENT: #{inspect any_event}"

    :ok
  end
end

This test handler is being started by a simple Supervisor which in turn is being started by the application. I've verified that the event handler is indeed running using :observer.start() and checking the Persistent Subscriptions tab from the EventStore web UI.

EventStore configuration

Furthermore I've verified that the EventStore is indeed running with the necessary configuration options; see the output from the EventStore start:

[00001,01,14:25:30.844] 
ES VERSION:               4.0.1.0 (HEAD/5f53330a4cc31fc6eb8b337cc630038b40a4f47a, Wed, 12 Apr 2017 15:28:32 +0200)
[00001,01,14:25:30.881] OS:                       Linux (Unix 4.9.36.0)
[00001,01,14:25:30.889] RUNTIME:                  4.6.2 (Stable 4.6.2.16/ac9e222) (64-bit)
[00001,01,14:25:30.889] GC:                       2 GENERATIONS
[00001,01,14:25:30.889] LOGS:                     /var/log/eventstore
[00001,01,14:25:30.903] MODIFIED OPTIONS:

        START STANDARD PROJECTIONS: true (Environment Variable)
        CLUSTER GOSSIP PORT:      2112 (Environment Variable)
        RUN PROJECTIONS:          All (Environment Variable)
        INT IP:                   0.0.0.0 (Config File)
        EXT IP:                   0.0.0.0 (Config File)
        INT HTTP PREFIXES:        http://*:2112/ (Config File)
        EXT HTTP PREFIXES:        http://*:2113/ (Config File)
        ADD INTERFACE PREFIXES:   false (Config File)

Testing

After dispatching a few test commands to commanded I've verified that the events are being written to EventStore (checked the web UI).

It seems that the subscription of the event handler never receives any kind of events.

While trying to boil down the problem, I've run the tests from the adapter which resulted in some failed tests (Output in a comment below).

How to proceed?

Can you give some input on the issue? We tried to track down the culprit in our team but as it stands we are not sure where to start. We would be happy to fix any problems and submit a pull request, as soon as we know what needs to be done.

Remove "prefix-" from stream names

It would be nice not to have aggregates prefixed with a commanded global-level prefix.

E.g. instead of "cmd-bankaccount-123" to have just "bankaccount-123"

wrong kind of value returned from append_to_stream

I get the following error. It seems that append_to_stream returns an error triple, but only error tuples are allowed.

** (WithClauseError) no with clause matching: {:error, :CommitTimeout, %Extreme.Msg.WriteEventsCompleted{commit_position: 0, first_event_number: -2147483648, last_event_number: -2147483648, message: "Commit phase timeout.", prepare_position: -2147483648, result: :CommitTimeout}}
    (commanded) lib/commanded/aggregates/aggregate.ex:458: Commanded.Aggregates.Aggregate.persist_events/4
    (commanded) lib/commanded/aggregates/aggregate.ex:433: Commanded.Aggregates.Aggregate.execute_command/2
    (commanded) lib/commanded/aggregates/aggregate.ex:192: Commanded.Aggregates.Aggregate.handle_call/3
    (stdlib) gen_server.erl:661: :gen_server.try_handle_call/4
    (stdlib) gen_server.erl:690: :gen_server.handle_msg/6
    (stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3

Compilation Error

Resolving Hex dependencies...
Dependency resolution completed:
  certifi 1.2.1
  commanded 0.14.0
  commanded_ecto_projections 0.6.0
  commanded_extreme_adapter 0.2.0
  connection 1.0.4
  cowboy 1.1.2
  cowlib 1.0.2
  db_connection 1.1.2
  decimal 1.4.0
  ecto 2.2.6
  exprotobuf 1.2.9
  extreme 0.9.1
  file_system 0.2.1
  gettext 0.13.1
  gpb 3.27.7
  hackney 1.8.6
  httpoison 0.11.2
  idna 5.0.2
  mariaex 0.8.3
  metrics 1.0.1
  mime 1.1.0
  mimerl 1.0.2
  nadia 0.4.2
  phoenix 1.3.0
  phoenix_ecto 3.3.0
  phoenix_html 2.10.4
  phoenix_live_reload 1.1.2
  phoenix_pubsub 1.0.2
  plug 1.4.3
  poison 3.1.0
  poolboy 1.5.1
  postgrex 0.13.3
  ranch 1.3.2
  ssl_verify_fun 1.1.1
  table_rex 0.10.0
  unicode_util_compat 0.2.0
  uuid 1.1.8
➜  mccool-bot git:(master) ✗ mix deps.compile commanded_extreme_adapter
==> commanded_extreme_adapter
Compiling 4 files (.ex)

== Compilation error in file lib/extreme.ex ==
** (UndefinedFunctionError) function Commanded.EventStore.TypeProvider.__using__/1 is undefined or private
    Commanded.EventStore.TypeProvider.__using__([])
    lib/extreme.ex:14: (module)
    (stdlib) erl_eval.erl:670: :erl_eval.do_apply/6
could not compile dependency :commanded_extreme_adapter, "mix compile" failed. You can recompile this dependency with "mix deps.compile commanded_extreme_adapter", update it with "mix deps.update commanded_extreme_adapter" or clean it with "mix deps.clean commanded_extreme_adapter"

Hi there, can you help me understand why I'm having this compilation error?

:subscription_already_exists error when event handler returns error

I'm using this adapter, and failing an event handler gives the following error (I've removed the specific application details).
It looks like the subscription is not removed properly and when my supervisor restarts the handler it tries to subscribe again, and fails since the subscription is already there. Not sure if this is a problem with this adapter or with the event handler module itself.

[error] MyEventHandler failed to handle event %Commanded.EventStore.RecordedEvent{causation_id: "82df5f23-54ab-48ac-a99e-1104b856d00c", correlation_id: "4e50bfec-0c4a-48f3-9bca-188ebde62914", created_at: ~N[2018-12-03 09:29:26.341], data: %MyData{}, event_id: "cc1dc16a-e48b-45d9-b2a8-0ae440e1187c", event_number: 28, event_type: "MyEventType", metadata: %{}, stream_id: "my_stream_prefix-my_id", stream_version: 28} due to: %MyError{}
[warn]  MyEventHandler has requested to stop: %MyError{}
[error] GenServer {Commanded.Registration.LocalRegistry, {Commanded.Event.Handler, "MyEventHandler"}} terminating
** (stop) %MyError{}
Last message: {:events, [%Commanded.EventStore.RecordedEvent{causation_id: "82df5f23-54ab-48ac-a99e-1104b856d00c", correlation_id: "4e50bfec-0c4a-48f3-9bca-188ebde62914", created_at: ~N[2018-12-03 09:29:26.341], data: %MyEvent{}, event_id: "cc1dc16a-e48b-45d9-b2a8-0ae440e1187c", event_number: 28, event_type: "MyEvent", metadata: %{}, stream_id: "omy_stream_prefix-my_id", stream_version: 28}]}
State: %Commanded.Event.Handler{consistency: :strong, handler_module: MyEventHandler, handler_name: "MyEventHandler", last_seen_event: nil, subscribe_from: :origin, subscription: #PID<0.290.0>}
[debug] Extreme event store subscription "MyEventHandler" down due to: %MyError{}
[info]  Stopping persistent subscription "$ce-myprefix::MyEventHandler" as subscriber is down due to: {:shutdown, :subscriber_shutdown}
[error] GenServer {Commanded.Registration.LocalRegistry, {Commanded.Event.Handler, "MyEventHandler"}} terminating
** (MatchError) no match of right hand side value: {:error, :subscription_already_exists}
    (commanded) lib/commanded/event/handler.ex:417: Commanded.Event.Handler.subscribe_to_all_streams/1
    (commanded) lib/commanded/registration/registration.ex:368: Commanded.Event.Handler.handle_cast/2
    (stdlib) gen_server.erl:637: :gen_server.try_dispatch/4
    (stdlib) gen_server.erl:711: :gen_server.handle_msg/6
    (stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
Last message: {:"$gen_cast", :subscribe_to_events}
State: %Commanded.Event.Handler{consistency: :strong, handler_module: MyEventHandler, handler_name: "MyEventHandler", last_seen_event: nil, subscribe_from: :origin, subscription: nil}

Support gRPC client

This might not make sense given this package name (The new gPRC client is named Spear), but is there any plan to support in the near future the new gRPC interface of EventstoreDB?
If that's the case, is there anything I can do to help?