Conduit

v0.12.10

Conduit

Conduit v0.12.10 Conduit.Message View Source

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

Link to this section Summary

Types

assigns()
body()
content_encoding()
content_type()
correlation_id()
created_at()
created_by()
destination()
headers()
message_id()
private()
source()
status()
t()
user_id()

Functions

ack(message)

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

assign(message, key, value)

Assigns a named value to the message

assigns(message, key)

Retrieves a named value from the message

delete_header(message, key)

Deletes a header from the message specified by key

get_fields(message)

Returns all non-nil fields from the message as a map

get_header(message, key)

Returns a header from the message specified by key

get_private(message, key)

Retrieves a named value from the message. This is intended for libraries and framework use

merge_fields(to, from, fields \\ [:source, :destination, :user_id, :correlation_id, :message_id, :content_type, :content_encoding, :created_by, :created_at, :status])

Merges fields to one message from another

merge_headers(to, from, headers)

Merges headers to one message from another

nack(message)

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

put_body(message, body)

Assigns the content of the message

put_content_encoding(message, content_encoding)

Assigns a content_encoding to the message

put_content_type(message, content_type)

Assigns a content_type to the message

put_correlation_id(message, correlation_id)

Assigns a correlation_id to the message

put_created_at(message, created_at)

Assigns a created_at to the message

put_created_by(message, created_by)

Assigns a created_by to the message

put_destination(message, destination)

Assigns the destination of the message

put_header(message, key, value)

Assigns a header for the message specified by key

put_headers(message, headers)

Assigns a header for the message specified by key

put_message_id(message, message_id)

Assigns a message_id to the message

put_new_correlation_id(message, correlation_id)

Assigns a correlation_id to the message when one isn’t set already

put_new_destination(message, destination)

Assigns a destination to the message when one isn’t set already

put_new_message_id(message, message_id)

Assigns a message_id to the message when one isn’t set already

put_new_source(message, source)

Assigns a source to the message when one isn’t set already

put_private(message, key, value)

Assigns a named value to the message. This is intended for libraries and framework use

put_source(message, source)

Assigns the source of the message

put_user_id(message, user_id)

Assigns a user_id to the message

take(from, opts)

Creates a new message with the fields and headers specified

Link to this section Types

Link to this type assigns() View Source 
assigns() :: %{optional(atom()) => any()}
Link to this type content_encoding() View Source 
content_encoding() :: String.t() | (... -> any()) | nil
Link to this type content_type() View Source 
content_type() :: String.t() | (... -> any()) | nil
Link to this type correlation_id() View Source 
correlation_id() :: binary() | integer() | (... -> any()) | nil
Link to this type created_at() View Source 
created_at() :: String.t() | integer() | (... -> any()) | nil
Link to this type created_by() View Source 
created_by() :: binary() | (... -> any()) | nil
Link to this type destination() View Source 
destination() :: binary() | (... -> any()) | nil
Link to this type headers() View Source 
headers() :: %{optional(String.t()) => any()}
Link to this type message_id() View Source 
message_id() :: binary() | integer() | (... -> any()) | nil
Link to this type private() View Source 
private() :: %{optional(atom()) => any()}
Link to this type source() View Source 
source() :: binary() | (... -> any()) | nil
Link to this type status() View Source 
status() :: :ack | :nack
Link to this type t() View Source 
t() :: %Conduit.Message{
  assigns: assigns(),
  body: body(),
  content_encoding: content_encoding(),
  content_type: content_type(),
  correlation_id: correlation_id(),
  created_at: created_at(),
  created_by: created_by(),
  destination: destination(),
  headers: headers(),
  message_id: message_id(),
  private: private(),
  source: source(),
  status: status(),
  user_id: user_id()
}
Link to this type user_id() View Source 
user_id() :: binary() | integer() | (... -> any()) | nil

Link to this section Functions

Link to this function ack(message) View Source 
ack(Conduit.Message.t()) :: Conduit.Message.t()

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
Link to this function assign(message, key, value) View Source 
assign(Conduit.Message.t(), atom(), any()) :: Conduit.Message.t()

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
Link to this function assigns(message, key) View Source 
assigns(Conduit.Message.t(), term()) :: Conduit.Message.t()

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
Link to this function delete_header(message, key) View Source 
delete_header(Conduit.Message.t(), String.t()) :: Conduit.Message.t()

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
Link to this function get_fields(message) View Source 
get_fields(Conduit.Message.t()) :: %{optional(atom()) => term()}

Returns all non-nil fields from the message as a map.

The following fields will be returned:

  • :source
  • :destination
  • :user_id
  • :correlation_id
  • :message_id
  • :content_type
  • :content_encoding
  • :created_by
  • :created_at

Examples 

iex> import Conduit.Message
iex> message =
iex>   %Conduit.Message{}
iex>   |> put_message_id("1")
iex>   |> put_correlation_id("2")
iex> get_fields(message)
%{
  message_id: "1",
  correlation_id: "2"
}
Link to this function get_header(message, key) View Source 
get_header(Conduit.Message.t(), String.t()) :: any()

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
Link to this function get_private(message, key) View Source 
get_private(Conduit.Message.t(), atom()) :: term()

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
Link to this function merge_fields(to, from, fields \\ [:source, :destination, :user_id, :correlation_id, :message_id, :content_type, :content_encoding, :created_by, :created_at, :status]) View Source 
merge_fields(
  to :: Conduit.Message.t(),
  from :: Conduit.Message.t(),
  fields :: [atom()]
) :: Conduit.Message.t()

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)
iex> new_message.correlation_id
"123"
iex> new_message = Conduit.Message.merge_fields(%Conduit.Message{}, old_message, [:correlation_id])
iex> new_message.correlation_id
"123"
Link to this function merge_headers(to, from, headers) View Source 
merge_headers(
  to :: Conduit.Message.t(),
  from :: Conduit.Message.t(),
  headers :: [String.t()]
) :: Conduit.Message.t()

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
Link to this function nack(message) View Source 
nack(Conduit.Message.t()) :: Conduit.Message.t()

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
Link to this function put_body(message, body) View Source 
put_body(Conduit.Message.t(), body()) :: Conduit.Message.t()

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"
Link to this function put_content_encoding(message, content_encoding) View Source 
put_content_encoding(Conduit.Message.t(), content_encoding()) ::
  Conduit.Message.t()

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"
Link to this function put_content_type(message, content_type) View Source 
put_content_type(Conduit.Message.t(), content_type()) :: Conduit.Message.t()

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"
Link to this function put_correlation_id(message, correlation_id) View Source 
put_correlation_id(Conduit.Message.t(), correlation_id()) :: Conduit.Message.t()

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
Link to this function put_created_at(message, created_at) View Source 
put_created_at(Conduit.Message.t(), created_at()) :: Conduit.Message.t()

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
Link to this function put_created_by(message, created_by) View Source 
put_created_by(Conduit.Message.t(), created_by()) :: Conduit.Message.t()

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"
Link to this function put_destination(message, destination) View Source 
put_destination(Conduit.Message.t(), destination()) :: Conduit.Message.t()

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"
Link to this function put_header(message, key, value) View Source 
put_header(Conduit.Message.t(), String.t(), any()) :: Conduit.Message.t()

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
Link to this function put_headers(message, headers) View Source 
put_headers(Conduit.Message.t(), %{optional(String.t()) => any()}) ::
  Conduit.Message.t()

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
Link to this function put_message_id(message, message_id) View Source 
put_message_id(Conduit.Message.t(), message_id()) :: Conduit.Message.t()

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
Link to this function put_new_correlation_id(message, correlation_id) View Source 
put_new_correlation_id(Conduit.Message.t(), correlation_id()) ::
  Conduit.Message.t()

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
Link to this function put_new_destination(message, destination) View Source 
put_new_destination(Conduit.Message.t(), destination()) :: Conduit.Message.t()

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"
Link to this function put_new_message_id(message, message_id) View Source 
put_new_message_id(Conduit.Message.t(), message_id()) :: Conduit.Message.t()

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
Link to this function put_new_source(message, source) View Source 
put_new_source(Conduit.Message.t(), source()) :: Conduit.Message.t()

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"
Link to this function put_private(message, key, value) View Source 
put_private(Conduit.Message.t(), atom(), any()) :: Conduit.Message.t()

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
Link to this function put_source(message, source) View Source 
put_source(Conduit.Message.t(), source()) :: Conduit.Message.t()

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"
Link to this function put_user_id(message, user_id) View Source 
put_user_id(Conduit.Message.t(), user_id()) :: Conduit.Message.t()

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
Link to this function take(from, opts) View Source 
take(
  from :: Conduit.Message.t(),
  opts :: [fields: [atom()], headers: [String.t()]]
) :: Conduit.Message.t()

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