fluent amqp plugin

in_amqp.rb

module Fluent
  class AmqpInput < Input
    Fluent::Plugin.register_input('amqp', self)

    def initialize
      super
      require 'amqp'
    end

    def configure(conf)
      super

      raise ConfigError, "'tag' parameter is required"   unless @tag = conf['tag']
      raise ConfigError, "'queue' parameter is required" unless @queue = conf['queue']

      @host = conf.has_key?('host') ? conf['host'] : 'localhost'
      @port = conf.has_key?('port') ? conf['port'] : 5672

      @auto_delete = conf.has_key?('auto_delete') ? conf['auto_delete'] : false 
    end

    def start
      super
      EventMachine.run do
        @connection = AMQP.connect(:host => @host)

        channel = AMQP::Channel.new(@connection)
        queue = channel.queue(@queue, :auto_delete => @auto_delete)
        exchange = channel.direct("")
        queue.subscribe do |record|
          time = Time.now.to_i
          Engine.emit(@tag, Fluent::Event.new(time, record))
        end                                                                                         
      end
    end

    def shutdown
      super
      @connection.close { EventMachive.stop }
    end
  end
end