title: RPC using EventMachine & RabbitMQ
body: <p>I've been starting to play around with RabbitMQ <a href="http://rubyamqp.info/articles/patterns_and_use_cases/#code_example" rel="nofollow">RPC sample code</a> provided in AMQP gem doc, trying to write very simple code performing <em>synchronous</em> remote calls:</p>
  
  <pre><code>require "amqp"
  
  module RPC
    class Base
      include EM::Deferrable
  
      def rabbit(rabbit_callback)
        rabbit_loop = Proc.new {
          AMQP.connect do |connection|
            AMQP::Channel.new(connection) do |channel|
              channel.queue("rpc.queue", :exclusive =&gt; false, :durable =&gt; true) do |requests_queue|
                self.callback(&amp;rabbit_callback)
                self.succeed(connection, channel, requests_queue)
              end # requests_queue
            end # AMQP.channel
          end # AMQP.connect
  
          Signal.trap("INT")  { connection.close { EM.stop } }
          Signal.trap("TERM") { connection.close { EM.stop } }
        }
  
        if !EM.reactor_running?
          EM.run do
            rabbit_loop.call
          end
        else
          rabbit_loop.call
        end
      end
    end
  
    class Server &lt; Base
  
      def run
        server_loop = Proc.new do |connection, channel, requests_queue|
          consumer = AMQP::Consumer.new(channel, requests_queue).consume
          consumer.on_delivery do |metadata, payload|
            puts "[requests] Got a request #{metadata.message_id}. Sending a reply to #{metadata.reply_to}..."
            channel.default_exchange.publish(Time.now.to_s,
                                             :routing_key    =&gt; metadata.reply_to,
                                             :correlation_id =&gt; metadata.message_id,
                                             :mandatory      =&gt; true)
            metadata.ack
          end
        end
        rabbit(server_loop)
      end
  
    end
  
    class Client &lt; Base
  
      def sync_push(request)
        result = nil
        sync_request = Proc.new do |connection, channel, requests_queue|
          message_id = Kernel.rand(10101010).to_s
  
          response_queue = channel.queue("", :exclusive =&gt; true, :auto_delete =&gt; true)
          response_queue.subscribe do |headers, payload|
            if headers.correlation_id == message_id
              result = payload
              connection.close { EM.stop }
            end
          end
  
          EM.add_timer(0.1) do 
            puts "[request] Sending a request...#{request} with id #{message_id}"
            channel.default_exchange.publish(request,
                                             :routing_key =&gt; requests_queue.name,
                                             :reply_to    =&gt; response_queue.name,
                                             :message_id  =&gt; message_id)
          end
        end
  
        rabbit(sync_request)
        result
      end
    end
  end
  </code></pre>
  
  <p>The idea is pretty simple: I want to have a message queue always ready (this is handled by <code>rabbit</code> method). Whenever the client wants to send a request, it starts by creating a temporary queue for the response along with a message id; it then publishes the request to the main message queue and waits for a response with the same message id in the temporary queue in order to know when the answer for this specific request is ready. I guess that the <code>message_id</code> is somehow redundant with the temporary queue (as the queue should also be unique).</p>
  
  <p>I now run dummy script using this client/server code</p>
  
  <pre><code># server session
  &gt;&gt; server = RPC::Server.new
  =&gt; #&lt;RPC::Server:0x007faaa23bb5b0&gt;
  &gt;&gt; server.run
  Updating client properties
  [requests] Got a request 3315740. Sending a reply to amq.gen-QCv8nP2dI5Qd6bg2Q1Xhk0...
  </code></pre>
  
  <p>and </p>
  
  <pre><code># client session
  &gt;&gt; client = RPC::Client.new
  =&gt; #&lt;RPC::Client:0x007ffb6be6aed8&gt;
  &gt;&gt; client.sync_push "test 1"
  Updating client properties
  [request] Sending a request...test 1 with id 3315740
  =&gt; "2012-11-02 21:58:45 +0100"
  &gt;&gt; client.sync_push "test 2"
  AMQ::Client::ConnectionClosedError: Trying to send frame through a closed connection. Frame is #&lt;AMQ::Protocol::MethodFrame:0x007ffb6b9c83d0 @payload="\x002\x00\n\x00\x00\x00\f\x00\x00\x00\x00", @channel=1&gt;
  </code></pre>
  
  <p>There are two points I really don't understand:</p>
  
  <ol>
  <li>related to EventMachine: in the <code>Client</code> code, why do I have to call <code>EM.add_timer</code> if I want my message to actually be published? And why using <code>EM.next_tick</code> doesn't work? My understanding is that "everything" is supposed to be "ready" when publish is called here.</li>
  <li>related to AMQP: why does my client crash due a closed connection for the second request? A brand new EM/AMQP loop is supposed to be created each time a new request is pushed.</li>
  </ol>
  
  <p>There sadly is very few code available online dealing with EM/AMQP so any help will be deeply appreciated!
  Any comment regarding the efficiency of this would also be much appreciated.</p>