Port to fluentd-1.0

lib/fluent/plugin/out_wendelin.rb

@@ -19,14 +19,14 @@
 # This plugin hooks into Fluentd in a way similiar to out_forward and
 # out_secure_forward and outputs event stream to a Wendelin HTTP endpoint.
 
-require 'fluent/output'
+require 'fluent/plugin/output'
 require_relative 'wendelin_client'
 
 
-module Fluent
+module Fluent::Plugin
 
-  class WendelinOutput < ObjectBufferedOutput   # XXX verify base class
-    Plugin.register_output('wendelin', self)
+  class WendelinOutput < Output
+    Fluent::Plugin.register_output('wendelin', self)
 
     # where Wendelin's Input Stream Tool is located,
     # ex http://example.com/erp5/portal_ingestion_policies/example_ingestion
@@ -45,9 +45,17 @@ module Fluent
     config_param    :read_timeout,       :integer,  :default => 60
     config_param    :keep_alive_timeout, :integer,  :default => 300
 	
+	config_section :buffer do
+	  config_set_default :chunk_keys, ["tag"]
+	end
+	
     def configure(conf)
         super
 		
+		unless @chunk_key_tag
+		  raise Fluent::ConfigError, "buffer chunk key must include 'tag' for wendelin output"
+		end
+
         credentials = {}
         if @user
             credentials['user']     = @user
@@ -68,12 +76,11 @@ module Fluent
         # TODO
     end
 
-
-    # hooked to ObjectBufferedOutput - write out records from a buffer chunk for a tag.
+    # Use normal "Synchronous Buffer" - write out records from a buffer chunk for a tag.
     #
-    # NOTE this is called from a separate thread (see OutputThread and BufferedOutput)
-    def write_objects(tag, chunk)
-        # NOTE if this fail and raises -> it will unroll to BufferedOutput#try_flush
+    def write(chunk)
+	    return if chunk.empty?
+        # NOTE if this fail and raises -> it will unroll to Output#try_flush
         # which detects errors and retries outputting logs up to retry maxcount
         # times and aborts outputting current logs if all try fail.
         #
@@ -86,7 +93,7 @@ module Fluent
 
         # for input_stream_ref use tag as-is - it will be processed/translated
         # further on server by Wendelin
-        reference = tag
+        reference = chunk.metadata.tag
 
         if @use_keep_alive
           @wendelin.ingest_with_keep_alive(reference, data_chunk)