Use PackedForward instead of Message

See fluent/fluentd#671

Author: okkez

lib/fluent/logger/fluent_logger.rb

@@ -20,6 +20,8 @@
 require 'monitor'
 require 'logger'
 require 'json'
+require 'base64'
+require 'securerandom'
 
 module Fluent
   module Logger
@@ -98,14 +100,16 @@ def last_error
       def post_with_time(tag, map, time)
         @logger.debug { "event: #{tag} #{map.to_json}" rescue nil } if @logger.debug?
         tag = "#{@tag_prefix}.#{tag}" if @tag_prefix
-        write [tag, time.to_i, map]
+        write(tag, time.to_i, map)
       end
 
       def close
         @mon.synchronize {
           if @pending
             begin
-              send_data(@pending)
+              @pending.each do |tag, record|
+                send_data([tag, record].to_msgpack)
+              end
             rescue => e
               set_last_error(e)
               @logger.error("FluentLogger: Can't send logs to #{@host}:#{@port}: #{$!}")
@@ -140,36 +144,40 @@ def suppress_sec
         end
       end
 
-      def write(msg)
+      def write(tag, time, map)
         begin
-          data = to_msgpack(msg)
+          record = to_msgpack([time, map])
         rescue => e
           set_last_error(e)
+          msg = [tag, time, map]
           @logger.error("FluentLogger: Can't convert to msgpack: #{msg.inspect}: #{$!}")
           return false
         end
 
         @mon.synchronize {
           if @pending
-            @pending << data
+            @pending[tag] << record
           else
-            @pending = data
+            @pending = Hash.new{|h, k| h[k] = "" }
+            @pending[tag] = record
           end
 
           # suppress reconnection burst
-          if !@connect_error_history.empty? && @pending.bytesize <= @limit
+          if !@connect_error_history.empty? && @pending.to_s.bytesize <= @limit
             if Time.now.to_i - @connect_error_history.last < suppress_sec
               return false
             end
           end
 
           begin
-            send_data(@pending)
+            @pending.each do |tag, record|
+              send_data([tag, record].to_msgpack)
+            end
             @pending = nil
             true
           rescue => e
             set_last_error(e)
-            if @pending.bytesize > @limit
+            if @pending.to_s.bytesize > @limit
               @logger.error("FluentLogger: Can't send logs to #{@host}:#{@port}: #{$!}")
               call_buffer_overflow_handler(@pending)
               @pending = nil
@@ -199,6 +207,7 @@ def send_data(data)
         #  end
         #  data = data[n..-1]
         #end
+
         true
       end
 

spec/fluent_logger_spec.rb

@@ -203,7 +203,7 @@ def wait_transfer
         logger_io.rewind
         log = logger_io.read
         expect(log).to match /Failed to connect/
-        expect(log).to match /Can't send logs to/
+        expect(log).to match /Can\'t send logs to/
       }
 
       it ('post limit over') do
@@ -213,11 +213,11 @@ def wait_transfer
         expect(queue.last).to be_nil
 
         logger_io.rewind
-        expect(logger_io.read).not_to match /Can't send logs to/
+        expect(logger_io.read).not_to match /Can\'t send logs to/
 
         logger.post('tag', {'a' => ('c' * 1000)})
         logger_io.rewind
-        expect(logger_io.read).to match /Can't send logs to/
+        expect(logger_io.read).to match /Can\'t send logs to/
       end
 
       it ('log connect error once') do
@@ -240,8 +240,11 @@ class BufferOverflowHandler
 
         def flush(messages)
           @buffer ||= []
-          MessagePack::Unpacker.new.feed_each(messages) do |msg|
-            @buffer << msg
+          messages.each do |tag, message|
+            unpacker = MessagePack::Unpacker.new(StringIO.new(message))
+            unpacker.each do |time, record|
+              @buffer << [tag, time, record]
+            end
           end
         end
       end