defmodule Eidetic.Projector do @moduledoc """ Eidetic Projector provides a middleware system for consuming events and building projections. To get started, you'll need to start the Projector within your supervisor: ```elixir [ # The second parameter is your options, documented below. {Eidetic.Projector, []} ] ``` You can pass in pre and post-processors, which implement the `Eidetic.Projector.Middleware` behaviour. ```elixir [ {Eidetic.Projector, [ deserializer: MyDeserialize, preprocess: [ Some.PreProcess.Middleware ], postprocess: [ Some.PostProcess.Middleware ] ]} ] ``` Other options accepted are: ```elixir {Eidetic.Projector, [ name: :some_other_name ]} ``` """ use GenServer alias Eidetic.Event alias Eidetic.Projector.Bundle alias Eidetic.Projector.Bundle.ConsumerMeta require Logger @doc """ Got to start somewhere """ def start_link(options \\ []) do name = Keyword.get(options, :name, __MODULE__) _ = Logger.info(fn -> "Starting Eidetic Projector with name #{name}" end) GenServer.start_link(__MODULE__, options, name: name) end @doc """ Convert the options passed through `start_link/1` into our config map """ @impl GenServer def init(options) do deserializer = Keyword.get(options, :deserializer, nil) preprocess = Keyword.get(options, :preprocess, []) postprocess = Keyword.get(options, :postprocess, []) config = %{ deserializer: deserializer, preprocess: preprocess, postprocess: postprocess } _ = Logger.info(fn -> "Eidetic Projector Configuration: #{inspect(config)}" end) {:ok, config} end @doc """ Consumer's pass through their `ConsumerMeta` information and the received payload. It gets deserialized and passed through the preprocess middleware, hoping that a handle is configured. If no handler, exit. If a handler is established, send the bundle to the handler and run the post-process middleware. """ def consume(consumer_meta = %ConsumerMeta{}, payload) do GenServer.call(__MODULE__, {:consume, consumer_meta, payload}) end @doc """ GenServer Endpoint for `consume/1` """ @impl GenServer def handle_call({:consume, consumer_meta = %ConsumerMeta{}, payload}, _from, config) do event = %Event{} = deserialize(config[:deserializer], payload) result = %Bundle{ consumer_meta: consumer_meta, event: event } |> preprocess(config[:preprocess]) |> handler() |> postprocess(config[:postprocess]) {:reply, result, config} end defp preprocess(bundle = %Bundle{}, middleware) do process(:pre, bundle, middleware) end defp postprocess({:ok, bundle = %Bundle{}}, middleware) do process(:post, bundle, middleware) end defp postprocess({:skip, bundle = %Bundle{}}, _middleware) do {:skip, bundle} end defp postprocess({:error, error}, _middleware) do {:error, error} end defp handler({:ok, bundle = %Bundle{handler: handler}}) when is_function(handler) do bundle.handler.(bundle) end defp handler({:ok, bundle = %Bundle{handler: nil}}) do {:ok, bundle} end defp handler({:ok, bundle = %Bundle{handler: handler}}) when is_atom(handler) do bundle.handler.handle(bundle) end defp handler({:skip, bundle = %Bundle{}}) do {:skip, bundle} end defp handler({:error, error}) do {:error, error} end defp process(stage, bundle = %Bundle{}, [middleware | tail]) do _ = Logger.debug(fn -> "Processing Middleware: #{inspect(middleware)}" end) case middleware.process(stage, bundle) do {:ok, updated_bundle} -> process(stage, updated_bundle, tail) {:skip, updated_bundle} -> {:skip, updated_bundle} {:error, updated_bundle} -> {:error, updated_bundle} end end defp process(_, %Bundle{} = bundle, []) do {:ok, bundle} end defp deserialize(deserializer, payload) when is_function(deserializer) do deserializer.(payload) end defp deserialize(deserializer, payload) when is_atom(deserializer) do deserializer.deserialize(payload) end end