defmodule Arland.RabbitMQ.Gramnad do alias :crypto, as: Crypto alias :ets, as: ETS require Logger use GenServer use AMQP @gramnad_main_queue "gramnad_queue" def start_link(init_args) do GenServer.start_link(__MODULE__, init_args, name: __MODULE__) end def init(args) do Logger.info("Starting RabbitMQ connection client") conn_link = "amqp://#{args.rabbitmq_username}:#{args.rabbitmq_password}@#{args.rabbitmq_url}:#{args.rabbitmq_port}" {:ok, conn} = Connection.open(conn_link) {:ok, chan} = Channel.open(conn) {:ok, queue} = setup_queue(chan) :ok = Basic.qos(chan, prefetch_count: 5) {:ok, _consumer_tag} = Basic.consume(chan, queue, self(), no_ack: false) ets_ref = ETS.new(:gramnad_rpc_pids, [:set]) {:ok, %{chan: chan, queue: queue, ets_ref: ets_ref}} end def handle_info({:basic_consume_ok, _data}, state) do Logger.info("Consumer process registered") {:noreply, state} end def handle_info({:basic_cancel, _data}, state) do Logger.warning("Consumer process cancel") {:stop, :normal, state} end def handle_info({:basic_cancel_ok, _data}, state) do Logger.warning("Consumer process cancel ok") {:noreply, state} end def handle_info({:basic_deliver, payload, attributes}, state) do pid_from_corr_id = gen_pid_from_corr_id( state.ets_ref, attributes.correlation_id ) GenServer.reply(pid_from_corr_id, payload) unregister_corr_id(state.ets_ref, attributes.correlation_id) Basic.ack(state.chan, attributes.delivery_tag) {:noreply, state} end def handle_call({:rpc, data}, from, state) do correlation_id = generate_correlation_id() Basic.publish( state.chan, "", @gramnad_main_queue, data |> Jason.encode!(), correlation_id: correlation_id, reply_to: state.queue ) register_corr_id(state.ets_ref, correlation_id, from) {:noreply, state} end def handle_call(msg, _from, state) do Logger.warn("Got unknown message: #{inspect(msg)}") {:reply, {:error, "Method undefined"}, state} end def handle_cast(msg, state) do Logger.warn("Got unknown message: #{inspect(msg)}") {:noreply, state} end defp register_corr_id(ets_ref, correlation_id, pid) do ETS.insert(ets_ref, {correlation_id, pid}) end defp unregister_corr_id(ets_ref, correlation_id) do ETS.delete(ets_ref, correlation_id) end defp gen_pid_from_corr_id(ets_ref, correlation_id) do [{_, pid}] = ETS.lookup(ets_ref, correlation_id) pid end defp setup_queue(chan, queue_name \\ "") do {:ok, responce} = Queue.declare( chan, queue_name, durable: false, auto_delete: true ) {:ok, responce.queue} end defp generate_correlation_id() do Crypto.strong_rand_bytes(16) |> Base.encode16() end end