-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]). -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) -> {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", 50). ?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", 55). ?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", 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) -> {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", 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) -> begin Result = carotte_ffi:queue_delete( Channel, Queue, If_unused, If_empty, true ), case Result of {ok, X} -> {ok, begin _@1 = X, nil end}; {error, E} -> {error, E} end 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) -> begin Result = carotte_ffi:queue_purge(Channel, Queue, true), case Result of {ok, X} -> {ok, begin _@1 = X, nil end}; {error, E} -> {error, E} end 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", 242). -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", 392). ?DOC( " Acknowledge a message delivery.\n" " Used when manual acknowledgment is enabled (NoAck(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.NoAck(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", 402). ?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", 410). ?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", 418). ?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", 432). -spec add_if_some(list(IIG), fun((IIH) -> IIG), gleam@option:option(IIH)) -> list(IIG). add_if_some(List, Constructor, Value) -> case Value of {some, V} -> [Constructor(V) | List]; none -> List end. -file("src/carotte/queue.gleam", 269). -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: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 => 274}) end ), _pipe@2 = gleam@erlang@process:select_record( _pipe@1, erlang:binary_to_atom(<<"basic.cancel_ok"/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 => 281}) end ), _pipe@3 = gleam@erlang@process:select_other( _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, Decoded@1 = case gleam@dynamic@decode:run(A, Decoder) of {ok, Decoded} -> Decoded; _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 => 356, value => _assert_fail, start => 10456, 'end' => 10503, pattern_start => 10467, pattern_end => 10478}) end, Decoded@1 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", 257). -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:select_record( _pipe, erlang:binary_to_atom(<<"basic.consume_ok"/utf8>>), 2, fun(Dyn) -> Consumer_tag@1 = case gleam@dynamic@decode:run( Dyn, {decoder, fun gleam@dynamic@decode:decode_string/1} ) of {ok, Consumer_tag} -> Consumer_tag; _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_start_consumer"/utf8>>, line => 261, value => _assert_fail, start => 6595, 'end' => 6655, pattern_start => 6606, pattern_end => 6622}) end, Consumer_tag@1 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 = proc_lib:spawn_link( fun() -> do_start_consumer(Channel, Fun) end ), consume(Channel, Queue, Consumer_pid).