-module(carotte@queue). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/carotte/queue.gleam"). -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, ack/3, ack_single/2, unsubscribe/2, unsubscribe_async/2, subscribe/3, subscribe_with_options/4]). -export_type([queue/0, deliver/0, queue_option/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 queue_option() :: {auto_ack, boolean()}. -type payload() :: {payload, binary(), list(carotte@publisher:publish_option()), carotte@publisher:header_list()}. -type declared_queue() :: {declared_queue, binary(), integer(), integer()}. -file("src/carotte/queue.gleam", 48). ?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", 53). ?DOC(" If set, the queue must already exist on the broker\n"). -spec as_passive(queue()) -> queue(). as_passive(Queue) -> {queue, erlang:element(2, Queue), true, erlang:element(4, Queue), erlang:element(5, Queue), erlang:element(6, Queue), erlang:element(7, Queue)}. -file("src/carotte/queue.gleam", 58). ?DOC(" If set, the queue will survive a broker restart\n"). -spec as_durable(queue()) -> queue(). as_durable(Queue) -> {queue, erlang:element(2, Queue), erlang:element(3, Queue), true, erlang:element(5, Queue), erlang:element(6, Queue), erlang:element(7, Queue)}. -file("src/carotte/queue.gleam", 63). ?DOC(" If set, only one subscriber can consume from the Queue\n"). -spec as_exclusive(queue()) -> queue(). as_exclusive(Queue) -> {queue, erlang:element(2, Queue), erlang:element(3, Queue), erlang:element(4, Queue), true, erlang:element(6, Queue), erlang:element(7, Queue)}. -file("src/carotte/queue.gleam", 68). ?DOC(" If set, the queue will be deleted when the last subscriber disconnect\n"). -spec with_auto_delete(queue()) -> queue(). with_auto_delete(Queue) -> {queue, erlang:element(2, Queue), erlang:element(3, Queue), erlang:element(4, Queue), erlang:element(5, Queue), true, erlang:element(7, Queue)}. -file("src/carotte/queue.gleam", 73). ?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", 100). ?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", 129). ?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", 139). ?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", 160). ?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", 170). ?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", 190). ?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", 208). ?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", 216). ?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", 235). ?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", 437). ?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" " queue.subscribe_with_options(\n" " channel: ch,\n" " queue: \"my_queue\",\n" " options: [queue.AutoAck(False)],\n" " callback: fn(msg, meta) {\n" " // Process message\n" " let _ = queue.ack(ch, meta.delivery_tag, False)\n" " },\n" " )\n" " ```\n" ). -spec ack(carotte@channel:channel(), integer(), boolean()) -> {ok, nil} | {error, carotte:carotte_error()}. ack(Channel, Delivery_tag, Multiple) -> carotte_ffi:ack(Channel, Delivery_tag, Multiple). -file("src/carotte/queue.gleam", 447). ?DOC( " Acknowledge a message delivery (acknowledges only this message).\n" " Convenience function for ack with multiple=False.\n" ). -spec ack_single(carotte@channel:channel(), integer()) -> {ok, nil} | {error, carotte:carotte_error()}. ack_single(Channel, Delivery_tag) -> carotte_ffi:ack(Channel, Delivery_tag, false). -file("src/carotte/queue.gleam", 455). ?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", 463). ?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", 477). -spec add_if_some(list(IHQ), fun((IHR) -> IHQ), gleam@option:option(IHR)) -> list(IHQ). add_if_some(List, Constructor, Value) -> case Value of {some, V} -> [Constructor(V) | List]; none -> List end. -file("src/carotte/queue.gleam", 294). -spec do_consume(fun((payload(), deliver()) -> any())) -> nil. do_consume(Fun) -> {Basic_deliver@2, Payload@3} = begin _pipe = gleam_erlang_ffi:new_selector(), _pipe@1 = gleam@erlang@process:select_record( _pipe, erlang:binary_to_atom(<<"basic.cancel"/utf8>>), 2, fun(_) -> gleam@erlang@process:send_exit(erlang:self()), erlang:error(#{gleam_error => panic, message => <<"`panic` expression evaluated."/utf8>>, file => <>, module => <<"carotte/queue"/utf8>>, function => <<"do_consume"/utf8>>, line => 299}) end ), _pipe@2 = gleam@erlang@process:select_record( _pipe@1, erlang:binary_to_atom(<<"basic.cancel_ok"/utf8>>), 1, fun(_) -> gleam@erlang@process:send_exit(erlang:self()), erlang:error(#{gleam_error => panic, message => <<"`panic` expression evaluated."/utf8>>, file => <>, module => <<"carotte/queue"/utf8>>, function => <<"do_consume"/utf8>>, line => 306}) end ), _pipe@3 = gleam@erlang@process:select_other( _pipe@2, fun(Delivery_dyn) -> Basic_deliver_decoder = begin 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 ) 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(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 ) -> 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, 1], Payload_properties_decoder, fun(Properties@13) -> gleam@dynamic@decode:subfield( [1, 2], {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Payload) -> gleam@dynamic@decode:subfield( [1, 1, 3], {decoder, fun gleam@dynamic@decode:decode_dynamic/1}, fun(Raw_headers) -> Headers = carotte_ffi:parse_amqp_headers( Raw_headers ), gleam@dynamic@decode:success( {payload, Payload, Properties@13, Headers} ) end ) end ) end ) end, 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 => <>, module => <<"carotte/queue"/utf8>>, function => <<"do_consume"/utf8>>, line => 396, value => _assert_fail, start => 11471, 'end' => 11557, pattern_start => 11482, pattern_end => 11499}) end, Payload@2 = case gleam@dynamic@decode:run( Delivery_dyn, Payload_decoder ) of {ok, Payload@1} -> Payload@1; _assert_fail@1 -> erlang:error(#{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"carotte/queue"/utf8>>, function => <<"do_consume"/utf8>>, line => 398, value => _assert_fail@1, start => 11564, 'end' => 11630, pattern_start => 11575, pattern_end => 11586}) end, Decoded = {Basic_deliver@1, Payload@2}, Decoded end ), gleam_erlang_ffi:select(_pipe@3) end, Fun(Payload@3, Basic_deliver@2), do_consume(Fun). -file("src/carotte/queue.gleam", 278). -spec do_start_consumer(fun((payload(), deliver()) -> any())) -> nil. do_start_consumer(Fun) -> _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), do_consume(Fun). -file("src/carotte/queue.gleam", 245). ?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" " Returns the consumer tag which can be used to unsubscribe\n" ). -spec subscribe( carotte@channel:channel(), binary(), fun((payload(), deliver()) -> nil) ) -> {ok, binary()} | {error, carotte:carotte_error()}. subscribe(Channel, Queue, Fun) -> Consumer_pid = proc_lib:spawn_link(fun() -> do_start_consumer(Fun) end), gleam_erlang_ffi:sleep(10), carotte_ffi:consume(Channel, Queue, Consumer_pid, true). -file("src/carotte/queue.gleam", 256). -spec subscribe_with_options( carotte@channel:channel(), binary(), list(queue_option()), fun((payload(), deliver()) -> nil) ) -> {ok, binary()} | {error, carotte:carotte_error()}. subscribe_with_options(Channel, Queue, Options, Fun) -> No_ack = case Options of [] -> true; [{auto_ack, Auto_ack} | _] -> Auto_ack end, Consumer_pid = proc_lib:spawn_link(fun() -> do_start_consumer(Fun) end), carotte_ffi:consume(Channel, Queue, Consumer_pid, No_ack).