Merge buffered and non buffered outputs

example/in_forward.conf

@@ -8,4 +8,7 @@
 
 <match test>
   @type stdout
+  # <buffer>
+  #   flush_interval 10s
+  # </buffer>
 </match>

example/out_buffered_null.conf

@@ -13,16 +13,20 @@
 </source>
 
 <match dummy.**>
-  @type buffered_null
-  try_flush_interval 60
-  flush_interval 60
-  buffer_chunk_limit 1k
-  buffer_queue_limit 2
+  @type null
+  <buffer>
+    flush_interval 60s
+    chunk_limit_size 1k
+    total_limit_size 4k
+  </buffer>
 </match>
 
 <label error_log>
   <match **>
-    @type stdout # or buffered_stdout
+    @type stdout
+    # <buffer>
+    #   flush_interval 1s
+    # </buffer>
   </match>
 </label>
 

lib/fluent/plugin/out_null.rb

@@ -18,10 +18,48 @@
 
 module Fluent::Plugin
   class NullOutput < Output
+    # This plugin is for tests of non-buffered/buffered plugins
     Fluent::Plugin.register_output('null', self)
+    Fluent::Plugin.register_output('buffered_null', self)
+
+    config_section :buffer do
+      config_set_default :chunk_keys, ['tag']
+      config_set_default :flush_at_shutdown, true
+      config_set_default :chunk_limit_size, 10 * 1024
+    end
+
+    def prefer_buffered_processing
+      false
+    end
+
+    def prefer_delayed_commit
+      @delayed
+    end
+
+    attr_accessor :feed_proc, :delayed
+
+    def initialize
+      super
+      @delayed = false
+      @feed_proc = nil
+    end
 
     def process(tag, es)
       # Do nothing
     end
+
+    def write(chunk)
+      if @feed_proc
+        @feed_proc.call(chunk)
+      end
+    end
+
+    def try_write(chunk)
+      if @feed_proc
+        @feed_proc.call(chunk)
+      end
+      # not to commit chunks for testing
+      # commit_write(chunk.unique_id)
+    end
   end
 end

lib/fluent/plugin/out_stdout.rb

@@ -19,30 +19,67 @@
 module Fluent::Plugin
   class StdoutOutput < Output
     Fluent::Plugin.register_output('stdout', self)
+    Fluent::Plugin.register_output('buffered_stdout', self)
 
     helpers :inject, :formatter, :compat_parameters
 
     DEFAULT_FORMAT_TYPE = 'json'
 
+    config_section :buffer do
+      config_set_default :chunk_keys, ['tag']
+      config_set_default :flush_at_shutdown, true
+      config_set_default :chunk_limit_size, 10 * 1024
+    end
+
     config_section :format do
       config_set_default :@type, DEFAULT_FORMAT_TYPE
     end
 
+    def prefer_buffered_processing
+      false
+    end
+
+    def prefer_delayed_commit
+      @delayed
+    end
+
+    attr_accessor :delayed
+
+    def initialize
+      super
+      @delayed = false
+    end
+
     def configure(conf)
       if conf['output_type'] && !conf['format']
         conf['format'] = conf['output_type']
       end
       compat_parameters_convert(conf, :inject, :formatter)
+
       super
+
       @formatter = formatter_create(conf: conf.elements('format').first, default_type: DEFAULT_FORMAT_TYPE)
     end
 
     def process(tag, es)
       es.each {|time,record|
-        r = inject_values_to_record(tag, time, record)
-        $log.write "#{Time.at(time).localtime} #{tag}: #{@formatter.format(tag, time, r).chomp}\n"
+        $log.write(format(tag, time, record))
       }
       $log.flush
     end
+
+    def format(tag, time, record)
+      record = inject_values_to_record(tag, time, record)
+      "#{Time.at(time).localtime} #{tag}: #{@formatter.format(tag, time, record).chomp}\n"
+    end
+
+    def write(chunk)
+      chunk.write_to($log)
+    end
+
+    def try_write(chunk)
+      chunk.write_to($log)
+      commit_write(chunk.unique_id)
+    end
   end
 end

test/plugin/test_in_monitor_agent.rb

@@ -119,6 +119,8 @@ def test_configure
           "@id"=>"null",
           "@type" => "null"
         },
+        "buffer_queue_length" => 0,
+        "buffer_total_queued_size" => 0,
         "output_plugin"   => true,
         "plugin_category" => "output",
         "plugin_id"       => "null",
@@ -285,6 +287,8 @@ def get(uri, header = {})
           "@id"   => "null",
           "@type" => "null"
         },
+        "buffer_queue_length" => 0,
+        "buffer_total_queued_size" => 0,
         "output_plugin"   => true,
         "plugin_category" => "output",
         "plugin_id"       => "null",