require 'fluent/plugin/input'
​
module Fluent::Plugin
  class SomeInput < Input
    # First, register the plugin. NAME is the name of this plugin
    # and identifies the plugin in the configuration file.
    Fluent::Plugin.register_input('NAME', self)
​
    # config_param defines a parameter. You can refer a parameter via @port instance variable
    # :default means this parameter is optional
    config_param :port, :integer, default: 8888
​
    # This method is called before starting.
    # 'conf' is a Hash that includes configuration parameters.
    # If the configuration is invalid, raise Fluent::ConfigError.
    def configure(conf)
      super
​
      # configured "port" is referred by `@port` or instance method #port
      if @port < 1024
        raise Fluent::ConfigError, "well known ports cannot be used for this purpose."
      end
​
      # You can also refer to raw parameter via conf[name].
      @port = conf['port']
      ...
    end
​
    # This method is called when starting.
    # Open sockets or files and create a thread here.
    def start
      super
​
      # my own start-up code
    end
​
    # This method is called when shutting down.
    def shutdown
      # my own shutdown code
​
      super
    end
  end
end

tag = "myapp.access"
time = Fluent::Engine.now
record = {"message"=>"body"}
router.emit(tag, time, record)

es = MultiEventStream.new
records.each { |record|
  es.add(time, record)
}
router.emit_stream(tag, es)

router.emit(tag, time, {'foo' => 'bar'})  # OK!
router.emit(tag, time, {:foo => 'bar'})   # NG!

# test/plugin/test_in_your_own.rb
​
require 'test/unit'
require 'fluent/test/driver/input'
​
# your own plugin
require 'fluent/plugin/in_your_own'
​
class YourOwnInputTest < Test::Unit::TestCase
  def setup
    Fluent::Test.setup  # this is required to setup router and others
  end
​
  # default configuration for tests
  CONFIG = %[
    param1 value1
    param2 value2
  ]
​
  def create_driver(conf = CONFIG)
    Fluent::Test::Driver::Input.new(Fluent::Plugin::YourOwnInput).configure(conf)
  end
​
  sub_test_case 'configured with invalid configurations' do
    test 'param1 should reject too short string' do
      assert_raise Fluent::ConfigError do
        create_driver(%[
          param1 a
        ])
      end
    end
​
    test 'param2 is set correctly' do
      d = create_driver
      assert_equal "value2", d.instance.param2
    end
​
    # ...
  end
​
  sub_test_case 'plugin will emit some events' do
    test 'test expects plugin emits events 4 times' do
      d = create_driver
​
      d.run(expect_emits: 4, timeout: 10)
      # this method blocks until input plugin emits events 4 times
      # or 10 seconds passes
​
      events = d.events # array of [tag, time, record]
      assert_equal "expected_tag", events[0][0]
      # ...
    end
  end
​
  # ...
end