Untitled

require "amqp"

 

AMQP_CONFIG = {

  host: '127.0.0.1',

  port: 5671,

  username: '<...>',

  password: '<...>',

  ssl: true

}

 

EventMachine.run do

  connection = AMQP.connect(AMQP_CONFIG)

  channel = AMQP::Channel.new(connection)

  channel.queue("org.haggett.amqptest", :exclusive => true, :auto_delete => true) do |queue, declare_ok|

    puts "Declared work queue."

   

    queue.subscribe(:ack => true) do |metadata, payload|

      received_time = payload

      start_time = Time.now.to_f

     

      EM.add_timer(rand(2.0)) do

        end_time = Time.now.to_f

       

        channel.default_exchange.publish([received_time, start_time, end_time].join(":"), :routing_key => metadata.reply_to, :mandatory => true)

        metadata.ack

      end

     

    end

  end

 

  Signal.trap("INT") { connection.close { EventMachine.stop } }

end