Current section
Files
Jump to
Current section
Files
src/carotte.erl
-module(carotte).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/carotte.gleam").
-export([default_client/0, start/1, close/1, is_connected/1, connection_state/1, reconnect/1, describe_connection_error/1, describe_channel_error/1, describe_exchange_error/1, describe_queue_error/1, describe_publish_error/1, describe_consume_error/1, open_channel/1, close_channel/1, set_qos/3, start_transaction/1, commit_transaction/1, rollback_transaction/1, exchange/1, declare_exchange/2, declare_exchange_async/2, delete_exchange/3, delete_exchange_async/3, bind_exchange/4, bind_exchange_async/4, unbind_exchange/4, unbind_exchange_async/4, default_queue/1, declare_queue/2, declare_queue_async/2, delete_queue/4, delete_queue_async/4, bind_queue/4, bind_queue_async/4, unbind_queue/4, purge_queue/2, purge_queue_async/2, queue_status/2, empty_headers/0, publish/5, named_consumer/1, subscribe/4, subscribe_with_options/5, unsubscribe/2, unsubscribe_async/2, ack/3, ack_single/2, nack/4, nack_single/3, reject/3, get_message/3, start_consumer/1, consumer_supervised/1, headers_from_list/1, headers_to_list/1]).
-export_type([connection_error/0, channel_error/0, exchange_error/0, queue_error/0, publish_error/0, consume_error/0, client/0, client_config/0, connection_state/0, disconnect_reason/0, connection_event/0, channel/0, exchange/0, exchange_type/0, queue_config/0, queue/0, deliver/0, payload/0, queue_option/0, header_list/0, header_value/0, publish_option/0, consumer_config/0, consumer/0, ffi_option/0, consumer_state/0, consumer_message/0]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
?MODULEDOC(
" # Carotte 🥕\n"
"\n"
" A type-safe RabbitMQ client for Gleam that provides a clean, idiomatic interface\n"
" for message queue operations on the Erlang VM.\n"
"\n"
" ## Quick Start\n"
"\n"
" ```gleam\n"
" import carotte\n"
" import gleam/bit_array\n"
" import gleam/erlang/process\n"
" import gleam/io\n"
"\n"
" pub fn main() {\n"
" // Connect to RabbitMQ\n"
" let assert Ok(client) = carotte.start(carotte.default_client())\n"
" let assert Ok(ch) = carotte.open_channel(client)\n"
"\n"
" // Declare exchange and queue\n"
" let assert Ok(_) = carotte.declare_exchange(carotte.exchange(\"my_exchange\"), ch)\n"
" let assert Ok(_) = carotte.declare_queue(carotte.default_queue(\"my_queue\"), ch)\n"
" let assert Ok(_) = carotte.bind_queue(channel: ch, queue: \"my_queue\", exchange: \"my_exchange\", routing_key: \"\")\n"
"\n"
" // Start consumer supervisor and subscribe\n"
" let consumers = process.new_name(\"consumers\")\n"
" let assert Ok(consumer) = carotte.start_consumer(consumers)\n"
" let assert Ok(_) = carotte.subscribe(consumer, channel: ch, queue: \"my_queue\", callback: fn(msg, _) {\n"
" let assert Ok(text) = bit_array.to_string(msg.payload)\n"
" io.println(\"Received: \" <> text)\n"
" })\n"
"\n"
" // Publish a message (payload is BitArray)\n"
" let assert Ok(_) = carotte.publish(channel: ch, exchange: \"my_exchange\", routing_key: \"\", payload: <<\"Hello!\">>, options: [])\n"
" }\n"
" ```\n"
"\n"
" ## Features\n"
"\n"
" - **Type-safe API**: Leverage Gleam's type system for safe message handling\n"
" - **OTP Supervision**: Integrate consumers into your application's supervision tree\n"
" via `consumer_supervised`, or use standalone mode with `start_consumer`\n"
" - **Operation-Specific Errors**: Granular error types (`ConnectionError`, `ChannelError`,\n"
" `ExchangeError`, `QueueError`, `PublishError`, `ConsumeError`) for precise error handling\n"
" - **Async Operations**: Non-blocking variants with `_async` suffix\n"
" - **Full Headers Support**: Type-safe message headers with `HeaderValue` types\n"
" - **Connection Helpers**: Built-in reconnection support and connection monitoring\n"
"\n"
" ## OTP Supervision\n"
"\n"
" For production use, integrate consumers into your supervision tree:\n"
"\n"
" ```gleam\n"
" import gleam/erlang/process\n"
" import gleam/otp/static_supervisor\n"
"\n"
" let consumers_name = process.new_name(\"consumers\")\n"
" let spec = carotte.consumer_supervised(consumers_name)\n"
"\n"
" static_supervisor.new(static_supervisor.OneForOne)\n"
" |> static_supervisor.add(spec)\n"
" |> static_supervisor.start()\n"
"\n"
" let consumer = carotte.named_consumer(consumers_name)\n"
" carotte.subscribe(consumer, channel: ch, queue: \"my_queue\", callback: handler)\n"
" ```\n"
"\n"
" ## Error Handling\n"
"\n"
" Each operation category has its own error type:\n"
"\n"
" | Error Type | Operations |\n"
" |------------|------------|\n"
" | `ConnectionError` | `start`, `close`, `reconnect` |\n"
" | `ChannelError` | `open_channel` |\n"
" | `ExchangeError` | `declare_exchange`, `delete_exchange`, `bind_exchange`, `unbind_exchange` |\n"
" | `QueueError` | `declare_queue`, `delete_queue`, `bind_queue`, `unbind_queue`, `purge_queue`, `queue_status` |\n"
" | `PublishError` | `publish` |\n"
" | `ConsumeError` | `subscribe`, `unsubscribe`, `ack` |\n"
"\n"
" Use `describe_*_error` functions to convert errors to human-readable strings.\n"
"\n"
).
-type connection_error() :: connection_blocked |
connection_closed |
{connection_auth_failure, binary()} |
{connection_refused, binary()} |
{connection_timeout, binary()} |
not_connected |
{reconnection_failed, connection_error()} |
already_connected |
{connection_unknown_error, binary()}.
-type channel_error() :: {channel_closed, binary()} |
channel_process_not_found |
channel_connection_closed |
{channel_unknown_error, binary()}.
-type exchange_error() :: {exchange_not_found, binary()} |
{exchange_access_refused, binary()} |
{exchange_precondition_failed, binary()} |
{exchange_channel_closed, binary()} |
{exchange_unknown_error, binary()}.
-type queue_error() :: {queue_not_found, binary()} |
{queue_access_refused, binary()} |
{queue_precondition_failed, binary()} |
{queue_resource_locked, binary()} |
{queue_channel_closed, binary()} |
{queue_unknown_error, binary()}.
-type publish_error() :: {publish_no_route, binary()} |
{publish_channel_closed, binary()} |
{publish_unknown_error, binary()}.
-type consume_error() :: consume_init_timeout |
{consume_init_failed, binary()} |
consume_process_not_found |
{consume_channel_closed, binary()} |
{consume_unknown_error, binary()}.
-opaque client() :: {client, gleam@erlang@process:pid_(), client_config()}.
-type client_config() :: {client_config,
binary(),
binary(),
binary(),
binary(),
integer(),
integer(),
integer(),
gleam@time@duration:duration(),
gleam@time@duration:duration()}.
-type connection_state() :: connected | {disconnected, disconnect_reason()}.
-type disconnect_reason() :: server_closed |
network_error |
user_closed |
{unknown, binary()} |
connection_process_not_alive.
-type connection_event() :: {connection_disconnected, disconnect_reason()} |
connection_reconnected.
-type channel() :: any().
-type exchange() :: {exchange,
binary(),
exchange_type(),
boolean(),
boolean(),
boolean(),
boolean()}.
-type exchange_type() :: fanout | direct | topic | headers.
-type queue_config() :: {queue_config,
binary(),
boolean(),
boolean(),
boolean(),
boolean(),
boolean()}.
-type queue() :: {queue, binary(), integer(), integer()}.
-type deliver() :: {deliver, binary(), integer(), boolean(), binary(), binary()}.
-type payload() :: {payload, bitstring(), list(publish_option()), header_list()}.
-type queue_option() :: {auto_ack, boolean()}.
-opaque header_list() :: {header_list,
list({binary(), gleam@erlang@atom:atom_(), gleam@dynamic:dynamic_()})}.
-type header_value() :: {bool_header, boolean()} |
{float_header, float()} |
{int_header, integer()} |
{string_header, binary()} |
{list_header, list(header_value())}.
-type publish_option() :: {mandatory, boolean()} |
{content_type, binary()} |
{content_encoding, binary()} |
{message_headers, header_list()} |
{persistent, boolean()} |
{correlation_id, binary()} |
{priority, integer()} |
{reply_to, binary()} |
{expiration, gleam@time@duration:duration()} |
{message_id, binary()} |
{timestamp, gleam@time@timestamp:timestamp()} |
{type, binary()} |
{user_id, binary()} |
{app_id, binary()}.
-opaque consumer_config() :: {consumer_config,
channel(),
binary(),
boolean(),
fun((payload(), deliver()) -> nil)}.
-opaque consumer() :: {consumer,
gleam@erlang@process:name(gleam@otp@factory_supervisor:message(consumer_config(), binary()))}.
-type ffi_option() :: any().
-type consumer_state() :: {consumer_state,
channel(),
binary(),
fun((payload(), deliver()) -> nil),
boolean()}.
-type consumer_message() :: {amqp_delivery, payload(), deliver()} |
amqp_cancelled |
shutdown.
-file("src/carotte.gleam", 459).
?DOC(
" Create a new client builder with default settings.\n"
" Uses guest/guest credentials on localhost:5672.\n"
"\n"
" ## Example\n"
" ```gleam\n"
" let client = carotte.default_client()\n"
" |> carotte.start()\n"
" ```\n"
).
-spec default_client() -> client_config().
default_client() ->
{client_config,
<<"guest"/utf8>>,
<<"guest"/utf8>>,
<<"/"/utf8>>,
<<"localhost"/utf8>>,
5672,
2074,
0,
gleam@time@duration:seconds(10),
gleam@time@duration:seconds(60)}.
-file("src/carotte.gleam", 483).
?DOC(
" Start a RabbitMQ client connection.\n"
" Returns an actor.StartResult which contains the client on success.\n"
"\n"
" ## Example\n"
" ```gleam\n"
" case carotte.start(builder) {\n"
" Ok(client) -> // Use the client\n"
" Error(connection_error) -> // Handle connection error\n"
" }\n"
" ```\n"
).
-spec start(client_config()) -> {ok, client()} | {error, connection_error()}.
start(Builder) ->
{Heartbeat_secs, Heartbeat_nanoseconds} = gleam@time@duration:to_seconds_and_nanoseconds(
erlang:element(9, Builder)
),
Heartbeat_secs@1 = case Heartbeat_nanoseconds > 0 of
true ->
Heartbeat_secs + 1;
false ->
Heartbeat_secs
end,
{Timeout_secs, Timeout_nanos} = gleam@time@duration:to_seconds_and_nanoseconds(
erlang:element(10, Builder)
),
Timeout_ms = (Timeout_secs * 1000) + (Timeout_nanos div 1000000),
gleam@result:map(
carotte_ffi:start(
erlang:element(2, Builder),
erlang:element(3, Builder),
erlang:element(4, Builder),
erlang:element(5, Builder),
erlang:element(6, Builder),
erlang:element(7, Builder),
erlang:element(8, Builder),
Heartbeat_secs@1,
Timeout_ms
),
fun(Pid) -> {client, Pid, Builder} end
).
-file("src/carotte.gleam", 525).
?DOC(
" Close the RabbitMQ client connection.\n"
" This will close all channels and the underlying AMQP connection.\n"
).
-spec close(client()) -> {ok, nil} | {error, connection_error()}.
close(Client) ->
carotte_ffi:close(Client).
-file("src/carotte.gleam", 533).
?DOC(" Check if the client connection is currently active.\n").
-spec is_connected(client()) -> boolean().
is_connected(Client) ->
carotte_ffi:is_process_alive(Client).
-file("src/carotte.gleam", 541).
?DOC(" Get the current connection state.\n").
-spec connection_state(client()) -> connection_state().
connection_state(Client) ->
case is_connected(Client) of
true ->
connected;
false ->
{disconnected, connection_process_not_alive}
end.
-file("src/carotte.gleam", 551).
?DOC(
" Attempt to reconnect a disconnected client.\n"
" Uses the original connection parameters.\n"
" Returns error if already connected or reconnection fails.\n"
).
-spec reconnect(client()) -> {ok, client()} | {error, connection_error()}.
reconnect(Client) ->
case is_connected(Client) of
true ->
{error, already_connected};
false ->
Builder = erlang:element(3, Client),
{Heartbeat_secs, Heartbeat_nanoseconds} = gleam@time@duration:to_seconds_and_nanoseconds(
erlang:element(9, Builder)
),
Heartbeat_secs@1 = case Heartbeat_nanoseconds > 0 of
true ->
Heartbeat_secs + 1;
false ->
Heartbeat_secs
end,
{Timeout_secs, Timeout_nanos} = gleam@time@duration:to_seconds_and_nanoseconds(
erlang:element(10, Builder)
),
Timeout_ms = (Timeout_secs * 1000) + (Timeout_nanos div 1000000),
case carotte_ffi:start(
erlang:element(2, Builder),
erlang:element(3, Builder),
erlang:element(4, Builder),
erlang:element(5, Builder),
erlang:element(6, Builder),
erlang:element(7, Builder),
erlang:element(8, Builder),
Heartbeat_secs@1,
Timeout_ms
) of
{ok, Pid} ->
{ok, {client, Pid, Builder}};
{error, E} ->
{error, {reconnection_failed, E}}
end
end.
-file("src/carotte.gleam", 590).
?DOC(
" Convert a ConnectionError to a human-readable string description.\n"
" Useful for logging or displaying error messages to users.\n"
).
-spec describe_connection_error(connection_error()) -> binary().
describe_connection_error(Err) ->
case Err of
connection_blocked ->
<<"Connection blocked"/utf8>>;
connection_closed ->
<<"Connection closed"/utf8>>;
{connection_auth_failure, Msg} ->
<<"Auth failure: "/utf8, Msg/binary>>;
{connection_refused, Msg@1} ->
<<"Connection refused: "/utf8, Msg@1/binary>>;
{connection_timeout, Msg@2} ->
<<"Connection timeout: "/utf8, Msg@2/binary>>;
not_connected ->
<<"Not connected"/utf8>>;
{reconnection_failed, Cause} ->
<<"Reconnection failed: "/utf8,
(describe_connection_error(Cause))/binary>>;
already_connected ->
<<"Already connected"/utf8>>;
{connection_unknown_error, Msg@3} ->
<<"Unknown error: "/utf8, Msg@3/binary>>
end.
-file("src/carotte.gleam", 606).
?DOC(" Convert a ChannelError to a human-readable string description.\n").
-spec describe_channel_error(channel_error()) -> binary().
describe_channel_error(Err) ->
case Err of
{channel_closed, Msg} ->
<<"Channel closed: "/utf8, Msg/binary>>;
channel_process_not_found ->
<<"Channel process not found"/utf8>>;
channel_connection_closed ->
<<"Connection closed"/utf8>>;
{channel_unknown_error, Msg@1} ->
<<"Unknown error: "/utf8, Msg@1/binary>>
end.
-file("src/carotte.gleam", 616).
?DOC(" Convert an ExchangeError to a human-readable string description.\n").
-spec describe_exchange_error(exchange_error()) -> binary().
describe_exchange_error(Err) ->
case Err of
{exchange_not_found, Msg} ->
<<"Exchange not found: "/utf8, Msg/binary>>;
{exchange_access_refused, Msg@1} ->
<<"Access refused: "/utf8, Msg@1/binary>>;
{exchange_precondition_failed, Msg@2} ->
<<"Precondition failed: "/utf8, Msg@2/binary>>;
{exchange_channel_closed, Msg@3} ->
<<"Channel closed: "/utf8, Msg@3/binary>>;
{exchange_unknown_error, Msg@4} ->
<<"Unknown error: "/utf8, Msg@4/binary>>
end.
-file("src/carotte.gleam", 627).
?DOC(" Convert a QueueError to a human-readable string description.\n").
-spec describe_queue_error(queue_error()) -> binary().
describe_queue_error(Err) ->
case Err of
{queue_not_found, Msg} ->
<<"Queue not found: "/utf8, Msg/binary>>;
{queue_access_refused, Msg@1} ->
<<"Access refused: "/utf8, Msg@1/binary>>;
{queue_precondition_failed, Msg@2} ->
<<"Precondition failed: "/utf8, Msg@2/binary>>;
{queue_resource_locked, Msg@3} ->
<<"Resource locked: "/utf8, Msg@3/binary>>;
{queue_channel_closed, Msg@4} ->
<<"Channel closed: "/utf8, Msg@4/binary>>;
{queue_unknown_error, Msg@5} ->
<<"Unknown error: "/utf8, Msg@5/binary>>
end.
-file("src/carotte.gleam", 639).
?DOC(" Convert a PublishError to a human-readable string description.\n").
-spec describe_publish_error(publish_error()) -> binary().
describe_publish_error(Err) ->
case Err of
{publish_no_route, Msg} ->
<<"No route: "/utf8, Msg/binary>>;
{publish_channel_closed, Msg@1} ->
<<"Channel closed: "/utf8, Msg@1/binary>>;
{publish_unknown_error, Msg@2} ->
<<"Unknown error: "/utf8, Msg@2/binary>>
end.
-file("src/carotte.gleam", 648).
?DOC(" Convert a ConsumeError to a human-readable string description.\n").
-spec describe_consume_error(consume_error()) -> binary().
describe_consume_error(Err) ->
case Err of
consume_init_timeout ->
<<"Consumer init timeout"/utf8>>;
{consume_init_failed, Msg} ->
<<"Consumer init failed: "/utf8, Msg/binary>>;
consume_process_not_found ->
<<"Consumer process not found"/utf8>>;
{consume_channel_closed, Msg@1} ->
<<"Channel closed: "/utf8, Msg@1/binary>>;
{consume_unknown_error, Msg@2} ->
<<"Unknown error: "/utf8, Msg@2/binary>>
end.
-file("src/carotte.gleam", 663).
?DOC(" Open a channel to a RabbitMQ server.\n").
-spec open_channel(client()) -> {ok, channel()} | {error, channel_error()}.
open_channel(Client) ->
carotte_ffi:open_channel(Client).
-file("src/carotte.gleam", 673).
?DOC(
" Close a channel.\n"
" This releases the channel resources on the server.\n"
" Once closed, the channel cannot be used for further operations.\n"
).
-spec close_channel(channel()) -> {ok, nil} | {error, channel_error()}.
close_channel(Channel) ->
carotte_ffi:close_channel(Channel).
-file("src/carotte.gleam", 695).
?DOC(
" Set Quality of Service (QoS) for a channel.\n"
" Controls the prefetch count for message delivery.\n"
"\n"
" ## Parameters\n"
" - `channel`: The channel to configure\n"
" - `prefetch_count`: Maximum number of unacknowledged messages. Set to 0 for unlimited.\n"
" - `global`: If True, applies to the entire connection. If False, applies only to this channel.\n"
"\n"
" ## Example\n"
" ```gleam\n"
" // Limit to 10 unacknowledged messages per consumer\n"
" let assert Ok(_) = carotte.set_qos(ch, prefetch_count: 10, global: False)\n"
" ```\n"
"\n"
" This is essential for load balancing across multiple consumers.\n"
).
-spec set_qos(channel(), integer(), boolean()) -> {ok, nil} |
{error, channel_error()}.
set_qos(Channel, Prefetch_count, Global) ->
carotte_ffi:set_qos(Channel, Prefetch_count, Global).
-file("src/carotte.gleam", 722).
?DOC(
" Enable transaction mode on a channel.\n"
" Once enabled, messages published on this channel will not be delivered\n"
" until `commit_transaction` is called, or discarded if `rollback_transaction` is called.\n"
"\n"
" ## Example\n"
" ```gleam\n"
" let assert Ok(_) = carotte.start_transaction(ch)\n"
" let assert Ok(_) = carotte.publish(channel: ch, exchange: \"ex\", routing_key: \"key\", payload: <<\"msg1\">>, options: [])\n"
" let assert Ok(_) = carotte.publish(channel: ch, exchange: \"ex\", routing_key: \"key\", payload: <<\"msg2\">>, options: [])\n"
" let assert Ok(_) = carotte.commit_transaction(ch) // Both messages delivered atomically\n"
" ```\n"
).
-spec start_transaction(channel()) -> {ok, nil} | {error, channel_error()}.
start_transaction(Channel) ->
carotte_ffi:tx_select(Channel).
-file("src/carotte.gleam", 727).
?DOC(
" Commit the current transaction on a channel.\n"
" All messages published since `start_transaction` (or the last commit) are delivered.\n"
).
-spec commit_transaction(channel()) -> {ok, nil} | {error, channel_error()}.
commit_transaction(Channel) ->
carotte_ffi:tx_commit(Channel).
-file("src/carotte.gleam", 732).
?DOC(
" Rollback the current transaction on a channel.\n"
" All messages published since `start_transaction` (or the last commit) are discarded.\n"
).
-spec rollback_transaction(channel()) -> {ok, nil} | {error, channel_error()}.
rollback_transaction(Channel) ->
carotte_ffi:tx_rollback(Channel).
-file("src/carotte.gleam", 745).
?DOC(
" Create an exchange with the given name and sensible defaults.\n"
" Returns a Direct exchange with all options set to False.\n"
"\n"
" To customize, use record update syntax:\n"
" ```gleam\n"
" Exchange(..exchange(\"events\"), exchange_type: Topic, durable: True)\n"
" ```\n"
).
-spec exchange(binary()) -> exchange().
exchange(Name) ->
{exchange, Name, direct, false, false, false, false}.
-file("src/carotte.gleam", 757).
?DOC(" Declare an exchange on the broker.\n").
-spec declare_exchange(exchange(), channel()) -> {ok, nil} |
{error, exchange_error()}.
declare_exchange(Exchange, Channel) ->
carotte_ffi:exchange_declare(Channel, Exchange).
-file("src/carotte.gleam", 765).
?DOC(" Declare an exchange on the broker without waiting for a response.\n").
-spec declare_exchange_async(exchange(), channel()) -> {ok, nil} |
{error, exchange_error()}.
declare_exchange_async(Exchange, Channel) ->
carotte_ffi:exchange_declare(
Channel,
{exchange,
erlang:element(2, Exchange),
erlang:element(3, Exchange),
erlang:element(4, Exchange),
erlang:element(5, Exchange),
erlang:element(6, Exchange),
true}
).
-file("src/carotte.gleam", 780).
?DOC(
" Delete an exchange from the broker.\n"
" If `unused` is set to true, the exchange will only be deleted if it has no queues bound to it.\n"
).
-spec delete_exchange(channel(), binary(), boolean()) -> {ok, nil} |
{error, exchange_error()}.
delete_exchange(Channel, Exchange, Unused) ->
carotte_ffi:exchange_delete(Channel, Exchange, Unused, false).
-file("src/carotte.gleam", 789).
?DOC(" Delete an exchange from the broker without waiting for a response.\n").
-spec delete_exchange_async(channel(), binary(), boolean()) -> {ok, nil} |
{error, exchange_error()}.
delete_exchange_async(Channel, Exchange, Unused) ->
carotte_ffi:exchange_delete(Channel, Exchange, Unused, true).
-file("src/carotte.gleam", 807).
?DOC(
" Bind an exchange to another exchange.\n"
" Routing keys are used to filter messages from the source exchange.\n"
).
-spec bind_exchange(channel(), binary(), binary(), binary()) -> {ok, nil} |
{error, exchange_error()}.
bind_exchange(Channel, Source, Destination, Routing_key) ->
carotte_ffi:exchange_bind(Channel, Source, Destination, Routing_key, false).
-file("src/carotte.gleam", 818).
?DOC(
" Bind an exchange to another exchange without waiting for a response.\n"
" Same semantics as `bind_exchange`.\n"
).
-spec bind_exchange_async(channel(), binary(), binary(), binary()) -> {ok, nil} |
{error, exchange_error()}.
bind_exchange_async(Channel, Source, Destination, Routing_key) ->
carotte_ffi:exchange_bind(Channel, Source, Destination, Routing_key, true).
-file("src/carotte.gleam", 837).
?DOC(" Unbind an exchange from another exchange.\n").
-spec unbind_exchange(channel(), binary(), binary(), binary()) -> {ok, nil} |
{error, exchange_error()}.
unbind_exchange(Channel, Source, Destination, Routing_key) ->
carotte_ffi:exchange_unbind(
Channel,
Source,
Destination,
Routing_key,
false
).
-file("src/carotte.gleam", 848).
?DOC(
" Unbind an exchange from another exchange asynchronously.\n"
" Same semantics as `unbind_exchange`.\n"
).
-spec unbind_exchange_async(channel(), binary(), binary(), binary()) -> {ok,
nil} |
{error, exchange_error()}.
unbind_exchange_async(Channel, Source, Destination, Routing_key) ->
carotte_ffi:exchange_unbind(Channel, Source, Destination, Routing_key, true).
-file("src/carotte.gleam", 884).
?DOC(
" Create a queue configuration with the given name and sensible defaults.\n"
" All boolean options default to False.\n"
"\n"
" To customize, use record update syntax:\n"
" ```gleam\n"
" QueueConfig(..default_queue(\"my_queue\"), durable: True, exclusive: True)\n"
" ```\n"
"\n"
" For an auto-generated queue name, pass an empty string:\n"
" ```gleam\n"
" default_queue(\"\")\n"
" |> declare_queue(channel)\n"
" // Returns Queue with broker-generated name like \"amq.gen-...\"\n"
" ```\n"
).
-spec default_queue(binary()) -> queue_config().
default_queue(Name) ->
{queue_config, Name, false, false, false, false, false}.
-file("src/carotte.gleam", 896).
?DOC(" Declare a queue on the broker.\n").
-spec declare_queue(queue_config(), channel()) -> {ok, queue()} |
{error, queue_error()}.
declare_queue(Queue, Channel) ->
carotte_ffi:queue_declare(
Channel,
erlang:element(2, Queue),
erlang:element(3, Queue),
erlang:element(4, Queue),
erlang:element(5, Queue),
erlang:element(6, Queue),
erlang:element(7, Queue)
).
-file("src/carotte.gleam", 923).
?DOC(" Declare a queue on the broker asynchronously.\n").
-spec declare_queue_async(queue_config(), channel()) -> {ok, nil} |
{error, queue_error()}.
declare_queue_async(Queue, Channel) ->
carotte_ffi:queue_declare(
Channel,
erlang:element(2, Queue),
erlang:element(3, Queue),
erlang:element(4, Queue),
erlang:element(5, Queue),
erlang:element(6, Queue),
true
).
-file("src/carotte.gleam", 953).
?DOC(
" Delete a queue from the broker.\n"
" If `if_unused` is set, the queue will only be deleted if it has no subscribers.\n"
" If `if_empty` is set, the queue will only be deleted if it has no messages.\n"
" Returns the number of messages that were in the queue when it was deleted.\n"
).
-spec delete_queue(channel(), binary(), boolean(), boolean()) -> {ok, integer()} |
{error, queue_error()}.
delete_queue(Channel, Queue, If_unused, If_empty) ->
carotte_ffi:queue_delete(Channel, Queue, If_unused, If_empty, false).
-file("src/carotte.gleam", 964).
?DOC(
" Delete a queue from the broker asynchronously.\n"
" Same semantics as `delete_queue`.\n"
).
-spec delete_queue_async(channel(), binary(), boolean(), boolean()) -> {ok, nil} |
{error, queue_error()}.
delete_queue_async(Channel, Queue, If_unused, If_empty) ->
gleam@result:map(
carotte_ffi:queue_delete(Channel, Queue, If_unused, If_empty, true),
fun(_) -> nil end
).
-file("src/carotte.gleam", 985).
?DOC(
" Bind a queue to an exchange.\n"
" The `routing_key` is used to filter messages from the exchange.\n"
).
-spec bind_queue(channel(), binary(), binary(), binary()) -> {ok, nil} |
{error, queue_error()}.
bind_queue(Channel, Queue, Exchange, Routing_key) ->
carotte_ffi:queue_bind(Channel, Queue, Exchange, Routing_key, false).
-file("src/carotte.gleam", 996).
?DOC(
" Bind a queue to an exchange asynchronously.\n"
" Same semantics as `bind_queue`.\n"
).
-spec bind_queue_async(channel(), binary(), binary(), binary()) -> {ok, nil} |
{error, queue_error()}.
bind_queue_async(Channel, Queue, Exchange, Routing_key) ->
carotte_ffi:queue_bind(Channel, Queue, Exchange, Routing_key, true).
-file("src/carotte.gleam", 1016).
?DOC(
" Unbind a queue from an exchange.\n"
" The `routing_key` is used to filter messages from the exchange.\n"
).
-spec unbind_queue(channel(), binary(), binary(), binary()) -> {ok, nil} |
{error, queue_error()}.
unbind_queue(Channel, Queue, Exchange, Routing_key) ->
carotte_ffi:queue_unbind(Channel, Queue, Exchange, Routing_key).
-file("src/carotte.gleam", 1034).
?DOC(" Purge a queue of all messages.\n").
-spec purge_queue(channel(), binary()) -> {ok, integer()} |
{error, queue_error()}.
purge_queue(Channel, Queue) ->
carotte_ffi:queue_purge(Channel, Queue, false).
-file("src/carotte.gleam", 1042).
?DOC(" Purge a queue of all messages asynchronously.\n").
-spec purge_queue_async(channel(), binary()) -> {ok, nil} |
{error, queue_error()}.
purge_queue_async(Channel, Queue) ->
gleam@result:map(
carotte_ffi:queue_purge(Channel, Queue, true),
fun(_) -> nil end
).
-file("src/carotte.gleam", 1058).
?DOC(" Get the status of a queue.\n").
-spec queue_status(channel(), binary()) -> {ok, queue()} |
{error, queue_error()}.
queue_status(Channel, Queue) ->
carotte_ffi:queue_declare(Channel, Queue, true, false, false, false, false).
-file("src/carotte.gleam", 1086).
?DOC(
" Create an empty HeaderList.\n"
" Useful for pattern matching or when no headers are needed.\n"
).
-spec empty_headers() -> header_list().
empty_headers() ->
{header_list, []}.
-file("src/carotte.gleam", 1199).
?DOC(
" Convert a PublishOption to a list of FFI tuples (atom, value)\n"
" Returns a list because some options may be skipped\n"
).
-spec publish_option_to_tuple(publish_option()) -> ffi_option().
publish_option_to_tuple(Option) ->
case Option of
{mandatory, V} ->
_pipe = {erlang:binary_to_atom(<<"mandatory_ffi"/utf8>>), V},
gleam@function:identity(_pipe);
{content_type, V@1} ->
_pipe@1 = {erlang:binary_to_atom(<<"content_type_ffi"/utf8>>), V@1},
gleam@function:identity(_pipe@1);
{content_encoding, V@2} ->
_pipe@2 = {erlang:binary_to_atom(<<"content_encoding_ffi"/utf8>>),
V@2},
gleam@function:identity(_pipe@2);
{message_headers, {header_list, V@3}} ->
_pipe@3 = {erlang:binary_to_atom(<<"message_headers_ffi"/utf8>>),
V@3},
gleam@function:identity(_pipe@3);
{persistent, V@4} ->
_pipe@4 = {erlang:binary_to_atom(<<"persistent_ffi"/utf8>>), V@4},
gleam@function:identity(_pipe@4);
{correlation_id, V@5} ->
_pipe@5 = {erlang:binary_to_atom(<<"correlation_id_ffi"/utf8>>),
V@5},
gleam@function:identity(_pipe@5);
{priority, V@6} ->
_pipe@6 = {erlang:binary_to_atom(<<"priority_ffi"/utf8>>), V@6},
gleam@function:identity(_pipe@6);
{reply_to, V@7} ->
_pipe@7 = {erlang:binary_to_atom(<<"reply_to_ffi"/utf8>>), V@7},
gleam@function:identity(_pipe@7);
{expiration, Dur} ->
{Seconds, Nanos} = gleam@time@duration:to_seconds_and_nanoseconds(
Dur
),
Millis = (Seconds * 1000) + (Nanos div 1000000),
_pipe@8 = {erlang:binary_to_atom(<<"expiration_ffi"/utf8>>),
erlang:integer_to_binary(Millis)},
gleam@function:identity(_pipe@8);
{message_id, V@8} ->
_pipe@9 = {erlang:binary_to_atom(<<"message_id_ffi"/utf8>>), V@8},
gleam@function:identity(_pipe@9);
{timestamp, Ts} ->
{Epoch_secs, _} = gleam@time@timestamp:to_unix_seconds_and_nanoseconds(
Ts
),
_pipe@10 = {erlang:binary_to_atom(<<"timestamp_ffi"/utf8>>),
Epoch_secs},
gleam@function:identity(_pipe@10);
{type, V@9} ->
_pipe@11 = {erlang:binary_to_atom(<<"type_ffi"/utf8>>), V@9},
gleam@function:identity(_pipe@11);
{user_id, V@10} ->
_pipe@12 = {erlang:binary_to_atom(<<"user_id_ffi"/utf8>>), V@10},
gleam@function:identity(_pipe@12);
{app_id, V@11} ->
_pipe@13 = {erlang:binary_to_atom(<<"app_id_ffi"/utf8>>), V@11},
gleam@function:identity(_pipe@13)
end.
-file("src/carotte.gleam", 1180).
?DOC(
" Publish a message to an exchange.\n"
" The `routing_key` is used to route messages to queues.\n"
" The `options` are used to set message properties.\n"
).
-spec publish(
channel(),
binary(),
binary(),
bitstring(),
list(publish_option())
) -> {ok, nil} | {error, publish_error()}.
publish(Channel, Exchange, Routing_key, Payload, Options) ->
Ffi_options = gleam@list:map(Options, fun publish_option_to_tuple/1),
carotte_ffi:publish(Channel, Exchange, Routing_key, Payload, Ffi_options).
-file("src/carotte.gleam", 1321).
?DOC(
" Get a reference to a running consumer supervisor by its registered name.\n"
"\n"
" Use this to get a supervisor reference after it has been started as part\n"
" of your supervision tree via `consumer_supervised`.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let consumer = carotte.named_consumer(consumers_name)\n"
" ```\n"
).
-spec named_consumer(
gleam@erlang@process:name(gleam@otp@factory_supervisor:message(consumer_config(), binary()))
) -> consumer().
named_consumer(Name) ->
{consumer, Name}.
-file("src/carotte.gleam", 1327).
?DOC(
" Start a consumer under supervision.\n"
" Returns the consumer_tag string which can be used to unsubscribe later.\n"
).
-spec subscribe(
consumer(),
channel(),
binary(),
fun((payload(), deliver()) -> nil)
) -> {ok, binary()} | {error, consume_error()}.
subscribe(Consumer, Channel, Queue, Callback) ->
{consumer, Name} = Consumer,
Config = {consumer_config, Channel, Queue, true, Callback},
Supervisor = gleam@otp@factory_supervisor:get_by_name(Name),
_pipe = gleam@otp@factory_supervisor:start_child(Supervisor, Config),
_pipe@1 = gleam@result:map(
_pipe,
fun(Started) -> erlang:element(3, Started) end
),
gleam@result:map_error(_pipe@1, fun(E) -> case E of
init_timeout ->
consume_init_timeout;
{init_failed, Msg} ->
{consume_init_failed, Msg};
{init_exited, _} ->
{consume_init_failed, <<"Consumer init exited"/utf8>>}
end end).
-file("src/carotte.gleam", 1350).
?DOC(
" Start a consumer with options under supervision.\n"
" Returns the consumer_tag string which can be used to unsubscribe later.\n"
).
-spec subscribe_with_options(
consumer(),
channel(),
binary(),
list(queue_option()),
fun((payload(), deliver()) -> nil)
) -> {ok, binary()} | {error, consume_error()}.
subscribe_with_options(Consumer, Channel, Queue, Options, Callback) ->
{consumer, Name} = Consumer,
Auto_ack = case Options of
[] ->
true;
[{auto_ack, Ack} | _] ->
Ack
end,
Config = {consumer_config, Channel, Queue, Auto_ack, Callback},
Supervisor = gleam@otp@factory_supervisor:get_by_name(Name),
_pipe = gleam@otp@factory_supervisor:start_child(Supervisor, Config),
_pipe@1 = gleam@result:map(
_pipe,
fun(Started) -> erlang:element(3, Started) end
),
gleam@result:map_error(_pipe@1, fun(E) -> case E of
init_timeout ->
consume_init_timeout;
{init_failed, Msg} ->
{consume_init_failed, Msg};
{init_exited, _} ->
{consume_init_failed, <<"Consumer init exited"/utf8>>}
end end).
-file("src/carotte.gleam", 1377).
?DOC(" Unsubscribe and stop a consumer gracefully.\n").
-spec unsubscribe(channel(), binary()) -> {ok, nil} | {error, consume_error()}.
unsubscribe(Channel, Consumer_tag) ->
carotte_ffi:unsubscribe(Channel, Consumer_tag, false).
-file("src/carotte.gleam", 1385).
?DOC(" Unsubscribe a consumer asynchronously.\n").
-spec unsubscribe_async(channel(), binary()) -> {ok, nil} |
{error, consume_error()}.
unsubscribe_async(Channel, Consumer_tag) ->
carotte_ffi:unsubscribe(Channel, Consumer_tag, true).
-file("src/carotte.gleam", 1420).
?DOC(
" Acknowledge a message delivery.\n"
" Used when manual acknowledgment is enabled (AutoAck(False)).\n"
"\n"
" ## Parameters\n"
" - `channel`: The channel to acknowledge on\n"
" - `delivery_tag`: The delivery tag from the message metadata\n"
" - `multiple`: If True, acknowledges all messages up to and including this delivery tag\n"
"\n"
" ## Example\n"
" ```gleam\n"
" carotte.subscribe_with_options(\n"
" supervisor,\n"
" channel: ch,\n"
" queue: \"my_queue\",\n"
" options: [carotte.AutoAck(False)],\n"
" callback: fn(msg, meta) {\n"
" // Process message\n"
" let _ = carotte.ack(ch, meta.delivery_tag, False)\n"
" },\n"
" )\n"
" ```\n"
).
-spec ack(channel(), integer(), boolean()) -> {ok, nil} |
{error, consume_error()}.
ack(Channel, Delivery_tag, Multiple) ->
carotte_ffi:ack(Channel, Delivery_tag, Multiple).
-file("src/carotte.gleam", 1430).
?DOC(
" Acknowledge a message delivery (acknowledges only this message).\n"
" Convenience function for ack with multiple=False.\n"
).
-spec ack_single(channel(), integer()) -> {ok, nil} | {error, consume_error()}.
ack_single(Channel, Delivery_tag) ->
carotte_ffi:ack(Channel, Delivery_tag, false).
-file("src/carotte.gleam", 1470).
?DOC(
" Negatively acknowledge a message delivery.\n"
" Used when manual acknowledgment is enabled (AutoAck(False)) and you want\n"
" to indicate that the message could not be processed.\n"
"\n"
" ## Parameters\n"
" - `channel`: The channel to nack on\n"
" - `delivery_tag`: The delivery tag from the message metadata\n"
" - `multiple`: If True, nacks all messages up to and including this delivery tag\n"
" - `requeue`: If True, the message(s) will be requeued; if False, they will be\n"
" discarded or dead-lettered (if a dead letter exchange is configured)\n"
"\n"
" ## Example\n"
" ```gleam\n"
" carotte.subscribe_with_options(\n"
" consumer,\n"
" channel: ch,\n"
" queue: \"my_queue\",\n"
" options: [carotte.AutoAck(False)],\n"
" callback: fn(msg, meta) {\n"
" case process_message(msg) {\n"
" Ok(_) -> carotte.ack_single(ch, meta.delivery_tag)\n"
" Error(_) -> carotte.nack(ch, meta.delivery_tag, False, True) // Requeue for retry\n"
" }\n"
" },\n"
" )\n"
" ```\n"
).
-spec nack(channel(), integer(), boolean(), boolean()) -> {ok, nil} |
{error, consume_error()}.
nack(Channel, Delivery_tag, Multiple, Requeue) ->
carotte_ffi:nack(Channel, Delivery_tag, Multiple, Requeue).
-file("src/carotte.gleam", 1487).
?DOC(
" Negatively acknowledge a single message.\n"
" Convenience function for nack with multiple=False.\n"
"\n"
" ## Parameters\n"
" - `channel`: The channel to nack on\n"
" - `delivery_tag`: The delivery tag from the message metadata\n"
" - `requeue`: If True, the message will be requeued; if False, it will be\n"
" discarded or dead-lettered\n"
).
-spec nack_single(channel(), integer(), boolean()) -> {ok, nil} |
{error, consume_error()}.
nack_single(Channel, Delivery_tag, Requeue) ->
carotte_ffi:nack(Channel, Delivery_tag, false, Requeue).
-file("src/carotte.gleam", 1528).
?DOC(
" Reject a message delivery.\n"
" Similar to nack but only works with a single message (no multiple option).\n"
" This is the original AMQP 0-9-1 method for rejecting messages.\n"
"\n"
" ## Parameters\n"
" - `channel`: The channel to reject on\n"
" - `delivery_tag`: The delivery tag from the message metadata\n"
" - `requeue`: If True, the message will be requeued; if False, it will be\n"
" discarded or dead-lettered (if a dead letter exchange is configured)\n"
"\n"
" ## Example\n"
" ```gleam\n"
" carotte.subscribe_with_options(\n"
" consumer,\n"
" channel: ch,\n"
" queue: \"my_queue\",\n"
" options: [carotte.AutoAck(False)],\n"
" callback: fn(msg, meta) {\n"
" case validate_message(msg) {\n"
" Ok(_) -> carotte.ack_single(ch, meta.delivery_tag)\n"
" Error(_) -> carotte.reject(ch, meta.delivery_tag, False) // Discard invalid message\n"
" }\n"
" },\n"
" )\n"
" ```\n"
).
-spec reject(channel(), integer(), boolean()) -> {ok, nil} |
{error, consume_error()}.
reject(Channel, Delivery_tag, Requeue) ->
carotte_ffi:reject(Channel, Delivery_tag, Requeue).
-file("src/carotte.gleam", 1665).
?DOC(" Decoder for basic.deliver AMQP message metadata\n").
-spec basic_deliver_decoder() -> gleam@dynamic@decode:decoder(deliver()).
basic_deliver_decoder() ->
gleam@dynamic@decode:subfield(
[0, 1],
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Consumer_tag) ->
gleam@dynamic@decode:subfield(
[0, 2],
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Delivery_tag) ->
gleam@dynamic@decode:subfield(
[0, 3],
{decoder, fun gleam@dynamic@decode:decode_bool/1},
fun(Redelivered) ->
gleam@dynamic@decode:subfield(
[0, 4],
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Exchange) ->
gleam@dynamic@decode:subfield(
[0, 5],
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Routing_key) ->
gleam@dynamic@decode:success(
{deliver,
Consumer_tag,
Delivery_tag,
Redelivered,
Exchange,
Routing_key}
)
end
)
end
)
end
)
end
)
end
).
-file("src/carotte.gleam", 1764).
-spec handle_consumer_message(consumer_state(), consumer_message()) -> gleam@otp@actor:next(consumer_state(), consumer_message()).
handle_consumer_message(State, Message) ->
case Message of
{amqp_delivery, Payload, Deliver} ->
(erlang:element(4, State))(Payload, Deliver),
gleam@otp@actor:continue(State);
amqp_cancelled ->
gleam@otp@actor:stop();
shutdown ->
gleam@otp@actor:stop()
end.
-file("src/carotte.gleam", 1789).
?DOC(
" Decoder for AMQP headers\n"
" Headers come as a list of {Name, Type, Value} tuples or undefined\n"
).
-spec amqp_headers_decoder() -> gleam@dynamic@decode:decoder(header_list()).
amqp_headers_decoder() ->
Header_tuple_decoder = begin
gleam@dynamic@decode:field(
0,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Name) ->
gleam@dynamic@decode:field(
1,
gleam@erlang@atom:decoder(),
fun(Type_atom) ->
gleam@dynamic@decode:field(
2,
{decoder, fun gleam@dynamic@decode:decode_dynamic/1},
fun(Value) ->
gleam@dynamic@decode:success(
{Name, Type_atom, Value}
)
end
)
end
)
end
)
end,
List_decoder = gleam@dynamic@decode:list(Header_tuple_decoder),
_pipe = gleam@dynamic@decode:one_of(
List_decoder,
[gleam@dynamic@decode:map(gleam@erlang@atom:decoder(), fun(_) -> [] end)]
),
gleam@dynamic@decode:map(_pipe, fun(Field@0) -> {header_list, Field@0} end).
-file("src/carotte.gleam", 1807).
-spec add_if_some(list(KGQ), fun((KGR) -> KGQ), gleam@option:option(KGR)) -> list(KGQ).
add_if_some(List, Constructor, Value) ->
case Value of
{some, V} ->
[Constructor(V) | List];
none ->
List
end.
-file("src/carotte.gleam", 1681).
?DOC(" Decoder for AMQP message properties (content type, encoding, etc.)\n").
-spec payload_properties_decoder() -> gleam@dynamic@decode:decoder(list(publish_option())).
payload_properties_decoder() ->
Properties = [],
gleam@dynamic@decode:subfield(
[1],
gleam@dynamic@decode:optional(
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Content_type) ->
Properties@1 = add_if_some(
Properties,
fun(Field@0) -> {content_type, Field@0} end,
Content_type
),
gleam@dynamic@decode:subfield(
[2],
gleam@dynamic@decode:optional(
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Content_encoding) ->
Properties@2 = add_if_some(
Properties@1,
fun(Field@0) -> {content_encoding, Field@0} end,
Content_encoding
),
gleam@dynamic@decode:subfield(
[4],
gleam@dynamic@decode:optional(
{decoder, fun gleam@dynamic@decode:decode_int/1}
),
fun(Delivery_mode) ->
Properties@3 = add_if_some(
Properties@2,
fun(Field@0) -> {persistent, Field@0} end,
case Delivery_mode of
{some, 2} ->
{some, true};
{some, 1} ->
{some, false};
_ ->
none
end
),
gleam@dynamic@decode:subfield(
[5],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_int/1}
),
fun(Priority) ->
Properties@4 = add_if_some(
Properties@3,
fun(Field@0) -> {priority, Field@0} end,
Priority
),
gleam@dynamic@decode:subfield(
[6],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(Correlation_id) ->
Properties@5 = add_if_some(
Properties@4,
fun(Field@0) -> {correlation_id, Field@0} end,
Correlation_id
),
gleam@dynamic@decode:subfield(
[7],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(Reply_to) ->
Properties@6 = add_if_some(
Properties@5,
fun(Field@0) -> {reply_to, Field@0} end,
Reply_to
),
gleam@dynamic@decode:subfield(
[8],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(Expiration_str) ->
Expiration_duration = case Expiration_str of
{some, S} ->
case gleam_stdlib:parse_int(
S
) of
{ok, Ms} ->
{some,
gleam@time@duration:milliseconds(
Ms
)};
{error,
_} ->
none
end;
none ->
none
end,
Properties@7 = add_if_some(
Properties@6,
fun(Field@0) -> {expiration, Field@0} end,
Expiration_duration
),
gleam@dynamic@decode:subfield(
[9],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(Message_id) ->
Properties@8 = add_if_some(
Properties@7,
fun(Field@0) -> {message_id, Field@0} end,
Message_id
),
gleam@dynamic@decode:subfield(
[10],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_int/1}
),
fun(
Timestamp_secs
) ->
Timestamp_value = case Timestamp_secs of
{some,
Secs} ->
{some,
gleam@time@timestamp:from_unix_seconds(
Secs
)};
none ->
none
end,
Properties@9 = add_if_some(
Properties@8,
fun(Field@0) -> {timestamp, Field@0} end,
Timestamp_value
),
gleam@dynamic@decode:subfield(
[11],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(
Message_type
) ->
Properties@10 = add_if_some(
Properties@9,
fun(Field@0) -> {type, Field@0} end,
Message_type
),
gleam@dynamic@decode:subfield(
[12],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(
User_id
) ->
Properties@11 = add_if_some(
Properties@10,
fun(Field@0) -> {user_id, Field@0} end,
User_id
),
gleam@dynamic@decode:subfield(
[13],
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(
App_id
) ->
Properties@12 = add_if_some(
Properties@11,
fun(Field@0) -> {app_id, Field@0} end,
App_id
),
gleam@dynamic@decode:success(
Properties@12
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
).
-file("src/carotte.gleam", 1740).
?DOC(" Decoder for complete AMQP message payload (properties + body + headers)\n").
-spec payload_decoder() -> gleam@dynamic@decode:decoder(payload()).
payload_decoder() ->
gleam@dynamic@decode:subfield(
[1, 1],
payload_properties_decoder(),
fun(Properties) ->
gleam@dynamic@decode:subfield(
[1, 2],
{decoder, fun gleam@dynamic@decode:decode_bit_array/1},
fun(Payload) ->
gleam@dynamic@decode:subfield(
[1, 1, 3],
amqp_headers_decoder(),
fun(Headers) ->
gleam@dynamic@decode:success(
{payload, Payload, Properties, Headers}
)
end
)
end
)
end
).
-file("src/carotte.gleam", 1570).
?DOC(
" Get a single message from a queue without subscribing.\n"
" This is a synchronous, polling-based approach to consuming messages.\n"
"\n"
" ## Parameters\n"
" - `channel`: The channel to use\n"
" - `queue`: The queue name to get a message from\n"
" - `auto_ack`: If True, the message is automatically acknowledged. If False, you must call `ack()`.\n"
"\n"
" ## Returns\n"
" - `Ok(Some(#(payload, deliver)))` if a message was available\n"
" - `Ok(None)` if the queue is empty\n"
" - `Error(consume_error)` if there was an error\n"
"\n"
" ## Example\n"
" ```gleam\n"
" case carotte.get_message(ch, queue: \"my_queue\", auto_ack: True) {\n"
" Ok(Some(#(payload, deliver))) -> {\n"
" let assert Ok(text) = bit_array.to_string(payload.payload)\n"
" io.println(\"Got message: \" <> text)\n"
" }\n"
" Ok(None) -> io.println(\"Queue is empty\")\n"
" Error(e) -> io.println(\"Error: \" <> carotte.describe_consume_error(e))\n"
" }\n"
" ```\n"
"\n"
" **Note:** For continuous message consumption, use `subscribe()` instead.\n"
" This function is best for one-off message retrieval or polling scenarios.\n"
).
-spec get_message(channel(), binary(), boolean()) -> {ok,
gleam@option:option({payload(), deliver()})} |
{error, consume_error()}.
get_message(Channel, Queue, Auto_ack) ->
gleam@result:'try'(
carotte_ffi:basic_get(Channel, Queue, Auto_ack),
fun(Opt_result) -> case Opt_result of
none ->
{ok, none};
{some, Delivery_dyn} ->
case gleam@dynamic@decode:run(
Delivery_dyn,
basic_deliver_decoder()
) of
{ok, Deliver} ->
case gleam@dynamic@decode:run(
Delivery_dyn,
payload_decoder()
) of
{ok, Payload} ->
{ok, {some, {Payload, Deliver}}};
{error, _} ->
{error,
{consume_unknown_error,
<<"Failed to decode payload"/utf8>>}}
end;
{error, _} ->
{error,
{consume_unknown_error,
<<"Failed to decode delivery"/utf8>>}}
end
end end
).
-file("src/carotte.gleam", 1747).
-spec build_consumer_selector() -> gleam@erlang@process:selector(consumer_message()).
build_consumer_selector() ->
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select_record(
_pipe,
erlang:binary_to_atom(<<"basic.cancel"/utf8>>),
2,
fun(_) -> amqp_cancelled end
),
_pipe@2 = gleam@erlang@process:select_record(
_pipe@1,
erlang:binary_to_atom(<<"basic.cancel_ok"/utf8>>),
1,
fun(_) -> amqp_cancelled end
),
gleam@erlang@process:select_other(
_pipe@2,
fun(Delivery_dyn) ->
Basic_deliver@1 = case gleam@dynamic@decode:run(
Delivery_dyn,
basic_deliver_decoder()
) of
{ok, Basic_deliver} -> Basic_deliver;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"carotte"/utf8>>,
function => <<"build_consumer_selector"/utf8>>,
line => 1756,
value => _assert_fail,
start => 57158,
'end' => 57244,
pattern_start => 57169,
pattern_end => 57186})
end,
Payload@1 = case gleam@dynamic@decode:run(
Delivery_dyn,
payload_decoder()
) of
{ok, Payload} -> Payload;
_assert_fail@1 ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"carotte"/utf8>>,
function => <<"build_consumer_selector"/utf8>>,
line => 1758,
value => _assert_fail@1,
start => 57249,
'end' => 57317,
pattern_start => 57260,
pattern_end => 57271})
end,
{amqp_delivery, Payload@1, Basic_deliver@1}
end
).
-file("src/carotte.gleam", 1618).
-spec start_consumer_actor(consumer_config()) -> {ok,
gleam@otp@actor:started(binary())} |
{error, gleam@otp@actor:start_error()}.
start_consumer_actor(Config) ->
_pipe@4 = gleam@otp@actor:new_with_initialiser(
5000,
fun(_) ->
Consumer_pid = erlang:self(),
case carotte_ffi:consume(
erlang:element(2, Config),
erlang:element(3, Config),
Consumer_pid,
erlang:element(4, Config)
) of
{ok, Consumer_tag} ->
_ = begin
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select_record(
_pipe,
erlang:binary_to_atom(<<"basic.consume_ok"/utf8>>),
1,
fun(_) -> nil end
),
gleam_erlang_ffi:select(_pipe@1, 1000)
end,
State = {consumer_state,
erlang:element(2, Config),
Consumer_tag,
erlang:element(5, Config),
erlang:element(4, Config)},
Selector = build_consumer_selector(),
{ok,
begin
_pipe@2 = gleam@otp@actor:initialised(State),
_pipe@3 = gleam@otp@actor:selecting(
_pipe@2,
Selector
),
gleam@otp@actor:returning(_pipe@3, Consumer_tag)
end};
{error, _} ->
{error, <<"Failed to subscribe to queue"/utf8>>}
end
end
),
_pipe@5 = gleam@otp@actor:on_message(_pipe@4, fun handle_consumer_message/2),
gleam@otp@actor:start(_pipe@5).
-file("src/carotte.gleam", 1260).
?DOC(
" Start the consumer supervisor directly without adding it to a supervision tree.\n"
"\n"
" Most of the time you want to use `consumer_supervised` and add the\n"
" supervisor to your application's supervision tree instead of using this\n"
" function directly.\n"
"\n"
" The supervisor will be linked to the calling process and registered with\n"
" the given name.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" let name = process.new_name(\"my_consumers\")\n"
" let assert Ok(consumer) = carotte.start_consumer(name)\n"
" ```\n"
).
-spec start_consumer(
gleam@erlang@process:name(gleam@otp@factory_supervisor:message(consumer_config(), binary()))
) -> {ok, consumer()} | {error, gleam@otp@actor:start_error()}.
start_consumer(Name) ->
_pipe = gleam@otp@factory_supervisor:worker_child(
fun start_consumer_actor/1
),
_pipe@1 = gleam@otp@factory_supervisor:named(_pipe, Name),
_pipe@2 = gleam@otp@factory_supervisor:start(_pipe@1),
gleam@result:map(_pipe@2, fun(_) -> {consumer, Name} end).
-file("src/carotte.gleam", 1301).
?DOC(
" Create a child specification for adding the consumer supervisor to your\n"
" application's supervision tree.\n"
"\n"
" This is the recommended way to start the consumer supervisor, as it ensures\n"
" proper lifecycle management within your OTP application.\n"
"\n"
" You must provide a name so that other parts of your application can\n"
" find the supervisor to subscribe consumers.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" import gleam/erlang/process\n"
" import gleam/otp/static_supervisor\n"
"\n"
" pub fn start_app() {\n"
" // Create a name at program startup\n"
" let consumers_name = process.new_name(\"consumers\")\n"
"\n"
" // Create the child specification (max 5 restarts in 10 seconds)\n"
" let consumer_spec = carotte.consumer_supervised(consumers_name)\n"
"\n"
" // Add to your supervision tree\n"
" static_supervisor.new(static_supervisor.OneForOne)\n"
" |> static_supervisor.add(consumer_spec)\n"
" |> static_supervisor.start()\n"
"\n"
" // Later, get the supervisor to subscribe\n"
" let sup = carotte.named_consumer(consumers_name)\n"
" carotte.subscribe(sup, channel: ch, queue: \"my_queue\", callback: handler)\n"
" }\n"
" ```\n"
).
-spec consumer_supervised(
gleam@erlang@process:name(gleam@otp@factory_supervisor:message(consumer_config(), binary()))
) -> gleam@otp@supervision:child_specification(gleam@otp@factory_supervisor:supervisor(consumer_config(), binary())).
consumer_supervised(Name) ->
_pipe = gleam@otp@factory_supervisor:worker_child(
fun start_consumer_actor/1
),
_pipe@1 = gleam@otp@factory_supervisor:named(_pipe, Name),
gleam@otp@factory_supervisor:supervised(_pipe@1).
-file("src/carotte.gleam", 1069).
-spec header_value_to_header_tuple(header_value()) -> {gleam@erlang@atom:atom_(),
gleam@dynamic:dynamic_()}.
header_value_to_header_tuple(Value) ->
case Value of
{bool_header, Inner} ->
{erlang:binary_to_atom(<<"bool"/utf8>>),
gleam@function:identity(Inner)};
{float_header, Inner@1} ->
{erlang:binary_to_atom(<<"float"/utf8>>),
gleam@function:identity(Inner@1)};
{int_header, Inner@2} ->
{erlang:binary_to_atom(<<"long"/utf8>>),
gleam@function:identity(Inner@2)};
{string_header, Inner@3} ->
{erlang:binary_to_atom(<<"longstr"/utf8>>),
gleam@function:identity(Inner@3)};
{list_header, Inner@4} ->
Mapped = begin
_pipe = gleam@list:map(
Inner@4,
fun header_value_to_header_tuple/1
),
gleam@function:identity(_pipe)
end,
{erlang:binary_to_atom(<<"array"/utf8>>), Mapped}
end.
-file("src/carotte.gleam", 1101).
?DOC(
" Create a HeaderList from a list of name-value pairs.\n"
" Use this to construct headers for messages.\n"
"\n"
" ## Example\n"
" ```gleam\n"
" let headers = headers_from_list([\n"
" #(\"user_id\", StringHeader(\"123\")),\n"
" #(\"retry_count\", IntHeader(3)),\n"
" #(\"is_test\", BoolHeader(True)),\n"
" ])\n"
" ```\n"
).
-spec headers_from_list(list({binary(), header_value()})) -> header_list().
headers_from_list(List) ->
_pipe = List,
_pipe@1 = gleam@list:map(
_pipe,
fun(Item) ->
{Name, Value} = Item,
{Type_atom, Value@1} = header_value_to_header_tuple(Value),
{Name, Type_atom, Value@1}
end
),
{header_list, _pipe@1}.
-file("src/carotte.gleam", 1172).
?DOC(" Decoder for AMQP header arrays (list of {Type, Value} tuples)\n").
-spec header_array_decoder() -> gleam@dynamic@decode:decoder(header_value()).
header_array_decoder() ->
_pipe = gleam@dynamic@decode:list(header_value_decoder()),
gleam@dynamic@decode:map(_pipe, fun(Field@0) -> {list_header, Field@0} end).
-file("src/carotte.gleam", 1136).
?DOC(" Decoder for a single HeaderValue from AMQP {Type, Value} tuple format\n").
-spec header_value_decoder() -> gleam@dynamic@decode:decoder(header_value()).
header_value_decoder() ->
gleam@dynamic@decode:field(
0,
gleam@erlang@atom:decoder(),
fun(Type_atom) ->
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_dynamic/1},
fun(Value) ->
Type_name = erlang:atom_to_binary(Type_atom),
case Type_name of
<<"bool"/utf8>> ->
case gleam@dynamic@decode:run(
Value,
{decoder,
fun gleam@dynamic@decode:decode_bool/1}
) of
{ok, B} ->
gleam@dynamic@decode:success(
{bool_header, B}
);
{error, _} ->
gleam@dynamic@decode:failure(
{bool_header, false},
<<"Expected bool"/utf8>>
)
end;
<<"long"/utf8>> ->
case gleam@dynamic@decode:run(
Value,
{decoder, fun gleam@dynamic@decode:decode_int/1}
) of
{ok, I} ->
gleam@dynamic@decode:success(
{int_header, I}
);
{error, _} ->
gleam@dynamic@decode:failure(
{int_header, 0},
<<"Expected int"/utf8>>
)
end;
<<"float"/utf8>> ->
case gleam@dynamic@decode:run(
Value,
{decoder,
fun gleam@dynamic@decode:decode_float/1}
) of
{ok, F} ->
gleam@dynamic@decode:success(
{float_header, F}
);
{error, _} ->
gleam@dynamic@decode:failure(
{float_header, +0.0},
<<"Expected float"/utf8>>
)
end;
<<"longstr"/utf8>> ->
case gleam@dynamic@decode:run(
Value,
{decoder,
fun gleam@dynamic@decode:decode_string/1}
) of
{ok, S} ->
gleam@dynamic@decode:success(
{string_header, S}
);
{error, _} ->
gleam@dynamic@decode:failure(
{string_header, <<""/utf8>>},
<<"Expected string"/utf8>>
)
end;
<<"array"/utf8>> ->
case gleam@dynamic@decode:run(
Value,
header_array_decoder()
) of
{ok, List_header} ->
gleam@dynamic@decode:success(List_header);
{error, _} ->
gleam@dynamic@decode:failure(
{list_header, []},
<<"Expected array"/utf8>>
)
end;
_ ->
gleam@dynamic@decode:failure(
{bool_header, false},
<<"Unknown header type: "/utf8,
Type_name/binary>>
)
end
end
)
end
).
-file("src/carotte.gleam", 1121).
?DOC(
" Convert a HeaderList back to a list of name-value pairs.\n"
" Use this to read headers from received messages.\n"
"\n"
" ## Example\n"
" ```gleam\n"
" case carotte.subscribe(supervisor, channel, \"my_queue\", fn(payload, _deliver) {\n"
" let headers = carotte.headers_to_list(payload.headers)\n"
" // headers: List(#(String, HeaderValue))\n"
" })\n"
" ```\n"
).
-spec headers_to_list(header_list()) -> list({binary(), header_value()}).
headers_to_list(Headers) ->
{header_list, Raw_headers} = Headers,
_pipe = Raw_headers,
gleam@list:filter_map(
_pipe,
fun(Header) ->
{Name, Type_atom, Value} = Header,
Typed_value = gleam@function:identity({Type_atom, Value}),
case gleam@dynamic@decode:run(Typed_value, header_value_decoder()) of
{ok, Header_value} ->
{ok, {Name, Header_value}};
{error, _} ->
{error, nil}
end
end
).