consumer.rb

consumer.rb

        desc "start_consumer", "start the test consumer"
        def start_consumer
          puts "Running #{AMQP::VERSION} version of the gem"

          AMQP.start(:host => "localhost", :user => "guest", :password => "guest", :vhost => "/", 
                            :logging => true, :port => 5672) do |connection|
                              
            channel = AMQP::Channel.new(connection)
            
            requests_queue = channel.queue("one", :exclusive => true, :auto_delete => true)
            
            Signal.trap("INT") do
              connection.close do
                EM.stop{exit}
              end
            end
            
            channel.prefetch(1)
            
            requests_queue.subscribe(:ack => true) do |header, body|
              puts "received in server #{body.inspect}"
              
              (0..5).each do |n|
                header.ack
                
                reply = {:reply => "Respone #{n}", :is_last => (n == 5)}
                
                AMQP::Exchange.default.publish(
                                                MultiJson.encode(reply), 
                                                :routing_key => header.reply_to,
                                                :correlation_id => header.correlation_id
                                              )

                                                
                sleep(2)
              end
            end
            
            puts " [x] Awaiting RPC requests"
          end          
        end