Current section
Files
Jump to
Current section
Files
lib/conduit/message.ex
defmodule Conduit.Message do
@moduledoc """
The Conduit message.
This module defines a `Conduit.Message` struct and the main functions
for working with Conduit messages.
Note this struct is used for sending and receiving messages from a
message queue.
## Public fields
These fields are for you to use in your application. The values in
`user_id`, `correlation_id`, `message_id`, `content_type`,
`content_encoding`, `created_by`, `created_at`, `headers`, and
`status` may have special meaning based on the adapter you use.
See your adapters documention to understand how to use them correctly.
* `source` - For incoming messages, this will be set to the queue the message was
consumed from.
* `destination` - For outgoing messages, this will be set to the destination queue (or
routing key) it is published to.
* `user_id` - An ID representing which user the message pertains to.
* `correlation_id` - An ID for a chain of messages, where the current message is one in
that chain.
* `message_id` - A unique ID for this message.
* `content_type` - The media type of the message body.
* `content_encoding` - The encoding of the message body.
* `created_by` - The name of the app that created the message.
* `created_at` - A timestamp or epoch representing when the message was created.
* `headers` - Information applicable to a specific message stored as a keyword list.
* `body` - The contents of the message.
* `status` - The operation to perform on the message. This only applies to messages
that are being received.
## Private fields
These fields are reserved for library/framework usage.
* `private` - shared library data as a map
"""
@type source :: binary | fun | nil
@type destination :: binary | fun | nil
@type user_id :: binary | integer | fun | nil
@type correlation_id :: binary | integer | fun | nil
@type message_id :: binary | integer | fun | nil
@type content_type :: String.t() | fun | nil
@type content_encoding :: String.t() | fun | nil
@type created_by :: binary | fun | nil
@type created_at :: String.t() | integer | fun | nil
@type headers :: %{String.t() => any}
@type body :: any
@type status :: :ack | :nack
@type assigns :: %{atom => any}
@type private :: %{atom => any}
@type t :: %__MODULE__{
source: source,
destination: destination,
user_id: user_id,
correlation_id: correlation_id,
message_id: message_id,
content_type: content_type,
content_encoding: content_encoding,
created_by: created_by,
created_at: created_at,
headers: headers,
body: body,
status: status,
assigns: assigns,
private: private
}
defstruct source: nil,
destination: nil,
user_id: nil,
correlation_id: nil,
message_id: nil,
content_type: nil,
content_encoding: nil,
created_by: nil,
created_at: nil,
headers: %{},
body: nil,
status: :ack,
assigns: %{},
private: %{}
@doc """
Creates a new message with the fields and headers specified.
## Examples
iex> import Conduit.Message
iex> old_message =
iex> %Conduit.Message{}
iex> |> put_correlation_id("123")
iex> |> put_header("retries", 1)
iex> new_message = Conduit.Message.take(old_message,
iex> headers: ["retries"], fields: [:correlation_id])
iex> new_message.correlation_id
"123"
iex> get_header(new_message, "retries")
1
"""
@spec take(from :: __MODULE__.t(), opts :: [fields: [atom], headers: [String.t()]]) :: __MODULE__.t()
def take(from, opts) do
%__MODULE__{}
|> merge_fields(from, Keyword.get(opts, :fields, []))
|> merge_headers(from, Keyword.get(opts, :headers, []))
end
@doc """
Merges fields to one message from another.
## Examples
iex> import Conduit.Message
iex> old_message = put_correlation_id(%Conduit.Message{}, "123")
iex> new_message = Conduit.Message.merge_fields(%Conduit.Message{}, old_message, [:correlation_id])
iex> new_message.correlation_id
"123"
"""
@allowed_fields [
:source,
:destination,
:user_id,
:correlation_id,
:message_id,
:content_type,
:content_encoding,
:created_by,
:created_at,
:status
]
@spec merge_fields(to :: __MODULE__.t(), from :: __MODULE__.t(), fields :: [atom]) :: __MODULE__.t()
def merge_fields(%__MODULE__{} = to, %__MODULE__{} = from, fields) do
fields =
MapSet.intersection(
MapSet.new(@allowed_fields),
MapSet.new(fields)
)
Map.merge(to, Map.take(from, fields))
end
@doc """
Merges headers to one message from another.
## Examples
iex> import Conduit.Message
iex> old_message = put_header(%Conduit.Message{}, "retries", 1)
iex> new_message = Conduit.Message.merge_headers(%Conduit.Message{}, old_message, ["retries"])
iex> get_header(new_message, "retries")
1
"""
@spec merge_headers(to :: __MODULE__.t(), from :: __MODULE__.t(), headers :: [String.t()]) :: __MODULE__.t()
def merge_headers(%__MODULE__{} = to, %__MODULE__{} = from, headers) do
headers = Map.take(from.headers, headers)
%{to | headers: Map.merge(to.headers, headers)}
end
@doc """
Assigns the source of the message.
## Examples
iex> import Conduit.Message
iex> message =
iex> %Conduit.Message{}
iex> |> put_source("my.queue")
iex> |> put_header("routing_key", "my.routing_key")
iex> message.source
"my.queue"
iex> message = put_source(message, fn mess ->
iex> get_header(mess, "routing_key")
iex> end)
iex> message.source
"my.routing_key"
"""
@spec put_source(__MODULE__.t(), source) :: __MODULE__.t()
def put_source(%__MODULE__{} = message, source) when is_function(source) do
put_source(message, source.(message))
end
def put_source(%__MODULE__{} = message, source) do
%{message | source: source}
end
@doc """
Assigns a source to the message when one isn't set already.
## Examples
iex> import Conduit.Message
iex> message = put_new_source(%Conduit.Message{}, "my.queue")
iex> message = put_new_source(message, "your.queue")
iex> message.source
"my.queue"
iex> message = put_new_source(%Conduit.Message{}, fn _mess -> "my.queue" end)
iex> message = put_new_source(message, fn _mess -> "your.queue" end)
iex> message.source
"my.queue"
"""
@spec put_new_source(__MODULE__.t(), source) :: __MODULE__.t()
def put_new_source(%__MODULE__{source: nil} = message, source) do
put_source(message, source)
end
def put_new_source(%__MODULE__{} = message, _) do
message
end
@doc """
Assigns the destination of the message.
## Examples
iex> import Conduit.Message
iex> message =
iex> %Conduit.Message{}
iex> |> put_source("over.there")
iex> |> put_destination("my.queue")
iex> message.destination
"my.queue"
iex> message = put_destination(message, fn mess -> mess.source <> ".error" end)
iex> message.destination
"over.there.error"
"""
@spec put_destination(__MODULE__.t(), destination) :: __MODULE__.t()
def put_destination(%__MODULE__{} = message, destination) when is_function(destination) do
put_destination(message, destination.(message))
end
def put_destination(%__MODULE__{} = message, destination) do
%{message | destination: destination}
end
@doc """
Assigns a destination to the message when one isn't set already.
## Examples
iex> import Conduit.Message
iex> message = put_new_destination(%Conduit.Message{}, "your.queue")
iex> message = put_new_destination(message, "my.queue")
iex> message.destination
"your.queue"
iex> message = put_new_destination(%Conduit.Message{}, fn _mess -> "your.queue" end)
iex> message = put_new_destination(message, fn _mess -> "my.queue" end)
iex> message.destination
"your.queue"
"""
@spec put_new_destination(__MODULE__.t(), destination) :: __MODULE__.t()
def put_new_destination(%__MODULE__{destination: nil} = message, destination) do
put_destination(message, destination)
end
def put_new_destination(%__MODULE__{} = message, _) do
message
end
@doc """
Assigns a user_id to the message.
## Examples
iex> import Conduit.Message
iex> message = put_user_id(%Conduit.Message{}, 1)
iex> message.user_id
1
iex> message = put_user_id(message, fn _mess -> 2 end)
iex> message.user_id
2
"""
@spec put_user_id(__MODULE__.t(), user_id) :: __MODULE__.t()
def put_user_id(%__MODULE__{} = message, user_id) when is_function(user_id) do
put_user_id(message, user_id.(message))
end
def put_user_id(%__MODULE__{} = message, user_id) do
%{message | user_id: user_id}
end
@doc """
Assigns a correlation_id to the message.
## Examples
iex> import Conduit.Message
iex> message = put_correlation_id(%Conduit.Message{}, 1)
iex> message.correlation_id
1
iex> message = put_correlation_id(message, fn _mess -> 2 end)
iex> message.correlation_id
2
"""
@spec put_correlation_id(__MODULE__.t(), correlation_id) :: __MODULE__.t()
def put_correlation_id(%__MODULE__{} = message, correlation_id)
when is_function(correlation_id) do
put_correlation_id(message, correlation_id.(message))
end
def put_correlation_id(%__MODULE__{} = message, correlation_id) do
%{message | correlation_id: correlation_id}
end
@doc """
Assigns a correlation_id to the message when one isn't set already.
## Examples
iex> import Conduit.Message
iex> message = put_new_correlation_id(%Conduit.Message{}, 1)
iex> message = put_new_correlation_id(message, 2)
iex> message.correlation_id
1
iex> message = put_new_correlation_id(%Conduit.Message{}, fn _mess -> 1 end)
iex> message = put_new_correlation_id(message, fn _mess -> 2 end)
iex> message.correlation_id
1
"""
@spec put_new_correlation_id(__MODULE__.t(), correlation_id) :: __MODULE__.t()
def put_new_correlation_id(%__MODULE__{correlation_id: nil} = message, correlation_id) do
put_correlation_id(message, correlation_id)
end
def put_new_correlation_id(%__MODULE__{} = message, _) do
message
end
@doc """
Assigns a message_id to the message.
## Examples
iex> import Conduit.Message
iex> message = put_message_id(%Conduit.Message{}, 1)
iex> message.message_id
1
iex> message = put_message_id(%Conduit.Message{}, fn _mess -> 1 end)
iex> message.message_id
1
"""
@spec put_message_id(__MODULE__.t(), message_id) :: __MODULE__.t()
def put_message_id(%__MODULE__{} = message, message_id) when is_function(message_id) do
put_message_id(message, message_id.(message))
end
def put_message_id(%__MODULE__{} = message, message_id) do
%{message | message_id: message_id}
end
@doc """
Assigns a message_id to the message when one isn't set already.
## Examples
iex> import Conduit.Message
iex> message = put_new_message_id(%Conduit.Message{}, 1)
iex> message = put_new_message_id(message, 2)
iex> message.message_id
1
iex> message = put_new_message_id(%Conduit.Message{}, fn _mess -> 1 end)
iex> message = put_new_message_id(message, fn _mess -> 2 end)
iex> message.message_id
1
"""
@spec put_new_message_id(__MODULE__.t(), message_id) :: __MODULE__.t()
def put_new_message_id(%__MODULE__{message_id: nil} = message, message_id) do
put_message_id(message, message_id)
end
def put_new_message_id(%__MODULE__{} = message, _) do
message
end
@doc """
Assigns a content_type to the message.
## Examples
iex> import Conduit.Message
iex> message = put_content_type(%Conduit.Message{}, "application/json")
iex> message.content_type
"application/json"
iex> message = put_content_type(%Conduit.Message{}, fn _mess -> "application/json" end)
iex> message.content_type
"application/json"
"""
@spec put_content_type(__MODULE__.t(), content_type) :: __MODULE__.t()
def put_content_type(%__MODULE__{} = message, content_type) when is_function(content_type) do
put_content_type(message, content_type.(message))
end
def put_content_type(%__MODULE__{} = message, content_type) do
%{message | content_type: content_type}
end
@doc """
Assigns a content_encoding to the message.
## Examples
iex> import Conduit.Message
iex> message = put_content_encoding(%Conduit.Message{}, "gzip")
iex> message.content_encoding
"gzip"
iex> message = put_content_encoding(%Conduit.Message{}, fn _mess -> "gzip" end)
iex> message.content_encoding
"gzip"
"""
@spec put_content_encoding(__MODULE__.t(), content_encoding) :: __MODULE__.t()
def put_content_encoding(%__MODULE__{} = message, content_encoding)
when is_function(content_encoding) do
put_content_encoding(message, content_encoding.(message))
end
def put_content_encoding(%__MODULE__{} = message, content_encoding) do
%{message | content_encoding: content_encoding}
end
@doc """
Assigns a created_by to the message.
## Examples
iex> import Conduit.Message
iex> message = put_created_by(%Conduit.Message{}, "my_app")
iex> message.created_by
"my_app"
iex> message = put_created_by(%Conduit.Message{}, fn _mess ->"my_app" end)
iex> message.created_by
"my_app"
"""
@spec put_created_by(__MODULE__.t(), created_by) :: __MODULE__.t()
def put_created_by(%__MODULE__{} = message, created_by) when is_function(created_by) do
put_created_by(message, created_by.(message))
end
def put_created_by(%__MODULE__{} = message, created_by) do
%{message | created_by: created_by}
end
@doc """
Assigns a created_at to the message.
## Examples
iex> import Conduit.Message
iex> message = put_created_at(%Conduit.Message{}, 1)
iex> message.created_at
1
iex> message = put_created_at(%Conduit.Message{}, fn _mess -> 1 end)
iex> message.created_at
1
"""
@spec put_created_at(__MODULE__.t(), created_at) :: __MODULE__.t()
def put_created_at(%__MODULE__{} = message, created_at) when is_function(created_at) do
put_created_at(message, created_at.(message))
end
def put_created_at(%__MODULE__{} = message, created_at) do
%{message | created_at: created_at}
end
@doc """
Returns a header from the message specified by `key`.
## Examples
iex> import Conduit.Message
iex> message = put_header(%Conduit.Message{}, "retries", 1)
iex> get_header(message, "retries")
1
"""
@spec get_header(__MODULE__.t(), String.t()) :: any
def get_header(%__MODULE__{headers: headers}, key) when is_binary(key) do
get_in(headers, [key])
end
@doc """
Assigns a header for the message specified by `key`.
## Examples
iex> import Conduit.Message
iex> message = put_header(%Conduit.Message{}, "retries", 1)
iex> get_header(message, "retries")
1
iex> message = put_header(message, "retries", fn mess -> get_header(mess, "retries") + 1 end)
iex> get_header(message, "retries")
2
"""
@spec put_header(__MODULE__.t(), String.t(), any) :: __MODULE__.t()
def put_header(%__MODULE__{} = message, key, value)
when is_function(value) and is_binary(key) do
put_header(message, key, value.(message))
end
def put_header(%__MODULE__{headers: headers} = message, key, value) when is_binary(key) do
%{message | headers: put_in(headers, [key], value)}
end
@doc """
Assigns a header for the message specified by `key`.
## Examples
iex> import Conduit.Message
iex> message = put_headers(%Conduit.Message{}, %{"retries" => 1})
iex> get_header(message, "retries")
1
iex> message = put_headers(message, %{"retries" => fn mess -> get_header(mess, "retries") + 1 end})
iex> get_header(message, "retries")
2
"""
@spec put_headers(__MODULE__.t(), %{String.t() => any}) :: __MODULE__.t()
def put_headers(%__MODULE__{} = message, headers) when is_map(headers) do
Enum.reduce(headers, message, fn {key, value}, mess ->
put_header(mess, key, value)
end)
end
@doc """
Deletes a header from the message specified by `key`.
## Examples
iex> import Conduit.Message
iex> message = put_header(%Conduit.Message{}, "retries", 1)
iex> message = delete_header(message, "retries")
iex> get_header(message, "retries")
nil
"""
@spec delete_header(__MODULE__.t(), String.t()) :: __MODULE__.t()
def delete_header(%__MODULE__{headers: headers} = message, key) do
%{message | headers: Map.delete(headers, key)}
end
@doc """
Assigns the content of the message.
## Examples
iex> import Conduit.Message
iex> message = put_body(%Conduit.Message{}, "hi")
iex> message.body
"hi"
iex> message = put_body(message, fn _mess -> "bye" end)
iex> message.body
"bye"
"""
@spec put_body(__MODULE__.t(), body) :: __MODULE__.t()
def put_body(%__MODULE__{} = message, body) when is_function(body) do
put_body(message, body.(message))
end
def put_body(%__MODULE__{} = message, body) do
%{message | body: body}
end
@doc """
Assigs the status of the message as acknowledged. This will be used
to signal to the message queue that processing the message was successful
and can be discarded.
## Examples
iex> import Conduit.Message
iex> message = ack(%Conduit.Message{})
iex> message.status
:ack
"""
@spec ack(__MODULE__.t()) :: __MODULE__.t()
def ack(message) do
%{message | status: :ack}
end
@doc """
Assigs the status of the message to a negative acknowledged. This will be used
to signal to the message queue that processing the message was not successful.
## Examples
iex> import Conduit.Message
iex> message = nack(%Conduit.Message{})
iex> message.status
:nack
"""
@spec nack(__MODULE__.t()) :: __MODULE__.t()
def nack(message) do
%{message | status: :nack}
end
@doc """
Retrieves a named value from the message.
## Examples
iex> import Conduit.Message
iex> message = assign(%Conduit.Message{}, :user_id, 1)
iex> assigns(message, :user_id)
1
"""
@spec assigns(__MODULE__.t(), term) :: __MODULE__.t()
def assigns(%__MODULE__{assigns: assigns}, key) do
get_in(assigns, [key])
end
@doc """
Assigns a named value to the message.
## Examples
iex> import Conduit.Message
iex> message = assign(%Conduit.Message{}, :user_id, 1)
iex> assigns(message, :user_id)
1
"""
@spec assign(__MODULE__.t(), atom, any) :: __MODULE__.t()
def assign(%__MODULE__{assigns: assigns} = message, key, value) when is_atom(key) do
%{message | assigns: Map.put(assigns, key, value)}
end
@doc """
Retrieves a named value from the message. This is intended for libraries and framework use.
## Examples
iex> import Conduit.Message
iex> message = put_private(%Conduit.Message{}, :message_id, 1)
iex> get_private(message, :message_id)
1
"""
@spec get_private(__MODULE__.t(), term) :: __MODULE__.t()
def get_private(%__MODULE__{private: private}, key) do
get_in(private, [key])
end
@doc """
Assigns a named value to the message. This is intended for libraries and framework use.
## Examples
iex> import Conduit.Message
iex> message = put_private(%Conduit.Message{}, :message_id, 1)
iex> get_private(message, :message_id)
1
"""
@spec put_private(__MODULE__.t(), atom, any) :: __MODULE__.t()
def put_private(%__MODULE__{private: private} = message, key, value) when is_atom(key) do
%{message | private: Map.put(private, key, value)}
end
end