defmodule Pulsar.Consumer do @moduledoc """ Pulsar consumer process that communicates with broker processes. This consumer uses service discovery to find the appropriate broker for the topic and then communicates with that broker process. ## Callback Module The consumer requires a callback module that uses `Pulsar.Consumer.Callback`, providing stateful message processing capabilities. To create a callback module: defmodule MyApp.MessageHandler do use Pulsar.Consumer.Callback def handle_message(%Pulsar.Message{payload: payload}, state) do # Process the message IO.inspect(payload) {:ok, state} end end See `Pulsar.Consumer.Callback` for detailed documentation and examples. """ use GenServer alias Pulsar.Config alias Pulsar.Protocol.Binary.Pulsar.Proto, as: Binary alias Pulsar.ServiceDiscovery require Logger defstruct [ :client, :topic, :subscription_name, :subscription_type, :consumer_id, :consumer_name, :callback_module, :callback_state, :broker_pid, :broker_monitor, :flow_initial, :flow_threshold, :flow_refill, :flow_outstanding_permits, :initial_position, :durable, :force_create_topic, :start_message_id, :start_timestamp, :nacked_messages, :redelivery_interval, :max_redelivery, :dead_letter_topic, :dead_letter_producer_pid ] @type t :: %__MODULE__{ topic: String.t(), subscription_name: String.t(), subscription_type: String.t(), consumer_id: integer(), consumer_name: String.t() | nil, callback_module: module(), callback_state: term(), broker_pid: pid(), broker_monitor: reference(), flow_initial: non_neg_integer(), flow_threshold: non_neg_integer(), flow_refill: non_neg_integer(), flow_outstanding_permits: non_neg_integer(), initial_position: atom(), force_create_topic: boolean(), start_message_id: {non_neg_integer(), non_neg_integer()}, start_timestamp: non_neg_integer(), nacked_messages: MapSet.t(), redelivery_interval: non_neg_integer() | nil, max_redelivery: non_neg_integer() | nil, dead_letter_topic: String.t() | nil, dead_letter_producer_pid: pid() | nil } ## Public API @doc """ Starts a consumer process with explicit parameters. ## Parameters - `topic` - The topic to subscribe to - `subscription_name` - Name of the subscription - `subscription_type` - Type of subscription (e.g., :Exclusive, :Shared) - `callback_module` - Module that uses `Pulsar.Consumer.Callback` - `opts` - Additional options: - `:init_args` - Arguments passed to callback module's init/1 function - `:flow_initial` - Initial flow permits (default: 100). Set to 0 to disable automatic flow control and use `send_flow/2` manually. - `:flow_threshold` - Flow permits threshold for automatic refill (default: 50). Ignored when `:flow_initial` is 0. - `:flow_refill` - Flow permits refill amount (default: 50). Ignored when `:flow_initial` is 0. - `:initial_position` - Initial position for subscription (`:latest` or `:earliest`, defaults to `:latest`) - `:redelivery_interval` - Interval in milliseconds for redelivering NACKed messages (default: nil, disabled) - `:dead_letter_policy` - Dead letter policy configuration (default: nil, disabled): - `:max_redelivery` - Maximum number of redeliveries before sending to dead letter topic (must be >= 1) - `:topic` - Dead letter topic (optional, defaults to `--DLQ`) - `:startup_delay_ms` - Fixed startup delay in milliseconds before consumer initialization (default: 1000, matches broker conn_timeout) - `:startup_jitter_ms` - Maximum random startup delay in milliseconds to avoid thundering herd (default: 1000) The total startup delay is `startup_delay_ms + random(0, startup_jitter_ms)`, applied on every consumer start/restart. The default `startup_delay_ms` matches the broker's `conn_timeout` to ensure the broker has time to reconnect before consumers start requesting topic lookups. The consumer will automatically use any available broker for service discovery. """ def start_link(topic, subscription_name, subscription_type, callback_module, opts \\ []) do {init_args, genserver_opts} = Keyword.pop(opts, :init_args, []) {initial_permits, genserver_opts} = Keyword.pop(genserver_opts, :flow_initial, 100) {refill_threshold, genserver_opts} = Keyword.pop(genserver_opts, :flow_threshold, 50) {refill_amount, genserver_opts} = Keyword.pop(genserver_opts, :flow_refill, 50) {initial_position, genserver_opts} = Keyword.pop(genserver_opts, :initial_position, :latest) {durable, genserver_opts} = Keyword.pop(genserver_opts, :durable, true) {force_create_topic, genserver_opts} = Keyword.pop(genserver_opts, :force_create_topic, true) {start_message_id, genserver_opts} = Keyword.pop(genserver_opts, :start_message_id, nil) {start_timestamp, genserver_opts} = Keyword.pop(genserver_opts, :start_timestamp, nil) {redelivery_interval, genserver_opts} = Keyword.pop(genserver_opts, :redelivery_interval, nil) {dead_letter_policy, genserver_opts} = Keyword.pop(genserver_opts, :dead_letter_policy, nil) {startup_delay_ms, genserver_opts} = Keyword.pop(genserver_opts, :startup_delay_ms, Config.startup_delay()) {startup_jitter_ms, genserver_opts} = Keyword.pop(genserver_opts, :startup_jitter_ms, Config.startup_jitter()) {client, _genserver_opts} = Keyword.pop(genserver_opts, :client, :default) consumer_config = %{ client: client, topic: topic, subscription_name: subscription_name, subscription_type: subscription_type, callback_module: callback_module, init_args: init_args, flow_initial: initial_permits, flow_threshold: refill_threshold, flow_refill: refill_amount, initial_position: initial_position, durable: durable, force_create_topic: force_create_topic, start_message_id: start_message_id, start_timestamp: start_timestamp, redelivery_interval: redelivery_interval, dead_letter_policy: dead_letter_policy, startup_delay_ms: startup_delay_ms, startup_jitter_ms: startup_jitter_ms } GenServer.start_link(__MODULE__, consumer_config, []) end @doc """ Gracefully stops a consumer process. """ @spec stop(GenServer.server(), term(), timeout()) :: :ok def stop(consumer, reason \\ :normal, timeout \\ :infinity) do GenServer.stop(consumer, reason, timeout) end @doc """ Sends a flow command to request more messages from the broker. Use this function when you've disabled automatic flow control by setting `:flow_initial` to 0. This allows you to implement custom flow control, such as integrating with Broadway's demand mechanism. ## Parameters - `consumer` - The consumer process PID - `permits` - Number of message permits to request ## Examples # Start consumer with no automatic flow control {:ok, consumer_pid} = Pulsar.start_consumer( topic, subscription, MyCallback, flow_initial: 0 # Disable automatic flow ) # Manually request messages based on your own demand Pulsar.Consumer.send_flow(consumer_pid, 10) # Example with Broadway demand def handle_demand(demand, state) do Pulsar.Consumer.send_flow(state.consumer, demand) # ... rest of logic end """ @spec send_flow(pid(), non_neg_integer()) :: :ok | {:error, term()} def send_flow(consumer, permits) when is_integer(permits) and permits > 0 do GenServer.call(consumer, {:send_flow, permits}) end @doc """ Manually acknowledges one or more messages. Use this when your callback returns `{:noreply, state}` to manually control acknowledgment. Supports batching multiple message IDs in a single ACK command for better performance. ## Parameters - `consumer` - The consumer process PID - `message_ids` - A single message ID or a list of message IDs to acknowledge ## Examples # Acknowledge a single message def handle_message({command, _metadata, _payload, _broker_metadata}, state) do message_id = command.message_id # Process message... spawn(fn -> # Do async processing Pulsar.Consumer.ack(consumer_pid, message_id) end) {:noreply, state} end # Acknowledge multiple messages in batch (more efficient) Pulsar.Consumer.ack(consumer_pid, [message_id1, message_id2, message_id3]) """ @spec ack(pid(), Binary.MessageIdData.t() | [Binary.MessageIdData.t()]) :: :ok | {:error, term()} def ack(consumer, message_ids) when is_list(message_ids) do GenServer.call(consumer, {:ack, message_ids}) end def ack(consumer, message_id) do ack(consumer, [message_id]) end @doc """ Manually negatively acknowledges one or more messages. Use this when your callback returns `{:noreply, state}` to manually control acknowledgment. Supports batching multiple message IDs in a single NACK for better performance. The messages will be tracked for redelivery if `:redelivery_interval` is configured. When the messages are redelivered and the redelivery count exceeds `:max_redelivery`, they will automatically be sent to the dead letter queue (if `:dead_letter_policy` is configured), regardless of whether you use manual or automatic acknowledgment. ## Parameters - `consumer` - The consumer process PID - `message_ids` - A single message ID or a list of message IDs to negatively acknowledge ## Examples # NACK a single message def handle_message({command, _metadata, _payload, _broker_metadata}, state) do message_id = command.message_id case process_message() do :ok -> Pulsar.Consumer.ack(self(), message_id) {:error, _reason} -> Pulsar.Consumer.nack(self(), message_id) end {:noreply, state} end # NACK multiple messages in batch (more efficient) Pulsar.Consumer.nack(consumer_pid, [message_id1, message_id2, message_id3]) """ @spec nack(pid(), Binary.MessageIdData.t() | [Binary.MessageIdData.t()]) :: :ok | {:error, term()} def nack(consumer, message_ids) when is_list(message_ids) do GenServer.call(consumer, {:nack, message_ids}) end def nack(consumer, message_id) do nack(consumer, [message_id]) end ## GenServer Callbacks @impl true def init(consumer_config) do %{ client: client, topic: topic, subscription_name: subscription_name, subscription_type: subscription_type, callback_module: callback_module, init_args: init_args, flow_initial: initial_permits, flow_threshold: refill_threshold, flow_refill: refill_amount, initial_position: initial_position, durable: durable, force_create_topic: force_create_topic, start_message_id: start_message_id, start_timestamp: start_timestamp, redelivery_interval: redelivery_interval, dead_letter_policy: dead_letter_policy, startup_delay_ms: startup_delay_ms, startup_jitter_ms: startup_jitter_ms } = consumer_config {max_redelivery, dead_letter_topic} = parse_dead_letter_policy(dead_letter_policy) state = %__MODULE__{ client: client, consumer_id: System.unique_integer([:positive, :monotonic]), consumer_name: nil, topic: topic, subscription_name: subscription_name, subscription_type: subscription_type, callback_module: callback_module, flow_initial: initial_permits, flow_threshold: refill_threshold, flow_refill: refill_amount, flow_outstanding_permits: 0, initial_position: initial_position, durable: durable, force_create_topic: force_create_topic, start_message_id: start_message_id, start_timestamp: start_timestamp, nacked_messages: MapSet.new(), redelivery_interval: redelivery_interval, max_redelivery: max_redelivery, dead_letter_topic: dead_letter_topic, dead_letter_producer_pid: nil } Logger.info("Starting consumer for topic #{state.topic}") total_startup_delay = startup_delay_ms + startup_jitter_ms if total_startup_delay > 0 do {:ok, state, {:continue, {:startup_delay, startup_delay_ms, startup_jitter_ms, init_args}}} else {:ok, state, {:continue, {:subscribe, init_args}}} end end @impl true def handle_continue({:startup_delay, base_delay_ms, jitter_ms, init_args}, state) do jitter = if jitter_ms > 0, do: :rand.uniform(jitter_ms), else: 0 total_sleep_ms = base_delay_ms + jitter Logger.debug("Consumer sleeping for #{total_sleep_ms}ms (base: #{base_delay_ms}ms, jitter: #{jitter}ms)") Process.sleep(total_sleep_ms) {:noreply, state, {:continue, {:subscribe, init_args}}} end def handle_continue({:subscribe, init_args}, state) do with {:ok, broker_pid} <- ServiceDiscovery.lookup_topic(state.topic, client: state.client), :ok <- Pulsar.Broker.register_consumer(broker_pid, state.consumer_id, self()), {:ok, response} <- subscribe_to_topic( broker_pid, state.topic, state.subscription_name, state.subscription_type, state.consumer_id, initial_position: state.initial_position, durable: state.durable, force_create_topic: state.force_create_topic ) do consumer_name = Map.get(response, :consumer_name, "unknown") {:noreply, %{ state | consumer_id: state.consumer_id, consumer_name: consumer_name, broker_pid: broker_pid }, {:continue, {:seek_subscription, init_args}}} else {:error, reason} -> {:stop, reason, state} end end def handle_continue({:seek_subscription, init_args}, state) do case maybe_seek_subscription( state.broker_pid, state.consumer_id, state.start_message_id, state.start_timestamp ) do {:ok, :skipped} -> {:noreply, state, {:continue, {:send_initial_flow, init_args}}} {:ok, _response} -> {:noreply, state, {:continue, {:resubscribe, init_args}}} {:error, {:UnknownError, "Reset subscription to publish time error: Failed to fence subscription"}} -> {:noreply, state, {:continue, {:resubscribe, init_args}}} {:error, reason} -> {:stop, reason, state} end end def handle_continue({:resubscribe, init_args}, state) do receive do # When sending a Seek, we expect the broker to send a CloseConsumer {:broker_message, %Binary.CommandCloseConsumer{}} -> case subscribe_to_topic( state.broker_pid, state.topic, state.subscription_name, state.subscription_type, state.consumer_id, initial_position: state.initial_position, durable: state.durable, force_create_topic: state.force_create_topic ) do {:ok, _response} -> {:noreply, state, {:continue, {:send_initial_flow, init_args}}} {:error, reason} -> {:stop, reason, state} end after # TO-DO: Should be configurable 1_000 -> {:stop, :no_close_consumer, state} end end def handle_continue({:send_initial_flow, init_args}, state) do # Only send initial flow if flow_initial > 0 (automatic flow control enabled) result = if state.flow_initial > 0 do send_initial_flow(state.broker_pid, state.consumer_id, state.flow_initial) else # Manual flow control - don't send initial flow :ok end case result do :ok -> broker_monitor = Process.monitor(state.broker_pid) schedule_redelivery(state.redelivery_interval) {:noreply, %{ state | broker_monitor: broker_monitor, flow_outstanding_permits: state.flow_initial }, {:continue, {:init_dead_letter_producer, init_args}}} {:error, reason} -> {:stop, reason, state} end end def handle_continue({:init_dead_letter_producer, init_args}, state) do if should_init_dead_letter_producer?(state) do case start_dead_letter_producer(state) do {:ok, producer_pid} -> Logger.info("Started dead letter producer for consumer on topic #{state.topic}") {:noreply, %{state | dead_letter_producer_pid: producer_pid}, {:continue, {:init_callback, init_args}}} {:error, reason} -> Logger.error("Failed to start dead letter producer: #{inspect(reason)}") {:stop, reason, state} end else {:noreply, state, {:continue, {:init_callback, init_args}}} end end def handle_continue({:init_callback, init_args}, state) do case state.callback_module.init(init_args) do {:ok, callback_state} -> {:noreply, %{state | callback_state: callback_state}} {:error, reason} -> {:stop, reason, nil} end end def message_id(%Binary.CommandMessage{} = message) do message.message_id end @impl true def handle_info({:broker_message, %Binary.CommandCloseConsumer{}}, state) do {:stop, :broker_close_requested, state} end def handle_info({:broker_message, {command, metadata, payload, broker_metadata}}, state) do base_message_id = message_id(command) redelivery_count = Map.get(command, :redelivery_count, 0) payload = maybe_uncompress(metadata, payload) payload = unwrap_messages(metadata, payload) num_messages = length(payload) state = decrement_permits(state, num_messages) new_state = process_messages_normally(state, command, metadata, payload, broker_metadata, base_message_id, redelivery_count) new_state = maybe_send_batch_to_dead_letter(new_state, payload, base_message_id, redelivery_count) {:noreply, new_state} end @impl true def handle_info(:trigger_redelivery, state) do new_state = if MapSet.size(state.nacked_messages) > 0 do nacked_list = MapSet.to_list(state.nacked_messages) redeliver_command = %Binary.CommandRedeliverUnacknowledgedMessages{ consumer_id: state.consumer_id, message_ids: nacked_list } :ok = Pulsar.Broker.send_command(state.broker_pid, redeliver_command) Logger.warning("Requested redelivery of #{length(nacked_list)} NACKed messages") %{state | nacked_messages: MapSet.new()} else state end schedule_redelivery(state.redelivery_interval) {:noreply, new_state} end @impl true def handle_info( {:DOWN, monitor_ref, :process, broker_pid, reason}, %__MODULE__{broker_monitor: monitor_ref, broker_pid: broker_pid} = state ) do Logger.info("Broker #{inspect(broker_pid)} crashed: #{inspect(reason)}, consumer will restart") {:stop, :broker_crashed, state} end # Handle other info messages by delegating to callback module @impl true def handle_info(message, state) do case state.callback_module.handle_info(message, state.callback_state) do {:noreply, new_callback_state} -> {:noreply, %{state | callback_state: new_callback_state}} {:noreply, new_callback_state, timeout_or_hibernate} -> {:noreply, %{state | callback_state: new_callback_state}, timeout_or_hibernate} {:stop, reason, new_callback_state} -> {:stop, reason, %{state | callback_state: new_callback_state}} end end defp maybe_send_batch_to_dead_letter(state, _payload, _base_message_id, redelivery_count) when is_nil(state.max_redelivery) or state.max_redelivery < 1 or redelivery_count < state.max_redelivery or is_nil(state.dead_letter_producer_pid) do state end defp maybe_send_batch_to_dead_letter(state, payload, base_message_id, redelivery_count) do Logger.warning( "Redelivery count of #{redelivery_count} exceeds max redelivery of #{state.max_redelivery}, sending batch to DLQ " ) nacked_ids = payload |> Enum.with_index() |> Enum.reduce([], fn {{_msg_metadata, msg_payload}, index}, nacked_acc -> message_id_to_ack = if index == 0 and length(payload) == 1 do base_message_id else %{base_message_id | batch_index: index} end case send_to_dead_letter(state, msg_payload, message_id_to_ack) do :ok -> # ACK the message since it's now in DLQ ack_command = %Binary.CommandAck{ consumer_id: state.consumer_id, ack_type: :Individual, message_id: [message_id_to_ack] } :ok = Pulsar.Broker.send_command(state.broker_pid, ack_command) nacked_acc {:error, dlq_reason} -> Logger.error("Failed to send message to dead letter topic: #{inspect(dlq_reason)}, leaving as nacked") [message_id_to_ack | nacked_acc] end end) # Add failed DLQ sends to nacked messages for redelivery new_nacked_messages = if state.redelivery_interval do MapSet.union(state.nacked_messages, MapSet.new(nacked_ids)) else state.nacked_messages end %{state | nacked_messages: new_nacked_messages} end defp process_messages_normally(state, command, metadata, payload, broker_metadata, base_message_id, redelivery_count) do {final_callback_state, nacked_ids} = payload |> Enum.with_index() |> Enum.reduce({state.callback_state, []}, fn {{msg_metadata, msg_payload}, index}, {callback_state, nacked_acc} -> message_id_to_ack = if msg_metadata == nil do base_message_id else %{base_message_id | batch_index: index} end msg_args = %Pulsar.Message{ command: command, metadata: metadata, payload: msg_payload, single_metadata: msg_metadata, broker_metadata: broker_metadata, message_id_to_ack: message_id_to_ack } process_single_message(state, msg_args, callback_state, message_id_to_ack, nacked_acc, redelivery_count) end) state = check_and_refill_permits(state) new_nacked_messages = if state.redelivery_interval do MapSet.union(state.nacked_messages, MapSet.new(nacked_ids)) else state.nacked_messages end %{ state | callback_state: final_callback_state, nacked_messages: new_nacked_messages } end defp process_single_message(state, msg_args, callback_state, message_id_to_ack, nacked_acc, redelivery_count) do result = state.callback_module.handle_message(msg_args, callback_state) case result do {:ok, new_callback_state} -> ack_command = %Binary.CommandAck{ consumer_id: state.consumer_id, ack_type: :Individual, message_id: [message_id_to_ack] } :ok = Pulsar.Broker.send_command(state.broker_pid, ack_command) {new_callback_state, nacked_acc} {:noreply, new_callback_state} -> {new_callback_state, nacked_acc} {:error, reason, new_callback_state} -> Logger.warning( "Message processing failed: #{inspect(reason)}, tracking for redelivery (count: #{redelivery_count})" ) {new_callback_state, [message_id_to_ack | nacked_acc]} unexpected_result -> Logger.warning("Unexpected callback result: #{inspect(unexpected_result)}, not acknowledging") {callback_state, nacked_acc} end end @impl true def terminate(_reason, nil) do :ok end def terminate(reason, state) do try do state.callback_module.terminate(reason, state.callback_state) rescue error -> Logger.warning("Error in callback terminate function: #{inspect(error)}") end :ok end @impl true def handle_call({:send_flow, permits}, _from, state) do flow_command = %Binary.CommandFlow{ consumer_id: state.consumer_id, messagePermits: permits } :ok = Pulsar.Broker.send_command(state.broker_pid, flow_command) new_permits = state.flow_outstanding_permits + permits {:reply, :ok, %{state | flow_outstanding_permits: new_permits}} end def handle_call({:ack, message_ids}, _from, state) when is_list(message_ids) do ack_command = %Binary.CommandAck{ consumer_id: state.consumer_id, ack_type: :Individual, message_id: message_ids } :ok = Pulsar.Broker.send_command(state.broker_pid, ack_command) {:reply, :ok, state} end def handle_call({:nack, message_ids}, _from, state) when is_list(message_ids) do # Manual NACK follows the same pattern as auto-NACK: # - Add to nacked_messages if redelivery_interval is configured (for periodic redelivery) # - Note: DLQ logic cannot be applied here since we don't have redelivery_count or payload # from the broker. The DLQ will only trigger on subsequent redeliveries when the message # comes back through handle_message with an updated redelivery_count. new_nacked_messages = if state.redelivery_interval do # Track all messages for periodic redelivery Enum.reduce(message_ids, state.nacked_messages, fn message_id, acc -> MapSet.put(acc, message_id) end) else # No periodic redelivery configured, so we don't track nacked messages # Note: Without redelivery_interval, messages won't be automatically redelivered # and DLQ won't be triggered. Consider configuring :redelivery_interval and # :dead_letter_policy for production use. Logger.debug("NACKed #{length(message_ids)} message(s), but no redelivery_interval configured") state.nacked_messages end {:reply, :ok, %{state | nacked_messages: new_nacked_messages}} end def handle_call(request, from, state) do case state.callback_module.handle_call(request, from, state.callback_state) do {:reply, reply, new_callback_state} -> {:reply, reply, %{state | callback_state: new_callback_state}} {:reply, reply, new_callback_state, timeout_or_hibernate} -> {:reply, reply, %{state | callback_state: new_callback_state}, timeout_or_hibernate} {:noreply, new_callback_state} -> {:noreply, %{state | callback_state: new_callback_state}} {:noreply, new_callback_state, timeout_or_hibernate} -> {:noreply, %{state | callback_state: new_callback_state}, timeout_or_hibernate} {:stop, reason, reply, new_callback_state} -> {:stop, reason, reply, %{state | callback_state: new_callback_state}} {:stop, reason, new_callback_state} -> {:stop, reason, %{state | callback_state: new_callback_state}} end end @impl true def handle_cast(request, state) do case state.callback_module.handle_cast(request, state.callback_state) do {:noreply, new_callback_state} -> {:noreply, %{state | callback_state: new_callback_state}} {:noreply, new_callback_state, timeout_or_hibernate} -> {:noreply, %{state | callback_state: new_callback_state}, timeout_or_hibernate} {:stop, reason, new_callback_state} -> {:stop, reason, %{state | callback_state: new_callback_state}} end end ## Private Functions defp initial_position(:latest), do: :Latest defp initial_position(:earliest), do: :Earliest defp subscribe_to_topic(broker_pid, topic, subscription_name, subscription_type, consumer_id, opts) do request_id = System.unique_integer([:positive, :monotonic]) initial_position = opts |> Keyword.get(:initial_position) |> initial_position() durable = Keyword.get(opts, :durable, true) force_create_topic = Keyword.get(opts, :force_create_topic, true) subscribe_command = %Binary.CommandSubscribe{ topic: topic, subscription: subscription_name, subType: subscription_type, consumer_id: consumer_id, request_id: request_id, initialPosition: initial_position, durable: durable, force_topic_creation: force_create_topic } Pulsar.Broker.send_request(broker_pid, subscribe_command) end defp maybe_seek_subscription(_broker_pid, _consumer_id, nil, nil) do {:ok, :skipped} end defp maybe_seek_subscription(broker_pid, consumer_id, message_id, timestamp) do request_id = System.unique_integer([:positive, :monotonic]) seek_command = %Binary.CommandSeek{ consumer_id: consumer_id, request_id: request_id } |> maybe_add_message_id(message_id) |> maybe_add_timestamp(timestamp) Pulsar.Broker.send_request(broker_pid, seek_command) end defp decrement_permits(state, count) do new_permits = max(state.flow_outstanding_permits - count, 0) %{state | flow_outstanding_permits: new_permits} end defp send_initial_flow(broker_pid, consumer_id, permits) do # Initial flow starts from 0 outstanding permits send_flow_command(broker_pid, consumer_id, permits, 0) end defp check_and_refill_permits(%{flow_initial: 0} = state) do state end defp check_and_refill_permits(state) do refill_threshold = state.flow_threshold refill_amount = state.flow_refill current_permits = state.flow_outstanding_permits if current_permits <= refill_threshold do do_refill_permits(state, refill_amount, current_permits) else state end end defp do_refill_permits(state, refill_amount, current_permits) do case send_flow_command(state.broker_pid, state.consumer_id, refill_amount, current_permits) do :ok -> %{state | flow_outstanding_permits: current_permits + refill_amount} error -> Logger.error("Failed to send flow command: #{inspect(error)}") state end end defp send_flow_command(broker_pid, consumer_id, permits, outstanding_permits) do flow_command = %Binary.CommandFlow{ consumer_id: consumer_id, messagePermits: permits } # Metadata for :start event start_metadata = %{ consumer_id: consumer_id, permits_requested: permits, permits_before: outstanding_permits } :telemetry.span( [:pulsar, :consumer, :flow_control], start_metadata, fn -> result = Pulsar.Broker.send_command(broker_pid, flow_command) stop_metadata = %{success: true, permits_after: outstanding_permits + permits} {result, Map.merge(start_metadata, stop_metadata)} end ) end defp maybe_add_message_id(command, nil), do: command defp maybe_add_message_id(command, {ledger_id, entry_id}) do %{command | message_id: %Binary.MessageIdData{ledgerId: ledger_id, entryId: entry_id}} end defp maybe_add_timestamp(command, nil), do: command defp maybe_add_timestamp(command, timestamp) do %{command | message_publish_time: timestamp} end defp schedule_redelivery(nil), do: :ok defp schedule_redelivery(interval) when is_integer(interval) and interval > 0 do Process.send_after(self(), :trigger_redelivery, interval) :ok end defp parse_dead_letter_policy(nil), do: {nil, nil} defp parse_dead_letter_policy(policy) when is_list(policy) do max_redelivery = Keyword.get(policy, :max_redelivery) topic = Keyword.get(policy, :topic) # Validate max_redelivery validated_max_redelivery = case max_redelivery do n when is_integer(n) and n >= 1 -> n _ -> nil end {validated_max_redelivery, topic} end defp should_init_dead_letter_producer?(state) do state.max_redelivery != nil and state.max_redelivery >= 1 end defp start_dead_letter_producer(state) do # Generate default dead letter topic if not provided dead_letter_topic = state.dead_letter_topic || "#{state.topic}-#{state.subscription_name}-DLQ" # Start a producer for the dead letter topic with the same client as the consumer Pulsar.Producer.start_link(dead_letter_topic, client: state.client) end defp send_to_dead_letter(state, payload, _message_id) do if state.dead_letter_producer_pid == nil do {:error, :no_dead_letter_producer} else case Pulsar.Producer.send_message(state.dead_letter_producer_pid, payload) do {:ok, _dlq_message_id} -> :ok {:error, _reason} = error -> error end end end defp maybe_uncompress(%Binary.MessageMetadata{compression: :NONE}, payload) do payload end defp maybe_uncompress(%Binary.MessageMetadata{compression: :ZLIB}, compressed_payload) do :zlib.uncompress(compressed_payload) end defp maybe_uncompress(%Binary.MessageMetadata{compression: :LZ4} = metadata, compressed_payload) do {:ok, payload} = NimbleLZ4.decompress(compressed_payload, metadata.uncompressed_size) payload end defp maybe_uncompress(%Binary.MessageMetadata{compression: :ZSTD}, compressed_payload) do :ezstd.decompress(compressed_payload) end defp maybe_uncompress(%Binary.MessageMetadata{compression: :SNAPPY}, compressed_payload) do {:ok, payload} = :snappyer.decompress(compressed_payload) payload end defp unwrap_messages(metadata, payload) do if metadata.num_messages_in_batch > 0 do parse_batch_messages(payload, metadata.num_messages_in_batch, []) else [{nil, payload}] end end defp parse_batch_messages(<<>>, 0, acc), do: Enum.reverse(acc) defp parse_batch_messages(_, 0, acc), do: Enum.reverse(acc) defp parse_batch_messages( <> = payload, count, acc ) do single_metadata = Binary.SingleMessageMetadata.decode(metadata) payload_size = single_metadata.payload_size <> = data # Build individual message as {metadata, payload} tuple message = {single_metadata, payload} parse_batch_messages(rest, count - 1, [message | acc]) rescue _ -> [{nil, payload}] end defp parse_batch_messages(payload, _, _) do [{nil, payload}] end end