peppy

peppy

Help: How to use AMQP channel from pool for publishing messages?

I set up a small process that maintains an AMQP connection/channel. This script automatically reconnects to RabbitMQ if the connection fails. I built this based on an old, depreciated and incomplete guide. I’m still a newbie, so I don’t fully understand how to use this. I know the script is connecting and reconnecting (as far as I know), but how do I actually use/fetch the channel?

For example, if I have a function in a different module, such as chatroom.ex, how do I fetch the existing channel in the pool so that I can publish messages to a queue?

def publish_function(chatmessage) do
   # Receive a chat room message here and publish it to "test_queue" in RabbitMQ. How do I use the existing connections?
   #something like:
  channel = ???? #channel in the consumer module?
  AMQP.Basic.publish(*channel*, "", "test_queue", chatmessage)

end

Here is what I have so far:

connectionmanager.ex:

defmodule ExchatWeb.AMQPConnectionManager do

  use GenServer
  use AMQP

  def start_link(_opts) do
    GenServer.start_link(__MODULE__, :ok, [name: __MODULE__])
  end

  def init(:ok) do
    children = [
      ExchatWeb.Publish
    ]

    Supervisor.start_link(children, strategy: :one_for_one, name: ExchatWeb.PublishSupervisor)
    establish_new_connection()
  end

  defp establish_new_connection do
    case AMQP.Connection.open do
      {:ok, conn} ->
        Process.link conn.pid
        {:ok, {conn, %{}}}
      {:error, reason} ->
        IO.puts "failed for #{inspect reason}"
        :timer.sleep 5000
        establish_new_connection()
    end
  end

  def request_channel(consumer) do
    GenServer.cast(__MODULE__, {:chan_request, consumer})
  end

  def handle_cast({:chan_request, consumer}, {conn, channel_mappings}) do
    new_mapping = store_channel_mapping(conn, consumer, channel_mappings)
    channel = Map.get(new_mapping, consumer)
    consumer.channel_available(channel)
    {:noreply, {conn, new_mapping}}
  end

  defp store_channel_mapping(conn, consumer, channel_mappings) do
    Map.put_new_lazy(channel_mappings, consumer, fn() -> create_channel(conn) end)
    IO.inspect(channel_mappings)
  end

  defp create_channel(conn) do
    {:ok, chan} = Channel.open(conn)
    chan
  end

end

publishconsumer.ex:

defmodule ExchatWeb.PublishConsumer do
  use GenServer

  def start_link(_opts) do
    GenServer.start_link(__MODULE__, :ok, [name: __MODULE__])
  end

  def init(_opts) do
    ExchatWeb.AMQPConnectionManager.request_channel(__MODULE__)
    {:ok, nil}
  end

  def channel_available(chan) do
    GenServer.cast(__MODULE__, {:channel_available, chan})
  end

  def handle_cast({:channel_available, chan}, _state) do
    IO.inspect(chan)
    #bind_to_queue chan
    {:noreply, chan}
  end

end

Most Liked

beepbeepbopbop

beepbeepbopbop

If I am reading this correctly, you want to checkout a channel from your connection pool and publish to said channel. Based off the code you’ve posted, I assume the following:

  • Your connection pool abstraction is using a process (although I would use something like Poolboy)
  • You need a method akin to ConnectionManager.checkout(...) which results in getting a channel that you can use.

In this instance, user land code may look like this:

def publish_msg(msg) do
  channel = ConnectionManager.checkout()
  AMQP.Basic.publish(channel, "", "test_queue", msg)
  # Check the connection back in.
end

This means that while ConnectionManager is a process, you need to change it to use a synchronous API, using the call() variants. You would also need to implement your check-in logic, unless you use a callback API where it automatically cleans up for you.

def callback_example(msg) do
  ConnectionManager.with_channel(fn channel ->
    AMQP.Basic.publish(channel, "", "test_queue", msg)
    # Check-ins are implicit.
  end)
end

Your handle_cast simply will not work as it is now. Since your user land code assumes that a channel is something that you return, you can’t use a handle_cast, unless you utilise GenServer.reply, which isn’t necessary in your use case.

Where Next?

Popular in Questions Top

lastday4you
I wanted to check elixir version in phoenix because i found that my elixir is 1.5 but when i use Enum.chunk_by it said the function is un...
New
mcarvalho
What is the difference between System.get_env and Application.get_env? For example, what are best practices to use one versus another.
New
mgjohns61585
Could someone help me? I'm making my first elixir program, number guessing game. I can't figure out how to convert the user's guess from ...
New
makeitrein
Hey all, just started picking up Elixir last week and am writing a scraper as a learning project. Baby step #1 is extracting the number ...
New
aadeshere1
I have a another noob question about loop. Since elixir is immutable, while loop is not directly possible. total = 10 while total != 0 ...
New
vonH
When I run the Plug and I recompile I wind up having to use Ctrl C to quit iex and start again. Witht the help of rlwrap I can use the cu...
New
fireproofsocks
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? https://hexdocs.pm/ecto/Ecto.Repo.h...
New
ovidiubadita
Hey all, I discovered Elixir and I love it. I always wanted to learn a functional programming and I intended to go for Haskell, but afte...
New
wernerlaude
In AR this is so simple @articles = current_user.articles How to do in Ecto? def index(conn, _params) do current_user = conn.assig...
New
idi527
I’ve been re-reading swift book again and noticed that multiline strings there don’t have a trailing line break, unlike in elixir iex(2)...
New

Other popular topics Top

JDanielMartinez
Hi! May someone helps me, please! I have two apps into an umbrella project: the first one is Database, which manages queries, and the se...
New
dotdotdotPaul
Okay, I'm having a heck of a time trying to figure out how to best handle the validation of belongs_to associations in Ecto. I'm sure I'...
New
lessless
I believe there are people here who are dealing with CSV files import on the daily basis, and since Excel is a really popular tool there ...
New
itssasanka
Hi all, Trying to get some more clarity over utc_datetime and naive_datetime for Ecto: https://hexdocs.pm/ecto/Ecto.Schema.html#module-...
New
grych
Hi folks, Few months ago I have announced the proof-of-concept of the library to manipulate the browsers DOM objects directly from Elixi...
639 49522 488
New
chrismccord
This release brings a number of exciting features, including integration with the new Phoenix LiveDashboard and Phoenix LiveView. There h...
New
vertexbuffer
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
baxterw3b
Hi guys, i’m new in the Elixir world, and i have to say, that i love it! i’m having some problem to understand anonymous functions with ...
New
siddhant3030
Hi, I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New
joeerl
Hello again - after a longish gap I’ve decided I really must dig into Elixir and see what’s been happening here - so I have a few questio...
New

We're in Beta

About us Mission Statement