انتشار پیام کارآمد و قابل اعتماد با RabbitMQ و Elixir

{:basic_ack, seqno, multiple}, state
{:basic_nack, seqno, multiple}, state

defmodule MyPublisher do
  use GenServer

  ...

  def start_link(channel) do
    # Start the process
    GenServer.start_link(__MODULE__, channel, name: __MODULE__)
  end

  def init(channel) do
    # Initialize the process state and do setup

    # Enable publisher confirms
    AMQP.Confirm.select(channel)

    # Register the current process as the async handler
    # for acks and nacks
    AMQP.Confirm.register_handler(channel, self())

    ...

    # Put the channel in the state. Proper
    # connection and channel management is
    # not covered in this post.
    {:ok, %{channel: channel}}
  end

  ...
end

  def publish(topic, payload) do
    # This function runs in the calling process
    GenServer.call(__MODULE__, {:publish, topic, payload})
  end

  def handle_call({:publish, topic, payload}, from, state) do
    # This function runs in the GenServer
    ...

    # Perform the actual publishing (ensure it is persistent)
    AMQP.Basic.publish(state.channel, "pub_sub", topic, payload, persistent: true)

    # Note that we are NOT replying here
    {:noreply, state}
  end

  def handle_info({:basic_ack, seqno, multiple}, state) do
    # The broker is saying that the publish with `seqno`
    # has been received. If `multiple` is true, all messages
    # up to `seqno` have been received.
    confirm_messages(seqno, multiple, fn from ->
      # For each of the received messages,
      # we can reply with `:ok`.
      GenServer.reply(from, :ok)
    end)

    {:noreply, state}
  end

  def handle_info({:basic_nack, seqno, multiple}, state) do
    # The broker is saying that the publish with `seqno`
    # has been lost. If `multiple` is true, all messages
    # up to `seqno` have been lost.
    confirm_messages(seqno, multiple, fn from ->
      # For each of the lost messages,
      # we can reply with `{:error, :nack}`.
      GenServer.reply(from, {:error, :nack})
    end)

    {:noreply, state}
  end

def init(channel) do
    ...

    # Create an ETS table in which the elements are ordered. It
    # will be owned by the GenServer process. We make it private
    # because there is no need for other processes to read or
    # write to it.
    :ets.new(@table_name, [:ordered_set, :private, :named_table])

    {:ok, %{channel: channel}}
  end

  ...

  def handle_call({:publish, topic, payload}, from, state) do
    # Before we publish, get a hold of the next publish
    # sequence number `seqno`.
    seqno = AMQP.Confirm.next_publish_seqno(state.channel)

    # Store the caller reference (`from`) against
    # the `seqno` in ETS
    :ets.insert(@table_name, {seqno, from})

    # Perform the actual publishing
    ...
  end

  defp confirm_messages(seqno, _multiple = false, reply_fun) do
    # Lookup and remove the entries for seqno
    case :ets.take(@table_name, seqno) do
      [{^seqno, from}] -> reply_fun.(from)
      _ -> :ok
    end
  end

  defp confirm_messages(seqno, true = _multiple, reply_fun) do
    :ets.select(@table_name, [
      {
        # bind `seqno` and `from` to $1 and $2
        {:"$1", :"$2"},
        # match entries where the bound seqno is =< to the incoming seqno 
        [{:"=<", :"$1", seqno}],
        # and return all the matching entries as tuples of $1 and $2
        [{{:"$1", :"$2"}}]}
    ])
    |> Enum.each(fn {key, from} ->
      reply_fun.(from)
      :ets.delete(@table_name, key)
    end)
  end

defmodule MyPublisher do
  use GenServer

  @table_name :pending_confirms

  def start_link(channel) do
    GenServer.start_link(__MODULE__, channel, name: __MODULE__)
  end

  def init(channel) do
    AMQP.Confirm.select(channel)
    AMQP.Confirm.register_handler(channel, self())

    :ets.new(@table_name, [:ordered_set, :private, :named_table])

    {:ok, %{channel: channel}}
  end

  def publish(topic, payload) do
    GenServer.call(__MODULE__, {:publish, topic, payload})
  end

  def handle_call({:publish, topic, payload}, from, state) do
    seqno = AMQP.Confirm.next_publish_seqno(state.channel)

    # Store the caller reference in ETS
    :ets.insert(@table_name, {seqno, from})

    # Perform the actual publishing
    AMQP.Basic.publish(state.channel, "pub_sub", topic, payload, persistent: true)

    {:noreply, state}
  end

  def handle_info({:basic_ack, seqno, multiple}, state) do
    confirm_messages(seqno, multiple, fn from ->
      GenServer.reply(from, :ok)
    end)

    {:noreply, state}
  end

  def handle_info({:basic_nack, seqno, multiple}, state) do
    confirm_messages(seqno, multiple, fn from ->
      GenServer.reply(from, {:error, :nack})
    end)

    {:noreply, state}
  end

  defp confirm_messages(seqno, true = _multiple, reply_fun) do
    :ets.select(@table_name, [
      {{:"$1", :"$2"}, [{:"=<", :"$1", seqno}], [{{:"$1", :"$2"}}]}
    ])
    |> Enum.each(fn {key, from} ->
      reply_fun.(from)
      :ets.delete(@table_name, key)
    end)
  end

  defp confirm_messages(seqno, false, reply_fun) do
    case :ets.take(@table_name, seqno) do
      [{^seqno, from}] -> reply_fun.(from)
      _ -> :ok
    end
  end
end