Packages
step_flow
1.8.1
1.9.0-rc2
1.9.0-rc1
1.9.0-rc0
1.8.2
1.8.1
1.8.1-rc8
1.8.1-rc7
1.8.1-rc6
1.8.1-rc5
1.8.1-rc4
1.8.1-rc3
1.8.1-rc2
1.8.1-rc1
1.8.1-rc0
1.8.0
1.8.0-rc3
1.8.0-rc2
1.8.0-rc1
1.8.0-rc0
1.7.3
1.7.3-rc4
1.7.3-rc3
1.7.3-rc2
1.7.3-rc1
1.7.3-rc0
1.7.2
1.7.2-rc4
1.7.2-rc3
1.7.2-rc2
1.7.2-rc1
1.7.2-rc0
1.7.1
1.7.0
1.7.0-rc1
1.7.0-rc0
1.6.1
1.6.1-rc1
1.6.1-rc0
1.6.0
1.6.0-rc9
1.6.0-rc8
1.6.0-rc7
1.6.0-rc6
1.6.0-rc5
1.6.0-rc4
1.6.0-rc3
1.6.0-rc20
1.6.0-rc2
1.6.0-rc19
1.6.0-rc18
1.6.0-rc17
1.6.0-rc16
1.6.0-rc15
1.6.0-rc14
1.6.0-rc13
1.6.0-rc12
1.6.0-rc11
1.6.0-rc10
1.6.0-rc1
1.5.0
1.5.0-rc1
1.4.2-rc2
1.4.2-rc1
1.4.1
1.4.1-rc1
1.4.0
1.4.0-rc4
1.4.0-rc3
1.4.0-rc2
1.4.0-rc1
1.3.1
1.3.0
1.3.0-rc
1.2.0
1.1.0
1.0.0
1.0.0-rc9
1.0.0-rc8
1.0.0-rc7
1.0.0-rc6
1.0.0-rc5
1.0.0-rc1
0.2.13
0.2.12
0.2.11
0.2.10
0.2.9
0.2.8
0.2.7
0.2.6
0.2.5
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
0.0.24
0.0.23
0.0.22
0.0.21
0.0.20
0.0.19
0.0.18
0.0.17
0.0.16
0.0.15
0.0.14
0.0.13
0.0.12
0.0.11
0.0.10
0.0.9
0.0.8
0.0.7
0.0.6
0.0.4
0.0.3
0.0.2
0.0.1
Step flow manager for Elixir applications
Current section
Files
Jump to
Current section
Files
lib/step_flow/amqp/helpers.ex
defmodule StepFlow.Amqp.Helpers do
require Logger
@moduledoc """
Helpers for AMQP.
"""
@doc """
Get AMQP URL from the configuration or environment variables.
- `hostname` Setup the hostname of the RabbitMQ service
- `username` Setup the username of the RabbitMQ service
- `password` Setup the password of the RabbitMQ service
- `port` Setup the port of the RabbitMQ service
- `virtual_host` Setup the virtual host of the RabbitMQ service
Hardcoded example:
config :step_flow, StepFlow.Amqp,
hostname: "example.com",
port: "5678",
username: "mediacloudai",
password: "mediacloudai",
virtual_host: "media_cloud_ai_dev"
Environment getter example:
config :step_flow, StepFlow.Amqp,
hostname: {:system, "AMQP_HOSTNAME"},
port: {:system, "AMQP_PORT"},
username: {:system, "AMQP_USERNAME"},
password: {:system, "AMQP_PASSWORD"},
virtual_host: {:system, "AMQP_VIRTUAL_HOST"},
"""
def get_amqp_connection_url do
scheme = StepFlow.Configuration.get_var_value(StepFlow.Amqp, :scheme, "amqp")
hostname = StepFlow.Configuration.get_var_value(StepFlow.Amqp, :hostname)
username = StepFlow.Configuration.get_var_value(StepFlow.Amqp, :username)
password = StepFlow.Configuration.get_var_value(StepFlow.Amqp, :password)
virtual_host = get_amqp_virtual_host()
port = get_amqp_port()
Logger.info("#{__MODULE__}: Connecting with hostname: #{hostname}")
url =
scheme <>
"://" <> username <> ":" <> password <> "@" <> hostname <> ":" <> port <> virtual_host
Logger.info("#{__MODULE__}: Connecting with url: #{url}")
url
end
defp convert_to_integer(value) do
case value do
v when is_bitstring(v) -> String.to_integer(value)
v when is_integer(v) -> value
end
end
defp convert_to_atom(value) do
case value do
v when is_bitstring(v) -> String.to_existing_atom(value)
v when is_atom(v) -> value
end
end
defp check_file_exists(path) do
case File.exists?(path) do
true -> path
false -> raise "Such a file does not exist: #{path}"
end
end
defp add_to_options(options, key, variable, update_function) do
case StepFlow.Configuration.get_var_value(StepFlow.Amqp, variable) do
nil ->
options
"" ->
options
value ->
options
|> Keyword.put(key, value)
|> Keyword.update(key, nil, fn v -> update_function.(v) end)
end
end
defp get_amqp_connection_ssl_options do
[]
|> add_to_options(:cacertfile, :ssl_cacertfile, &check_file_exists/1)
|> add_to_options(:certfile, :ssl_certfile, &check_file_exists/1)
|> add_to_options(:keyfile, :ssl_keyfile, &check_file_exists/1)
|> add_to_options(:depth, :ssl_depth, &convert_to_integer/1)
|> add_to_options(:verify, :ssl_verify, &convert_to_atom/1)
|> add_to_options(:fail_if_no_peer_cert, :ssl_fail_if_no_peer_cert, &convert_to_atom/1)
|> Enum.filter(fn {_key, value} -> !is_nil(value) end)
end
def get_amqp_connection_options do
options =
case get_amqp_connection_ssl_options() do
[] ->
[]
nil ->
[]
ssl_options ->
[]
|> Keyword.put(:ssl_options, ssl_options)
end
Logger.info("#{__MODULE__}: Connecting with options: #{inspect(options)}")
options
end
defp get_amqp_port do
StepFlow.Configuration.get_var_value(StepFlow.Amqp, :port, 5672)
|> StepFlow.Configuration.to_string()
end
defp get_amqp_virtual_host do
StepFlow.Configuration.get_var_value(StepFlow.Amqp, :virtual_host, "")
|> case do
"" -> ""
virtual_host -> "/" <> virtual_host
end
end
def get_amqp_message_options(options \\ []) do
options
|> Keyword.put(:persistent, get_amqp_delivery_mode() == 2)
end
def get_amqp_delivery_mode do
StepFlow.Configuration.get_var_value(StepFlow.Amqp, :delivery_mode, "1")
|> convert_to_integer
end
def get_amqp_server_configuration do
StepFlow.Configuration.get_var_value(StepFlow.Amqp, :server_configuration, "standalone")
end
end