Current section

Files

Jump to
rambla lib rambla.ex
Raw

lib/rambla.ex

defmodule Rambla do
@moduledoc """
Interface for the message publishing through `Rambla`.
`Rambla` maintains connection pools with `Finitomata.Pool` for each service.
The typical config for `Rambla` service follows the pattern introduced by
`AMQP` library:
```elixir
service›: [
connections: [
connection_name›: [key_1›: ‹value_1,]
],
channels: [
channel_name›: [
connection: ‹connection_name,
options: [key_1›: ‹value_1,]
]
]
]
```
To start pools, simply embed `Rambla` into the supervision tree, it’d
start a supervisor with children for all the configured services.
The configuration of the service implies all the `Rambla`’s code for it will
be compiled, but the dependency itself must be added to the `deps` section
of the `Mix.Project` file.
The excerpt from the `Rambla.MixProject` itself follows
```elixir
# optional backends
{:amqp, "~> 3.0", optional: true},
{:redix, "~> 1.0", optional: true},
{:gen_smtp, "~> 0.4 or ~> 1.0", optional: true},
{:telemetria, "~> 0.4 or ~> 1.0", optional: true},
# s3
{:ex_aws, "~> 2.1", optional: true},
{:ex_aws_s3, "~> 2.0", optional: true},
{:ex_aws_sts, "~> 2.0", optional: true},
{:hackney, "~> 1.9", optional: true},
{:sweet_xml, "~> 0.6", optional: true},
{:configparser_ex, "~> 4.0", optional: true},
```
---
Channel names are used across connections to publish messages.
`Rambla.publish(:channel_1, message)` would publish the message to all channels
named `channel_1`.
"""
@channels for {service, opts} when is_list(opts) <-
Application.get_all_env(:rambla) ++ [{:amqp, Application.get_all_env(:amqp)}],
{:channels, opts} <- opts,
{name, _} <- opts,
reduce: %{},
do: (acc -> Map.update(acc, name, [service], &[service | &1]))
@doc "Returns a map `%{‹service› => [‹channels›]}`"
def channels, do: @channels
@doc "Returns a list of all the configured connections"
def services,
do: @channels |> Map.values() |> Enum.reduce([], &Kernel.++/2) |> Enum.uniq()
@doc false
def handler_for_service(name) do
case Application.get_env(:rambla, name) do
[{_, _} | _] = cfg -> Keyword.get(cfg, :handler)
_ -> nil
end || Module.concat(Rambla.Handlers, name |> to_string() |> Macro.camelize())
end
use Supervisor
@doc "Starts the supervisor with all the configured services"
def start_link(opts \\ []) do
case Keyword.pop(opts, :name) do
{nil, opts} -> Supervisor.start_link(__MODULE__, opts)
{name, opts} -> Supervisor.start_link(__MODULE__, opts, name: name)
end
end
@impl true
def init(_opts) do
@channels
|> Map.values()
|> Enum.reduce(&Kernel.++/2)
|> Enum.uniq()
|> Enum.map(&handler_for_service/1)
|> case do
[] -> :ignore
children -> Supervisor.init(children, strategy: :one_for_one)
end
end
@enable_deprecated Application.compile_env(:rambla, :enable_deprecated, true)
if @enable_deprecated do
@doc """
Starts the pools configured in the `config.exs` / `releases.exs` file.
This call is equivalent to `start_pools(Application.get_env(:rambla, :pools))`.
"""
@doc deprecated: "Use configuration instead"
def start_pools do
IO.warn("This call is deprecated and will be removed")
Rambla.ConnectionPool.start_pools()
end
@doc "Starts the pools as specified by options (`map()` or `keyword()`)"
@doc deprecated: "Use configuration instead"
def start_pools(opts) do
IO.warn("This call is deprecated and will be removed")
Rambla.ConnectionPool.start_pools(opts)
end
@doc "Returns the currently active pools"
@doc deprecated: "Use configuration instead"
def pools do
IO.warn("This call is deprecated and will be removed")
Rambla.ConnectionPool.pools()
end
end
@doc """
Publishes the message to the target channels. The message structure depends on
the destination. For `RabbitMQ` is might be whatever, for `Smtp` it expects
to have `to:`, `subject:` and `body:` fields.
"""
def publish(target, message, pid \\ nil)
if @enable_deprecated do
def publish(target, message, opts) when is_tuple(target) or is_map(opts) do
IO.warn("This call is deprecated and will be removed")
Rambla.ConnectionPool.publish(target, message, opts || %{})
end
def publish(target, message, opts) when target in [:amqp, :redis, :http, :smtp, :process] do
IO.warn("This call is deprecated and will be removed")
Rambla.ConnectionPool.publish(target, message, opts || %{})
end
def publish(target, message, opts)
when target in [Rambla.Amqp, Rambla.Redis, Rambla.Http, Rambla.Smtp, Rambla.Process] do
IO.warn("This call is deprecated and will be removed")
Rambla.ConnectionPool.publish(target, message, opts || %{})
end
end
def publish(channels, message, pid) when not is_list(channels) do
publish([channels], message, pid)
end
def publish(channels, message, pid) do
for channel <- channels,
service <- Map.get(@channels, channel, []),
handler <- [handler_for_service(service)] do
handler.publish(channel, message, pid)
end
end
if @enable_deprecated do
@doc """
Publishes the message to the destination synchronously, avoiding the pool.
"""
@doc deprecated: "Use configuration instead"
defdelegate publish_synch(target, message), to: Rambla.ConnectionPool
@doc """
Publishes the message to the destination synchronously, avoiding the pool.
Unlike `publish_synch/2`, allows to specify additional options per request.
"""
@doc deprecated: "Use configuration instead"
defdelegate publish_synch(target, message, opts), to: Rambla.ConnectionPool
@doc """
Executes any arbitrary function in the context of one of workers in the
respective connection pool for the target.
The function would receive a pid of the connection process.
"""
defdelegate raw(target, f), to: Rambla.ConnectionPool
end
end