Packages

off_broadway_pulsar

1.3.3
1.5.1 1.5.0 1.4.0 1.3.11 1.3.10 1.3.9 1.3.8 1.3.7 1.3.6 1.3.5 1.3.4 1.3.3 1.3.2 1.3.1 1.3.0 1.2.11 1.2.10 1.2.9 1.2.8 1.2.7 1.2.6 1.2.5 1.2.4 1.2.3 1.2.2 1.2.1 1.2.0 1.1.0 1.0.4

A Broadway producer for Apache Pulsar.

HexDocs

Current section

Files

Jump to
Readme
29 Versions
2 Dependencies
Dependants
Files
Activity
off_broadway_pulsar lib off_broadway pulsar consumer.ex
Raw

lib/off_broadway/pulsar/consumer.ex 

defmodule OffBroadway.Pulsar.Consumer do

  @moduledoc false

  @behaviour Pulsar.Consumer.Callback



  @impl true

  def init([broadway_producer, topic]) do

    # Notify producer that consumer is ready and needs initial flow.

    # topic is included so the producer can use it as a stable map key,

    # avoiding stale PID accumulation when this consumer restarts.

    # This is sent on first startup AND on every consumer restart.

    send(broadway_producer, {:consumer_ready, self(), topic})

    {:ok, %{broadway_producer: broadway_producer, topic: topic}}

  end



  @impl true

  def handle_message(%Pulsar.Message{} = message, state) do

    # self() is the consumer PID used for ACK/NACK routing.

    # state.topic is the stable key used for flow-permit accounting.

    send(state.broadway_producer, {:pulsar_message, message, self(), state.topic})



    # Return {:noreply, state} to use manual ACK mode

    # Broadway will handle ACK/NACK through the acknowledger

    {:noreply, state}

  end

end