Fluent output plugin for MessagePack RPC

out_msgpack_rpc.rb

module Fluent
  class MessagePackRPCOutput < Fluent::BufferedOutput
    Fluent::Plugin.register_output('msgpack_rpc', self)

    config_param :host    , :string  , :default => 'localhost'
    config_param :port    , :integer
    config_param :method  , :string
    config_param :unit    , :string  , :default => 'chunk'


    def initialize
      super
      require 'msgpack/rpc'
      @available_unit = {'chunk'=>:chunk,'record'=>:record}
    end

    def configure(conf)
      super
    end

    def start
      super
      @client = initialize_client
      if @available_unit.has_key?(@unit) then
        @unit = @available_unit[@unit]
      else
        raise
      end
    end

    def shutdown
      super
      if [email protected]? then
        @client.close
      end
    end

    def format(tag, time, record)
      record.to_msgpack
    end

    def write(chunk)
      case @unit
      when :chunk
        @client.call(@method, chunk.read)
      when :record
        #TODO
      end
    end

    private
    def initialize_client
      begin
        cli = MessagePack::RPC::Client.new(@host, @port)
        #connection test
        res = cli.call(@method, MessagePack.pack('test'))
      rescue
        shutdown
        raise
      end
      cli
    end
  end
end