filter_flowcounter_simple

.travis.yml

@@ -1,6 +1,8 @@
 rvm:
  - 1.9.3
  - 2.0.0
- - 2.1.*
+ - 2.1
+ - 2.2
 gemfile:
  - Gemfile
+ - Gemfile.v0.10

Gemfile.v0.10

@@ -0,0 +1,4 @@
+source "https://rubygems.org"
+
+gemspec
+gem 'fluentd', '~> 0.10.0'

examples/filter.conf

@@ -0,0 +1,13 @@
+<source>
+ type dummy
+ tag dummy
+</source>
+
+<filter dummy>
+ type flowcounter_simple
+ unit second
+</filter>
+
+<match dummy>
+ type null
+</match>

examples/output.conf

@@ -0,0 +1,9 @@
+<source>
+ type dummy
+ tag dummy
+</source>
+
+<match dummy>
+ type flowcounter_simple
+ unit second
+</match>

fluent-plugin-flowcounter-simple.gemspec

@@ -18,4 +18,6 @@ Gem::Specification.new do |gem|
  gem.add_development_dependency "rake"
  gem.add_development_dependency "pry"
  gem.add_development_dependency "pry-nav"
+ gem.add_development_dependency "test-unit"
+ gem.add_development_dependency "test-unit-rr"
 end

lib/fluent/plugin/filter_flowcounter_simple.rb

@@ -0,0 +1,21 @@
+# I'm lazy to to implement filter plugin. Use output plugin instance
+require_relative 'out_flowcounter_simple'
+require 'forwardable'
+
+class Fluent::FlowCounterSimpleFilter < Fluent::Filter
+ Fluent::Plugin.register_filter('flowcounter_simple', self)
+
+ extend Forwardable
+ attr_reader :output
+ def_delegators :@output, :configure, :start, :shutdown, :flush_emit
+
+ def initialize
+ super
+ @output = Fluent::FlowCounterSimpleOutput.new
+ end
+
+ def filter_stream(tag, es)
+ @output.emit(tag, es, Fluent::NullOutputChain.instance)
+ es
+ end
+end if defined?(Fluent::Filter)

test/helper.rb

@@ -13,6 +13,7 @@
 $LOAD_PATH.unshift(File.dirname(__FILE__))
 require 'fluent/test'
 require 'fluent/plugin/out_flowcounter_simple'
+require 'fluent/plugin/filter_flowcounter_simple'
 
 class Test::Unit::TestCase
 end

test/plugin/test_filter_flowcounter_simple.rb

@@ -0,0 +1,58 @@
+require_relative '../helper'
+require "test/unit/rr"
+
+class FlowCounterSimpleFilterTest < Test::Unit::TestCase
+ include Fluent
+
+ def setup
+ Fluent::Test.setup
+ @time = Fluent::Engine.now
+ end
+
+ CONFIG = %[
+ unit second
+ ]
+
+ def create_driver(conf = CONFIG)
+ Fluent::Test::FilterTestDriver.new(Fluent::FlowCounterSimpleFilter).configure(conf, true)
+ end
+
+ def test_filter
+ msgs = []
+ 10.times do
+ msgs << {'message'=> 'a' * 100}
+ msgs << {'message'=> 'b' * 100}
+ end
+ d = create_driver
+ filtered, log = filter(d, msgs)
+ assert_equal msgs, filtered
+ assert( log.include?("count:20"), log )
+ end
+
+ private
+
+ def filter(d, msgs)
+ stub(d.instance.output).start
+ stub(d.instance.output).shutdown
+ d.run {
+ msgs.each {|msg|
+ d.filter(msg, @time)
+ }
+ }
+ log = capture_log(d.instance.output.log) do
+ d.instance.flush_emit(0)
+ end
+ filtered = d.filtered_as_array
+ filtered_msgs = filtered.map {|m| m[2] }
+ [filtered_msgs, log]
+ end
+
+ def capture_log(log)
+ tmp = log.out
+ log.out = StringIO.new
+ yield
+ return log.out.string
+ ensure
+ log.out = tmp
+ end
+end if defined?(Fluent::Test::FilterTestDriver)

test/plugin/test_out_flowcounter_simple.rb

@@ -1,4 +1,4 @@
-require 'helper'
+require_relative '../helper'
 
 class FlowCounterSimpleOutputTest < Test::Unit::TestCase
  def setup
@@ -67,7 +67,7 @@ def test_comment
  assert( out.include?("comment:foobar"), out )
  end
 
-private
+ private
 
  def capture_log(log)
  tmp = log.out