#
# Fluent
#
# Copyright (C) 2011 Kazuki Ohta
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#
module Fluent

class ScribeOutput < BufferedOutput
  Fluent::Plugin.register_output('scribe', self)

  def initialize
    super
    require 'thrift'
    $:.unshift File.join(File.dirname(__FILE__), 'thrift')
    require 'fb303_types'
    require 'fb303_constants'
    require 'facebook_service'
    require 'scribe_types'
    require 'scribe_constants'
    require 'scribe'
  end

  def configure(conf)
    super

    @host = 'localhost'
    @port = 38888
  end

  def start
    super
  end

  def shutdown
    super
  end

  def format(tag, event)
    [tag, event.record].to_msgpack
  end

  def write(chunk)
    records = []
    chunk.open { |io|
      begin
        MessagePack::Unpacker.new(io).each { |record| records << record }
      rescue EOFError
        # EOFError always occured when reached end of chunk.
      end
    }

    socket = Thrift::Socket.new @host, @port.to_i
    transport = Thrift::FramedTransport.new socket
    protocol = Thrift::BinaryProtocol.new transport, false, false
    client = Scribe::Client.new protocol
    transport.open

    entries = []
    records.each { |r|
      entry = LogEntry.new
      entry.category = r[0]
      entry.message = r[1].to_json
      entries << entry
    }
    
    client.Log(entries)
    transport.close
  end
end


end