merge

Author: Henry Hazan

README.md

@@ -1,29 +1,12 @@
-CQRS Eventsourcing Engine
-=========================
+CQRS Eventsourcing Workflow Engine
+==================================
 
 [![Join the chat at https://gitter.im/cqrs-engine/Lobby](https://badges.gitter.im/cqrs-engine/Lobby.svg)](https://gitter.im/cqrs-engine/Lobby?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 
 ### IMPORTANT
 Only the data structures are working, for testing and collecting feedbacks. 
 We believe that at the begining of December 2016, the framework will be usable.
 
-### IMPORTANT AGAIN 
-We started to align our ideias with @slashdotdash, and join forces. Please, see
-commanded issues. An example app started here: https://github.com/work-capital/ev_sim 
-
-
-
-### Thanks
-Special thanks to: 
-
-[burmajam](https://github.com/burmajam) for sharing the very 
-well written extreme driver to connect to Eventstore. 
-
-[slashdotdash](https://github.com/slashdotdash/commanded) for sharing the CQRS
-framework, where many parts of the code here are from his framework.
-
-
-### Motivation
 
 ### Pure functions data structures
 In the folder engine > types you will find the data structures, so you can write
@@ -41,10 +24,13 @@ your pure functions over them, under the "side effects" dimension.
 As aggregates listen for commands, process managers listen for events (sometimes commands also), and as aggregates emmits events, process managers dispatch commands.
 
 * pure functional data structures for aggregates and process managers
+* use monads (monadex) to simulate different business scenarios
 * one abstraction to implement side-effects
 * multiple data-stores
 * plugable message queue for publishing events
 * one gen_server implementation for aggregates and process managers
+* automatic process-manager creation based on correlation-ids (as suggested by Greg Young)
+* easy use of FSM on process managers
 
 ### Develop
 
@@ -56,9 +42,8 @@ Send events from the prompt:
 
 ```
 iex -S mix
+TODO: add example
 
-Engine.Bus.send_command(%{%Account.Command.CreateAccount{} | :id => "jsdf"})
-Engine.Bus.send_command(%{%Account.Command.DepositMoney{} | :id => "jsdf", :amount => 23})
 ```
 
 
@@ -71,30 +56,17 @@ docker run --name eventstore-node -it -p 2113:2113 -p 1113:1113 eventstore/event
 ```
 
 #### Resources
-Below you can see several resources I researched before writing this lib. 
+Below you can see several resources I researched before writing this lib.
+Special thanks for Ben Smith, where many ideas were copied from
+[commanded](https://github.com/slashdotdash/commanded) library.
 
+* [burmajam](https://github.com/burmajam) for sharing the very 
+well written extreme driver to connect to Eventstore. 
+* [slashdotdash](https://github.com/slashdotdash/commanded) for sharing the CQRS
+framework, where many parts of the code here are from his framework.
 * [cqrs-erlang](https://github.com/bryanhunter/cqrs-with-erlang) - A memory
   model using standard spawn functions CQRS in erlang. 
 * [gen-aggregate](https://github.com/burmajam/gen_aggregate/) - Macro for the
   aggregate structure, using buffers. 
 
 
-#### CQRS concepts
-
-http://softwareengineering.stackexchange.com/questions/157522/cqrs-event-sourcing-is-it-correct-that-commands-are-generally-communicated 
-
-
-If something sends a command, it entails expectation that it will be fulfilled. If you simply publish and hope that something somewhere picks it up and acts on it, there is no guarantee that this will be the case. By extrapolation, you also don't know if multiple handlers don't decide to act on a command, possibly resulting in the same change being applied more than once. 
-
-Events, on the other hand, are informative in nature, and it's reasonable to expect zero, two, or more components to be interested in a particular event. We don't really care in the scope of making the requested change. 
-
-**Example** 
-
-This could be compared to real life. If you have three children, walk into a room and simply shout "Clean the bathroom," you have no guarantee that someone will, and perhaphs if it won't be done twice (if you have obedient children that is ;-) You should fare better if you assign a specific child to do what you want done. 
-
-When that child finishes its job however, it's convenient if it shouts out "bathroom has been cleaned," so that everyone who wants to brush their teeth knows they can now do so. 
-
-
-
-
-

config/config.exs

@@ -1,66 +1,3 @@
-# This file is responsible for configuring your application
-# and its dependencies with the aid of the Mix.Config module.
 use Mix.Config
 import_config "#{Mix.env}.exs"
 
-#-------------------------------
-#  ENGINE
-#-------------------------------
-
-# Choose the event storage, can be Engine.Storage.Eventstore
-# or Engine.Storage.Postgres
-
-config :engine,
-  storage: Engine.Storage.Eventstore,
-  nodes: [:'master@localhost', :'slave1@localhost'],          # to use with SYN if we have many nodes
-  snapshot_period: 50
-
-#-------------------------------
-#  EXTREME [Eventstore Driver]
-#-------------------------------
-
-
-config :extreme, :event_store,
-  db_type: :node,
-  host: "localhost",
-  port: 1113,
-  username: "admin",
-  password: "changeit",
-  reconnect_delay: 2_000,
-  max_attempts: :infinity
-
-
-#-------------------------------
-#  LOGGER
-#-------------------------------
-
-# see config/dev.exs , config/test.exs, etc
-
-
-
-# This configuration is loaded before any dependency and is restricted
-# to this project. If another project depends on this project, this
-# file won't be loaded nor affect the parent project. For this reason,
-# if you want to provide default values for your application for
-# 3rd-party users, it should be done in your "mix.exs" file.
-
-# You can configure for your application as:
-#
-#     config :engine, key: :value
-#
-# And access this configuration in your application as:
-#
-#     Application.get_env(:engine, :key)
-#
-# Or configure a 3rd-party app:
-#
-#     config :logger, level: :info
-#
-
-# It is also possible to import configuration files, relative to this
-# directory. For example, you can emulate configuration per environment
-# by uncommenting the line below and defining dev.exs, test.exs and such.
-# Configuration from the imported file will override the ones defined
-# here (which is why it is important to import them last).
-#
-#     import_config "#{Mix.env}.exs"

config/dev.exs

@@ -1,13 +1,26 @@
 use Mix.Config
 
+#-------------------------------
+#  WORKFLOW
+#-------------------------------
+
+
+config :workflow,
+  adapter: Workflow.Extreme.Adapter
+
+#-------------------------------
+#  EXTREME [Eventstore Driver]
+#-------------------------------
+
 
-# config :mix_test_watch,
-# 	clear: true,
-#   tasks: [
-#   	"test"
-#     # "dogma"
-# ]
-#
+config :extreme, :event_store,
+  db_type: :node,
+  host: "localhost",
+  port: 1113,
+  username: "admin",
+  password: "changeit",
+  reconnect_delay: 2_000,
+  max_attempts: :infinity
 
 #-------------------------------
 #  LOGGER

config/test.exs

@@ -1,13 +1,27 @@
 use Mix.Config
 
 #-------------------------------
-#  ENGINE
+#  WORKFLOW
 #-------------------------------
 
 
-config :engine,
-  nodes: [:'master@localhost', :'slave1@localhost'],          # to use with SYN if we have many nodes
-  snapshot_period: 3
+config :workflow,
+  adapter: Workflow.Extreme.Adapter
+
+#-------------------------------
+#  EXTREME [Eventstore Driver]
+#-------------------------------
+
+
+config :extreme, :event_store,
+  db_type: :node,
+  host: "localhost",
+  port: 1113,
+  username: "admin",
+  password: "changeit",
+  reconnect_delay: 2_000,
+  max_attempts: :infinity
+
 
 #-------------------------------
 #  LOGGER
@@ -44,17 +58,3 @@ config :logger, :log_error,
   format: "$dateT$time $node $metadata[$level] $levelpad$message\n",
   level: :error
 
-#-------------------------------
-#  TEST WATCH
-#-------------------------------
-
-
-# see https://github.com/lpil/mix-test.watch how 
-# to exclude folders and files
-# config :mix_test_watch,
-# 	clear: true,     # clean the console
-#   tasks: [
-#   	"test"
-#     #"dogma"
-# ]
-

lib/adapter.ex

@@ -0,0 +1,30 @@
+defmodule Workflow.Adapter do
+  @moduledoc """
+  Implement this functions below to add a new data storage for your events and snapshots.
+  The persistence modules will retreive a pure list of events, ready for replay, instead of
+  dealing with localized messages. It's an adapter responsability to filter and answer only the
+  necessary data to be used inside Commanded
+  """
+
+  @type stream_id              :: String.t
+  @type start_version          :: number
+  @type read_event_batch_size  :: number
+  @type batch                  :: [struct()]
+  @type stream                 :: String.t     # The Stream ID
+  @type reason                 :: atom
+  @type expected_version       :: number
+  @type event_data             :: [struct()]
+
+
+
+  @doc "Load a batch of events from storage"
+  @callback read_stream_forward(stream_id, start_version, read_event_batch_size) ::
+    {:ok, batch} | {:error, reason}
+
+  @doc "Load a list of events from an specific position"
+  @callback append_to_stream(stream_id, expected_version, event_data) ::
+    :ok | {:error, reason}
+
+
+
+end

lib/container.ex

@@ -0,0 +1,86 @@
+defmodule Workflow.Container do
+  @moduledoc """
+  Genserver to hold Aggregates or Process Managers
+  """
+  use GenServer
+  require Logger
+  
+  alias Workflow.Container
+  alias Workflow.Persistence
+
+  defstruct [
+    module:       nil,
+    uuid:         nil,
+    data:         nil,
+    version:      nil
+  ]
+
+  ## API
+
+  def start_link(module, uuid) do
+    GenServer.start_link(__MODULE__, %Container{
+      module: module,
+      uuid:   uuid
+    })
+  end
+
+  def get_data(container), do:
+    GenServer.call(container, {:data})
+
+  def get_state(container), do:
+    GenServer.call(container, {:state})
+
+  def process_message(container, message), do:
+    GenServer.call(container, {:process_message, message})
+
+
+  ## CALLBACKS
+  
+  def init(%Container{} = state) do
+    GenServer.cast(self, {:restore})
+    {:ok, state}
+  end
+
+  def handle_call({:data}, _from, %Container{data: data} = state), do:
+    {:reply, data, state}
+
+  def handle_call({:state}, _from, %Container{} = state), do:
+    {:reply, state, state}
+
+  @doc "Replay the events from the eventstore db"
+  def handle_cast({:restore}, %Container{module: module} = state) do
+    state = Persistence.rebuild_from_events(%Container{state |
+      version: 0,
+      data: struct(module) # empty data structure to be filled
+    })
+    {:noreply, state}
+  end
+
+  @doc "Handle a command (for an aggregate) or an event (for the process manager)"
+  def handle_call({:process_message, message}, _from, %Container{} = state) do
+    {reply, state} = process(message, state)
+    {:reply, reply, state}
+  end
+
+  ## INTERNALS
+
+  defp process(message, 
+    %Container{uuid: uuid, version: expected_version, data: data, module: module} = state) do
+      event = module.handle(data, message)   # process message for an aggregate or process manager
+      wrapped_event = List.wrap(event)
+
+      new_data = Persistence.apply_events(module, data, wrapped_event)
+      Persistence.persist_events(wrapped_event, uuid, expected_version)
+      state = %Container{ state |
+        data: new_data,
+        version: expected_version + length(wrapped_event)
+      }
+      {:ok, state}
+  end
+
+  # update the process instance's state by applying the event
+  def mutate_state(module, data, event), do:
+    module.apply(data, event)
+
+
+end

lib/dispatcher.ex

@@ -0,0 +1,17 @@
+defmodule Workflow.Dispatcher do
+  @moduledoc"""
+  Dispatch commands and events (messages) to aggregates and process managers
+  """
+  require Logger
+
+  alias Workflow.Repository
+  alias Workflow.Container
+
+  def dispatch(message, module, uuid, timeout) do
+    Logger.debug(fn -> "attempting to dispatch message: #{inspect message}, to: module: #{inspect module}" end)
+    {:ok, container} = Repository.open(module, uuid)
+    Container.execute(container, message, timeout)
+  end
+
+
+end