Packages

off_broadway_pulsar

1.2.8
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]) do

    # Notify producer that consumer is ready and needs initial flow

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

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

    {:ok, %{broadway_producer: broadway_producer}}

  end



  @impl true

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

    # Send the Pulsar.Message struct directly to the Broadway producer

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



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

    # Broadway will handle ACK/NACK through the acknowledger

    {:noreply, state}

  end

end