Current section
Files
Jump to
Current section
Files
src/carotte@queue.erl
-module(carotte@queue).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([new/1, as_passive/1, as_durable/1, as_exclusive/1, with_auto_delete/1, declare/2, declare_async/2, delete/4, delete_async/4, bind/4, bind_async/4, unbind/4, purge/2, purge_async/2, status/2, unsubscribe/2, unsubscribe_async/2, subscribe/3]).
-export_type([queue/0, deliver/0, payload/0, declared_queue/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.
-opaque queue() :: {queue,
binary(),
boolean(),
boolean(),
boolean(),
boolean(),
boolean()}.
-type deliver() :: {deliver, binary(), integer(), boolean(), binary(), binary()}.
-type payload() :: {payload, binary(), list(carotte@publisher:publish_option())}.
-type declared_queue() :: {declared_queue, binary(), integer(), integer()}.
-file("src/carotte/queue.gleam", 40).
?DOC(" Create a new queue with the given name\n").
-spec new(binary()) -> queue().
new(Name) ->
{queue, Name, false, false, false, false, false}.
-file("src/carotte/queue.gleam", 45).
?DOC(" If set, the queue must already exist on the broker\n").
-spec as_passive(queue()) -> queue().
as_passive(Queue) ->
_record = Queue,
{queue,
erlang:element(2, _record),
true,
erlang:element(4, _record),
erlang:element(5, _record),
erlang:element(6, _record),
erlang:element(7, _record)}.
-file("src/carotte/queue.gleam", 50).
?DOC(" If set, the queue will survive a broker restart\n").
-spec as_durable(queue()) -> queue().
as_durable(Queue) ->
_record = Queue,
{queue,
erlang:element(2, _record),
erlang:element(3, _record),
true,
erlang:element(5, _record),
erlang:element(6, _record),
erlang:element(7, _record)}.
-file("src/carotte/queue.gleam", 55).
?DOC(" If set, only one subscriber can consume from the Queue\n").
-spec as_exclusive(queue()) -> queue().
as_exclusive(Queue) ->
_record = Queue,
{queue,
erlang:element(2, _record),
erlang:element(3, _record),
erlang:element(4, _record),
true,
erlang:element(6, _record),
erlang:element(7, _record)}.
-file("src/carotte/queue.gleam", 60).
?DOC(" If set, the queue will be deleted when the last subscriber disconnect\n").
-spec with_auto_delete(queue()) -> queue().
with_auto_delete(Queue) ->
_record = Queue,
{queue,
erlang:element(2, _record),
erlang:element(3, _record),
erlang:element(4, _record),
erlang:element(5, _record),
true,
erlang:element(7, _record)}.
-file("src/carotte/queue.gleam", 65).
?DOC(" Declare a queue on the broker\n").
-spec declare(queue(), carotte@channel:channel()) -> {ok, declared_queue()} |
{error, carotte:carotte_error()}.
declare(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/queue.gleam", 92).
?DOC(" Declare a queue on the broker asynchronously\n").
-spec declare_async(queue(), carotte@channel:channel()) -> {ok, nil} |
{error, carotte:carotte_error()}.
declare_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/queue.gleam", 121).
?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"
).
-spec delete(carotte@channel:channel(), binary(), boolean(), boolean()) -> {ok,
integer()} |
{error, carotte:carotte_error()}.
delete(Channel, Queue, If_unused, If_empty) ->
carotte_ffi:queue_delete(Channel, Queue, If_unused, If_empty, false).
-file("src/carotte/queue.gleam", 131).
?DOC(" Delete a queue from the broker asynchronously. Same semantics as `delete`\n").
-spec delete_async(carotte@channel:channel(), binary(), boolean(), boolean()) -> {ok,
nil} |
{error, carotte:carotte_error()}.
delete_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/queue.gleam", 152).
?DOC(
" Bind a queue to an exchange\n"
" The `routing_key` is used to filter messages from the exchange\n"
).
-spec bind(carotte@channel:channel(), binary(), binary(), binary()) -> {ok, nil} |
{error, carotte:carotte_error()}.
bind(Channel, Queue, Exchange, Routing_key) ->
carotte_ffi:queue_bind(Channel, Queue, Exchange, Routing_key, false).
-file("src/carotte/queue.gleam", 162).
?DOC(" Bind a queue to an exchange asynchronously. Same semantics as `bind`\n").
-spec bind_async(carotte@channel:channel(), binary(), binary(), binary()) -> {ok,
nil} |
{error, carotte:carotte_error()}.
bind_async(Channel, Queue, Exchange, Routing_key) ->
carotte_ffi:queue_bind(Channel, Queue, Exchange, Routing_key, true).
-file("src/carotte/queue.gleam", 182).
?DOC(
" Unbind a queue from an exchange\n"
" The `routing_key` is used to filter messages from the exchange\n"
).
-spec unbind(carotte@channel:channel(), binary(), binary(), binary()) -> {ok,
nil} |
{error, carotte:carotte_error()}.
unbind(Channel, Queue, Exchange, Routing_key) ->
carotte_ffi:queue_unbind(Channel, Queue, Exchange, Routing_key).
-file("src/carotte/queue.gleam", 200).
?DOC(" Purge a queue of all messages\n").
-spec purge(carotte@channel:channel(), binary()) -> {ok, integer()} |
{error, carotte:carotte_error()}.
purge(Channel, Queue) ->
carotte_ffi:queue_purge(Channel, Queue, false).
-file("src/carotte/queue.gleam", 208).
?DOC(" Purge a queue of all messages asynchronously\n").
-spec purge_async(carotte@channel:channel(), binary()) -> {ok, nil} |
{error, carotte:carotte_error()}.
purge_async(Channel, Queue) ->
gleam@result:map(
carotte_ffi:queue_purge(Channel, Queue, true),
fun(_) -> nil end
).
-file("src/carotte/queue.gleam", 224).
?DOC(" Get the status of a queue\n").
-spec status(carotte@channel:channel(), binary()) -> {ok, declared_queue()} |
{error, carotte:carotte_error()}.
status(Channel, Queue) ->
carotte_ffi:queue_declare(Channel, Queue, true, false, false, false, false).
-file("src/carotte/queue.gleam", 243).
-spec consume(carotte@channel:channel(), binary(), gleam@erlang@process:pid_()) -> {ok,
binary()} |
{error, carotte:carotte_error()}.
consume(Channel, Queue, Pid) ->
carotte_ffi:consume(Channel, Queue, Pid).
-file("src/carotte/queue.gleam", 379).
?DOC(" Unsubscribe a consumer from a queue\n").
-spec unsubscribe(carotte@channel:channel(), binary()) -> {ok, nil} |
{error, carotte:carotte_error()}.
unsubscribe(Channel, Consumer_tag) ->
carotte_ffi:unsubscribe(Channel, Consumer_tag, false).
-file("src/carotte/queue.gleam", 387).
?DOC(" Unsubscribe a consumer from a queue asynchronously\n").
-spec unsubscribe_async(carotte@channel:channel(), binary()) -> {ok, nil} |
{error, carotte:carotte_error()}.
unsubscribe_async(Channel, Consumer_tag) ->
carotte_ffi:unsubscribe(Channel, Consumer_tag, true).
-file("src/carotte/queue.gleam", 401).
-spec add_if_some(list(FTB), fun((FTC) -> FTB), gleam@option:option(FTC)) -> list(FTB).
add_if_some(List, Constructor, Value) ->
case Value of
{some, V} ->
[Constructor(V) | List];
none ->
List
end.
-file("src/carotte/queue.gleam", 273).
-spec do_consume(
carotte@channel:channel(),
fun((payload(), deliver()) -> any())
) -> any().
do_consume(Channel, Fun) ->
{Basic_deliver@1, Payload@2} = begin
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:selecting_record2(
_pipe,
erlang:binary_to_atom(<<"basic.cancel"/utf8>>),
fun(_) ->
gleam@erlang@process:send_exit(erlang:self()),
erlang:error(#{gleam_error => panic,
message => <<"`panic` expression evaluated."/utf8>>,
module => <<"carotte/queue"/utf8>>,
function => <<"do_consume"/utf8>>,
line => 280})
end
),
_pipe@2 = gleam@erlang@process:selecting_record2(
_pipe@1,
erlang:binary_to_atom(<<"basic.cancel_ok"/utf8>>),
fun(_) ->
gleam@erlang@process:send_exit(erlang:self()),
erlang:error(#{gleam_error => panic,
message => <<"`panic` expression evaluated."/utf8>>,
module => <<"carotte/queue"/utf8>>,
function => <<"do_consume"/utf8>>,
line => 287})
end
),
_pipe@3 = gleam@erlang@process:selecting_anything(
_pipe@2,
fun(A) ->
Basic_deliver_decoder = begin
gleam@dynamic@decode:subfield(
[1],
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Consumer_tag) ->
gleam@dynamic@decode:subfield(
[2],
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Delivery_tag) ->
gleam@dynamic@decode:subfield(
[3],
{decoder,
fun gleam@dynamic@decode:decode_bool/1},
fun(Redelivered) ->
gleam@dynamic@decode:subfield(
[4],
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Exchange) ->
gleam@dynamic@decode:subfield(
[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
)
end,
Payload_properties_decoder = begin
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(Persistent) ->
Properties@3 = add_if_some(
Properties@2,
fun(Field@0) -> {persistent, Field@0} end,
case Persistent of
{some, 1} ->
{some, true};
{some, 2} ->
{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
) ->
Properties@7 = add_if_some(
Properties@6,
fun(Field@0) -> {expiration, Field@0} end,
Expiration
),
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
) ->
Properties@9 = add_if_some(
Properties@8,
fun(Field@0) -> {timestamp, Field@0} end,
Timestamp
),
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
)
end,
Payload_decoder = begin
gleam@dynamic@decode:subfield(
[1],
Payload_properties_decoder,
fun(Properties@13) ->
gleam@dynamic@decode:subfield(
[2],
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Payload) ->
gleam@dynamic@decode:success(
{payload, Payload, Properties@13}
)
end
)
end
)
end,
Decoder = begin
gleam@dynamic@decode:subfield(
[0],
Basic_deliver_decoder,
fun(Basic_deliver) ->
gleam@dynamic@decode:subfield(
[1],
Payload_decoder,
fun(Payload@1) ->
gleam@dynamic@decode:success(
{Basic_deliver, Payload@1}
)
end
)
end
)
end,
_assert_subject = gleam@dynamic@decode:run(A, Decoder),
{ok, Decoded} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"carotte/queue"/utf8>>,
function => <<"do_consume"/utf8>>,
line => 362})
end,
Decoded
end
),
gleam_erlang_ffi:select(_pipe@3)
end,
Fun(Payload@2, Basic_deliver@1),
carotte_ffi:ack(Channel, erlang:element(3, Basic_deliver@1), false),
do_consume(Channel, Fun).
-file("src/carotte/queue.gleam", 258).
-spec do_start_consumer(
carotte@channel:channel(),
fun((payload(), deliver()) -> any())
) -> any().
do_start_consumer(Channel, Fun) ->
_ = begin
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:selecting_record2(
_pipe,
erlang:binary_to_atom(<<"basic.consume_ok"/utf8>>),
fun(Dyn) ->
_assert_subject = gleam@dynamic@decode:run(
Dyn,
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
{ok, Consumer_tag} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"carotte/queue"/utf8>>,
function => <<"do_start_consumer"/utf8>>,
line => 264})
end,
Consumer_tag
end
),
gleam_erlang_ffi:select(_pipe@1)
end,
do_consume(Channel, Fun).
-file("src/carotte/queue.gleam", 233).
?DOC(
" Subscribe to a queue\n"
" The `callback` function will be called with each message received, receiving the message Payload and a `Deliver` struct\n"
).
-spec subscribe(
carotte@channel:channel(),
binary(),
fun((payload(), deliver()) -> nil)
) -> {ok, binary()} | {error, carotte:carotte_error()}.
subscribe(Channel, Queue, Fun) ->
Consumer_pid = gleam@erlang@process:start(
fun() -> do_start_consumer(Channel, Fun) end,
false
),
consume(Channel, Queue, Consumer_pid).