defmodule Rill.Messaging.Message do defmodule KeyError do defexception [:message] end alias Rill.Messaging.Message.Metadata alias Rill.MessageStore.MessageData.Read alias Rill.MapCopy alias Rill.MessageStore.StreamName @type message_or_type :: String.t() | atom() | struct() @doc """ Behaves like `defstruct/1`, but doesn't allow to use attributes named `:id` or `:metadata` (reserved). The attribute `:id` is automatically defined with default `nil`. The attribute `:metadata` is automatically defined with default `%Rill.Messaging.Message.Metadata{}` """ @spec defstruct(attrs :: [atom()] | keyword()) :: any() defmacro defstruct(attrs) when is_list(attrs) do attrs = Enum.map(attrs, fn attr -> if is_tuple(attr), do: to_atom(attr), else: {to_atom(attr), nil} end) if Keyword.has_key?(attrs, :id) or Keyword.has_key?(attrs, :metadata) do raise KeyError, message: "Fields :id and :metadata are reserved" end attrs = attrs |> Keyword.merge(id: nil, metadata: %Rill.Messaging.Message.Metadata{}) |> Macro.escape() quote location: :keep do @behaviour Rill.Schema defstruct(unquote(attrs)) defimpl Rill.Schema.DataStructure do def to_map(data), do: @for.to_map(data) end def to_map(%__MODULE__{} = data) do Rill.Messaging.Message.to_map(data) end def build(%Read{} = message_data) do unquote(__MODULE__).build(__MODULE__, message_data) end @spec follow(preceding_message :: struct()) :: struct() def follow(%{} = preceding_message) do Rill.Messaging.Message.follow(__MODULE__, preceding_message) end @spec follow( preceding_message :: struct(), subsequent_message :: module() | struct() ) :: struct() def follow(%{} = preceding_message, subsequent_message) do Rill.Messaging.Message.follow( __MODULE__, preceding_message, subsequent_message ) end defoverridable Rill.Schema end end @doc """ Provides `defstruct/1` macro which allows creation of struct with required message keys (:id, :metadata) """ defmacro __using__(_opts \\ []) do quote location: :keep do require unquote(__MODULE__) import Kernel, except: [defstruct: 1] import unquote(__MODULE__), only: [defstruct: 1] end end @spec transient_attributes() :: [atom()] def transient_attributes do [ :id, :metadata ] end @spec message_type(msg :: message_or_type()) :: [atom()] def message_type(msg) do msg |> struct_name() |> Module.split() |> List.last() end @spec message_type?(msg :: message_or_type(), type :: String.t()) :: boolean() def message_type?(msg, type) do message_type(msg) == type end @spec message_name(msg :: message_or_type()) :: String.t() def message_name(msg) do msg |> message_type() |> canonize_name() end @spec canonize_name(name :: String.t()) :: String.t() def canonize_name(name) do Rill.Casing.to_snake(name) end @spec struct_name(msg :: message_or_type()) :: String.t() def struct_name(msg) when is_binary(msg), do: msg def struct_name(msg) when is_atom(msg), do: to_string(msg) def struct_name(%{__struct__: module}), do: to_string(module) @spec follows?(%{metadata: %Metadata{}}, %{metadata: %Metadata{}}) :: boolean() def follows?( %{metadata: %Metadata{} = metadata}, %{metadata: %Metadata{} = other_metadata} ) do Metadata.follows?(metadata, other_metadata) end @doc "Builds struct for `struct_name`" @spec build(struct_name :: module(), message_data :: %Read{}) :: struct() def build(struct_name, %Read{} = message_data) do data = message_data.data metadata = message_data.metadata msg = build(struct_name, data, metadata) Map.put(msg, :id, message_data.id) end @doc "Builds struct for `struct_name`" @spec build(struct_name :: module(), data :: map(), metadata :: map()) :: struct() def build(struct_name, data \\ nil, metadata \\ nil) do data = data || %{} metadata = if is_nil(metadata), do: %Metadata{}, else: Metadata.build(metadata) instance = struct(struct_name) {new_instance, _} = MapCopy.copy_existing(instance, data) Map.put(new_instance, :metadata, metadata) end def build_metadata(nil), do: %Metadata{} @spec build_metadata(metadata :: %Metadata{} | nil) :: %Metadata{} def build_metadata(%Metadata{} = metadata) do metadata |> Metadata.to_map() |> Metadata.build() end @doc "Builds struct for `struct_name` with `correlation_stream_name` set" @spec correlate( struct_name :: module(), correlation_stream_name :: StreamName.t() ) :: struct() def correlate(struct_name, correlation_stream_name) when is_atom(struct_name) and is_binary(correlation_stream_name) do map = build(struct_name) metadata = Map.put(map.metadata, :correlation_stream_name, correlation_stream_name) Map.put(map, :metadata, metadata) end @spec correlate( message :: struct(), correlation_stream_name :: StreamName.t() ) :: struct() def correlate(%{} = message, correlation_stream_name) when is_binary(correlation_stream_name) do metadata = Metadata.correlate(message.metadata, correlation_stream_name) Map.put(message, :metadata, metadata) end @type copy_opts :: {:metadata, nil | %Metadata{}} @spec copy( source :: struct(), receiver :: struct(), opts :: [copy_opts()] ) :: struct() def copy(%{} = source, %{} = receiver, opts \\ []) do metadata = Keyword.get(opts, :metadata) {new_receiver, _} = MapCopy.copy_existing(receiver, source) if is_nil(metadata) do new_receiver else {new_metadata, _} = MapCopy.copy_existing(new_receiver.metadata, metadata) Map.put(new_receiver, :metadata, new_metadata) end end @spec follow( preceding_message :: struct(), subsequent_message :: struct() ) :: struct() def follow(%{} = preceding_message, %{} = subsequent_message) do struct_name = subsequent_message.__struct__ follow(struct_name, preceding_message, subsequent_message) end @spec follow( struct_name :: module(), preceding_message :: struct() ) :: struct() def follow(struct_name, %{} = preceding_message) do follow(struct_name, preceding_message, struct_name) end @spec follow( struct_name :: module(), preceding_message :: struct(), subsequent_message :: module() ) :: struct() def follow(struct_name, preceding_message, subsequent_message) when is_atom(subsequent_message) do subsequent_message = build(subsequent_message) follow(struct_name, preceding_message, subsequent_message) end @spec follow( struct_name :: module(), preceding_message :: struct(), subsequent_message :: struct() ) :: struct() def follow(struct_name, %{} = preceding_message, %{} = subsequent_message) when is_atom(struct_name) do {subsequent_message, _} = MapCopy.copy_existing(subsequent_message, preceding_message) metadata = Metadata.follow( subsequent_message.metadata, preceding_message.metadata ) Map.put(subsequent_message, :metadata, metadata) end @spec to_map(msg :: struct()) :: map() def to_map(%{id: _, metadata: _} = msg) do msg |> Map.from_struct() |> Map.drop([:id, :metadata]) end defp to_atom(value) when is_atom(value), do: value defp to_atom(value) when is_binary(value), do: String.to_atom(value) end