From f5e9c5323f7733dd3ad62c6e88cdbe987142cf7a Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 26 Jun 2014 17:08:44 +0900
Subject: [PATCH 1/6] Remove obsoleted TcpInput

StreamInput based TcpInput is obsoleted since 2+ years ago.
So we can remove this class for newer plugin
---
 lib/fluent/plugin/in_stream.rb | 28 ----------------------------
 test/plugin/test_in_stream.rb  | 24 ------------------------
 2 files changed, 52 deletions(-)

diff --git a/lib/fluent/plugin/in_stream.rb b/lib/fluent/plugin/in_stream.rb
index d741eebf05..63d21375da 100644
--- a/lib/fluent/plugin/in_stream.rb
+++ b/lib/fluent/plugin/in_stream.rb
@@ -160,34 +160,6 @@ def on_close
     end
   end
 
-
-  # obsolete
-  # ForwardInput is backward compatible with TcpInput
-  #class TcpInput < StreamInput
-  #  Plugin.register_input('tcp', self)
-  #
-  #  config_param :port, :integer, :default => DEFAULT_LISTEN_PORT
-  #  config_param :bind, :string, :default => '0.0.0.0'
-  #
-  #  def configure(conf)
-  #    super
-  #  end
-  #
-  #  def listen
-  #    log.debug "listening fluent socket on #{@bind}:#{@port}"
-  #    Coolio::TCPServer.new(@bind, @port, Handler, method(:on_message))
-  #  end
-  #end
-  class TcpInput < ForwardInput
-    Plugin.register_input('tcp', self)
-
-    def initialize
-      super
-      $log.warn "'tcp' input is obsoleted and will be removed soon. Use 'forward' instead."
-    end
-  end
-
-
   class UnixInput < StreamInput
     Plugin.register_input('unix', self)
 
diff --git a/test/plugin/test_in_stream.rb b/test/plugin/test_in_stream.rb
index b30356bbe2..ce1bba0464 100644
--- a/test/plugin/test_in_stream.rb
+++ b/test/plugin/test_in_stream.rb
@@ -105,30 +105,6 @@ def send_data(data)
   end
 end
 
-class TcpInputTest < Test::Unit::TestCase
-  include StreamInputTest
-
-  PORT = unused_port
-  CONFIG = %[
-    port #{PORT}
-    bind 127.0.0.1
-  ]
-
-  def create_driver(conf=CONFIG)
-    super(Fluent::TcpInput, conf)
-  end
-
-  def test_configure
-    d = create_driver
-    assert_equal PORT, d.instance.port
-    assert_equal '127.0.0.1', d.instance.bind
-  end
-
-  def connect
-    TCPSocket.new('127.0.0.1', PORT)
-  end
-end
-
 class UnixInputTest < Test::Unit::TestCase
   include StreamInputTest
 

From ae8d900d35e420fa25e6bd0c1ee6052980be9cb2 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 26 Jun 2014 18:38:12 +0900
Subject: [PATCH 2/6] Add udp input plugin

---
 lib/fluent/plugin/in_udp.rb      |  17 +++++
 lib/fluent/plugin/socket_util.rb | 116 +++++++++++++++++++++++++++++++
 test/plugin/test_in_udp.rb       |  97 ++++++++++++++++++++++++++
 3 files changed, 230 insertions(+)
 create mode 100644 lib/fluent/plugin/in_udp.rb
 create mode 100755 test/plugin/test_in_udp.rb

diff --git a/lib/fluent/plugin/in_udp.rb b/lib/fluent/plugin/in_udp.rb
new file mode 100644
index 0000000000..31228ac4b4
--- /dev/null
+++ b/lib/fluent/plugin/in_udp.rb
@@ -0,0 +1,17 @@
+require 'fluent/plugin/socket_util'
+
+module Fluent
+  class UdpInput < SocketUtil::BaseInput
+    Plugin.register_input('udp', self)
+
+    config_set_default :port, 5160
+    config_param :body_size_limit, :size, :default => 4096
+
+    def listen(callback)
+      log.debug "listening udp socket on #{@bind}:#{@port}"
+      @usock = SocketUtil.create_udp_socket(@bind)
+      @usock.bind(@bind, @port)
+      SocketUtil::UdpHandler.new(@usock, log, @body_size_limit, callback)
+    end
+  end
+end
diff --git a/lib/fluent/plugin/socket_util.rb b/lib/fluent/plugin/socket_util.rb
index 9c689b6805..2667f3b95e 100644
--- a/lib/fluent/plugin/socket_util.rb
+++ b/lib/fluent/plugin/socket_util.rb
@@ -1,3 +1,5 @@
+require 'cool.io'
+
 module Fluent
   module SocketUtil
     def create_udp_socket(host)
@@ -10,5 +12,119 @@ def create_udp_socket(host)
       end
     end
     module_function :create_udp_socket
+
+    class UdpHandler < Coolio::IO
+      def initialize(io, log, body_size_limit, callback)
+        super(io)
+        @io = io
+        @log = log
+        @body_size_limit = body_size_limit
+        @callback = callback
+      end
+
+      def on_readable
+        msg, addr = @io.recvfrom_nonblock(@body_size_limit)
+        @callback.call(msg, addr)
+      rescue => e
+        @log.error "unexpected error", :error => e, :error_class => e.class
+      end
+    end
+
+    class TcpHandler < Coolio::Socket
+      def initialize(io, log, delimiter, callback)
+        super(io)
+        if io.is_a?(TCPSocket)
+          @addr = (io.peeraddr rescue PEERADDR_FAILED)
+
+          opt = [1, @timeout.to_i].pack('I!I!')  # { int l_onoff; int l_linger; }
+          io.setsockopt(Socket::SOL_SOCKET, Socket::SO_LINGER, opt)
+        end
+        @delimiter = delimiter
+        @callback = callback
+        @log = log
+        @log.trace { "accepted fluent socket object_id=#{self.object_id}" }
+        @buffer = "".force_encoding('ASCII-8BIT')
+      end
+
+      def on_connect
+      end
+
+      def on_read(data)
+        @buffer << data
+        pos = 0
+
+        while i = @buffer.index(@delimiter, pos)
+          msg = @buffer[pos..i]
+          @callback.call(msg, @addr)
+          pos = i + 1
+        end
+        @buffer.slice!(0, pos) if pos > 0
+      rescue => e
+        @log.error "unexpected error", :error => e, :error_class => e.class
+        close
+      end
+
+      def on_close
+        @log.trace { "closed fluent socket object_id=#{self.object_id}" }
+      end
+    end
+
+    class BaseInput < Fluent::Input
+      def initialize
+        super
+        require 'fluent/parser'
+      end
+
+      config_param :tag, :string
+      config_param :format, :string
+      config_param :port, :integer, :default => 5150
+      config_param :bind, :string, :default => '0.0.0.0'
+      config_param :source_host_key, :string, :default => nil
+
+      def configure(conf)
+        super
+
+        @parser = TextParser.new
+        @parser.configure(conf)
+      end
+
+      def start
+        @loop = Coolio::Loop.new
+        @handler = listen(method(:on_message))
+        @loop.attach(@handler)
+        @thread = Thread.new(&method(:run))
+      end
+
+      def shutdown
+        @loop.watchers.each { |w| w.detach }
+        @loop.stop
+        @handler.close
+        @thread.join
+      end
+
+      def run
+        @loop.run
+      rescue => e
+        log.error "unexpected error", :error => e, :error_class => e.class
+        log.error_backtrace
+      end
+
+      private
+
+      def on_message(msg, addr)
+        @parser.parse(msg) { |time, record|
+          unless time && record
+            log.warn "pattern not match: #{msg.inspect}"
+            return
+          end
+
+          record[@source_host_key] = addr[3] if @source_host_key
+          Engine.emit(@tag, time, record)
+        }
+      rescue => e
+        log.error msg.dump, :error => e, :error_class => e.class, :host => addr[3]
+        log.error_backtrace
+      end
+    end
   end
 end
diff --git a/test/plugin/test_in_udp.rb b/test/plugin/test_in_udp.rb
new file mode 100755
index 0000000000..df91597e54
--- /dev/null
+++ b/test/plugin/test_in_udp.rb
@@ -0,0 +1,97 @@
+require 'fluent/test'
+require 'helper'
+
+class UdpInputTest < Test::Unit::TestCase
+  def setup
+    Fluent::Test.setup
+  end
+
+  PORT = unused_port
+  CONFIG = %!
+    port #{PORT}
+    bind 127.0.0.1
+    tag udp
+    format /^\\[(?<time>[^\\]]*)\\] (?<message>.*)/
+  !
+
+  IPv6_CONFIG = %!
+    port #{PORT}
+    bind ::1
+    tag udp
+    format /^\\[(?<time>[^\\]]*)\\] (?<message>.*)/
+  !
+
+  def create_driver(conf = CONFIG)
+    Fluent::Test::InputTestDriver.new(Fluent::UdpInput).configure(conf)
+  end
+
+  def test_configure
+    configs = {'127.0.0.1' => CONFIG}
+    configs.merge!('::1' => IPv6_CONFIG) if ipv6_enabled?
+
+    configs.each_pair { |k, v|
+      d = create_driver(v)
+      assert_equal PORT, d.instance.port
+      assert_equal k, d.instance.bind
+      assert_equal 4096, d.instance.body_size_limit
+    }
+  end
+
+  def test_time_format
+    configs = {'127.0.0.1' => CONFIG}
+    configs.merge!('::1' => IPv6_CONFIG) if ipv6_enabled?
+
+    configs.each_pair { |k, v|
+      d = create_driver(v)
+
+      tests = [
+        {'msg' => '[Sep 11 00:00:00] localhost logger: foo', 'expected' => Time.strptime('Sep 11 00:00:00', '%b %d %H:%M:%S').to_i},
+        {'msg' => '[Sep  1 00:00:00] localhost logger: foo', 'expected' => Time.strptime('Sep  1 00:00:00', '%b  %d %H:%M:%S').to_i},
+      ]
+
+      d.run do
+        u = Fluent::SocketUtil.create_udp_socket(k)
+        u.connect(k, PORT)
+        tests.each {|test|
+          u.send(test['msg'], 0)
+        }
+        sleep 1
+      end
+
+      emits = d.emits
+      emits.each_index {|i|
+        assert_equal(tests[i]['expected'], emits[i][1])
+      }
+    }
+  end
+
+  def test_msg_size
+    d = create_driver
+    tests = create_test_case
+
+    d.run do
+      u = UDPSocket.new
+      u.connect('127.0.0.1', PORT)
+      tests.each { |test|
+        u.send(test['msg'], 0)
+      }
+      sleep 1
+    end
+
+    compare_test_result(d.emits, tests)
+  end
+
+  def create_test_case
+    [
+      {'msg' => '[Sep 10 00:00:00] localhost: ' + 'x' * 100 + "\n", 'expected' => 'localhost: ' + 'x' * 100},
+      {'msg' => '[Sep 10 00:00:00] localhost: ' + 'x' * 1024 + "\n", 'expected' => 'localhost: ' + 'x' * 1024},
+    ]
+  end
+
+  def compare_test_result(emits, tests)
+    assert_equal(2, emits.size)
+    emits.each_index {|i|
+      assert_equal(tests[i]['expected'], emits[i][2]['message'])
+    }
+  end
+end

From c2a304f84069f26fc9a5d28e00b8bec4d0cf6aaa Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 26 Jun 2014 18:38:34 +0900
Subject: [PATCH 3/6] Add tcp input plugin

---
 lib/fluent/plugin/in_tcp.rb | 15 +++++++
 test/plugin/test_in_tcp.rb  | 85 +++++++++++++++++++++++++++++++++++++
 2 files changed, 100 insertions(+)
 create mode 100644 lib/fluent/plugin/in_tcp.rb
 create mode 100755 test/plugin/test_in_tcp.rb

diff --git a/lib/fluent/plugin/in_tcp.rb b/lib/fluent/plugin/in_tcp.rb
new file mode 100644
index 0000000000..04335a43a6
--- /dev/null
+++ b/lib/fluent/plugin/in_tcp.rb
@@ -0,0 +1,15 @@
+require 'fluent/plugin/socket_util'
+
+module Fluent
+  class TcpInput < SocketUtil::BaseInput
+    Plugin.register_input('tcp', self)
+
+    config_set_default :port, 5170
+    config_param :delimiter, :string, :default => "\n" # syslog family add "\n" to each message and this seems only way to split messages in tcp stream
+
+    def listen(callback)
+      log.debug "listening tcp socket on #{@bind}:#{@port}"
+      Coolio::TCPServer.new(@bind, @port, SocketUtil::TcpHandler, log, @delimiter, callback)
+    end
+  end
+end
diff --git a/test/plugin/test_in_tcp.rb b/test/plugin/test_in_tcp.rb
new file mode 100755
index 0000000000..7953eceed2
--- /dev/null
+++ b/test/plugin/test_in_tcp.rb
@@ -0,0 +1,85 @@
+require 'fluent/test'
+require 'helper'
+
+class TcpInputTest < Test::Unit::TestCase
+  def setup
+    Fluent::Test.setup
+  end
+
+  PORT = unused_port
+  CONFIG = %[
+    port #{PORT}
+    bind 127.0.0.1
+    tag tcp
+    format json
+  ]
+
+  IPv6_CONFIG = %[
+    port #{PORT}
+    bind ::1
+    tag tcp
+    format json
+  ]
+
+  def create_driver(conf = CONFIG)
+    Fluent::Test::InputTestDriver.new(Fluent::TcpInput).configure(conf)
+  end
+
+  def test_configure
+    configs = {'127.0.0.1' => CONFIG}
+    configs.merge!('::1' => IPv6_CONFIG) if ipv6_enabled?
+
+    configs.each_pair { |k, v|
+      d = create_driver(v)
+      assert_equal PORT, d.instance.port
+      assert_equal k, d.instance.bind
+      assert_equal "\n", d.instance.delimiter
+    }
+  end
+
+  def test_msg_size
+    d = create_driver
+    tests = create_test_case
+
+    d.run do
+      tests.each {|test|
+        TCPSocket.open('127.0.0.1', PORT) do |s|
+          s.send(test['msg'], 0)
+        end
+      }
+      sleep 1
+    end
+
+    compare_test_result(d.emits, tests)
+  end
+
+  def test_msg_size_with_same_connection
+    d = create_driver
+    tests = create_test_case
+
+    d.run do
+      TCPSocket.open('127.0.0.1', PORT) do |s|
+        tests.each {|test|
+          s.send(test['msg'], 0)
+        }
+      end
+      sleep 1
+    end
+
+    compare_test_result(d.emits, tests)
+  end
+
+  def create_test_case
+    [
+      {'msg' => '{"k":"v","message":"tcptest1"}' + "\n", 'expected' => 'tcptest1'},
+      {'msg' => '{"k":"v","message":"tcptest2"}' + "\n", 'expected' => 'tcptest2'},
+    ]
+  end
+
+  def compare_test_result(emits, tests)
+    assert_equal(2, emits.size)
+    emits.each_index {|i|
+      assert_equal(tests[i]['expected'], emits[i][2]['message'])
+    }
+  end
+end

From 8a068541d3739c4e75176249d73c1d3b314de2e0 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 26 Jun 2014 18:59:32 +0900
Subject: [PATCH 4/6] in_syslog uses socket utility handlers instead of own
 handlers

---
 lib/fluent/plugin/in_syslog.rb | 64 +++-------------------------------
 1 file changed, 5 insertions(+), 59 deletions(-)

diff --git a/lib/fluent/plugin/in_syslog.rb b/lib/fluent/plugin/in_syslog.rb
index 747068e489..ead2072b5a 100644
--- a/lib/fluent/plugin/in_syslog.rb
+++ b/lib/fluent/plugin/in_syslog.rb
@@ -122,7 +122,7 @@ def run
     end
 
     protected
-    def receive_data_parser(data)
+    def receive_data_parser(data, addr)
       m = SYSLOG_REGEXP.match(data)
       unless m
         log.warn "invalid syslog message: #{data.dump}"
@@ -144,7 +144,7 @@ def receive_data_parser(data)
       log.error_backtrace
     end
 
-    def receive_data(data)
+    def receive_data(data, addr)
       m = SYSLOG_ALL_REGEXP.match(data)
       unless m
         log.warn "invalid syslog message", :data=>data
@@ -183,9 +183,10 @@ def listen(callback)
       if @protocol_type == :udp
         @usock = SocketUtil.create_udp_socket(@bind)
         @usock.bind(@bind, @port)
-        UdpHandler.new(@usock, callback)
+        SocketUtil::UdpHandler.new(@usock, log, 2048, callback)
       else
-        Coolio::TCPServer.new(@bind, @port, TcpHandler, log, callback)
+        # syslog family add "\n" to each message and this seems only way to split messages in tcp stream
+        Coolio::TCPServer.new(@bind, @port, SocketUtil::TcpHandler, log, "\n", callback)
       end
     end
 
@@ -199,60 +200,5 @@ def emit(pri, time, record)
     rescue => e
       log.error "syslog failed to emit", :error => e.to_s, :error_class => e.class.to_s, :tag => tag, :record => Yajl.dump(record)
     end
-
-    class UdpHandler < Coolio::IO
-      def initialize(io, callback)
-        super(io)
-        @io = io
-        @callback = callback
-      end
-
-      def on_readable
-        msg, addr = @io.recvfrom_nonblock(2048)
-        #host = addr[3]
-        #port = addr[1]
-        #@callback.call(host, port, msg)
-        @callback.call(msg)
-      rescue
-        # TODO log?
-      end
-    end
-
-    class TcpHandler < Coolio::Socket
-      def initialize(io, log, on_message)
-        super(io)
-        if io.is_a?(TCPSocket)
-          opt = [1, @timeout.to_i].pack('I!I!')  # { int l_onoff; int l_linger; }
-          io.setsockopt(Socket::SOL_SOCKET, Socket::SO_LINGER, opt)
-        end
-        @on_message = on_message
-        @log = log
-        @log.trace { "accepted fluent socket object_id=#{self.object_id}" }
-        @buffer = "".force_encoding('ASCII-8BIT')
-      end
-
-      def on_connect
-      end
-
-      def on_read(data)
-        @buffer << data
-        pos = 0
-
-        # syslog family add "\n" to each message and this seems only way to split messages in tcp stream
-        while i = @buffer.index("\n", pos)
-          msg = @buffer[pos..i]
-          @on_message.call(msg)
-          pos = i + 1
-        end
-        @buffer.slice!(0, pos) if pos > 0
-      rescue => e
-        @log.error "syslog error", :error => e, :error_class => e.class
-        close
-      end
-
-      def on_close
-        @log.trace { "closed fluent socket object_id=#{self.object_id}" }
-      end
-    end
   end
 end

From 3b042a53f8f77d7c44fe9ef512c8c16bad7991c6 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Mon, 30 Jun 2014 09:11:03 +0900
Subject: [PATCH 5/6] Remove trailing delimiter on both udp and tcp

---
 lib/fluent/plugin/socket_util.rb | 7 +++++--
 test/plugin/test_in_tcp.rb       | 8 ++++----
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/lib/fluent/plugin/socket_util.rb b/lib/fluent/plugin/socket_util.rb
index 2667f3b95e..3de850d11d 100644
--- a/lib/fluent/plugin/socket_util.rb
+++ b/lib/fluent/plugin/socket_util.rb
@@ -24,6 +24,7 @@ def initialize(io, log, body_size_limit, callback)
 
       def on_readable
         msg, addr = @io.recvfrom_nonblock(@body_size_limit)
+        msg.chomp!
         @callback.call(msg, addr)
       rescue => e
         @log.error "unexpected error", :error => e, :error_class => e.class
@@ -31,6 +32,8 @@ def on_readable
     end
 
     class TcpHandler < Coolio::Socket
+      PEERADDR_FAILED = ["?", "?", "name resolusion failed", "?"]
+
       def initialize(io, log, delimiter, callback)
         super(io)
         if io.is_a?(TCPSocket)
@@ -54,9 +57,9 @@ def on_read(data)
         pos = 0
 
         while i = @buffer.index(@delimiter, pos)
-          msg = @buffer[pos..i]
+          msg = @buffer[pos...i]
           @callback.call(msg, @addr)
-          pos = i + 1
+          pos = i + @delimiter.length
         end
         @buffer.slice!(0, pos) if pos > 0
       rescue => e
diff --git a/test/plugin/test_in_tcp.rb b/test/plugin/test_in_tcp.rb
index 7953eceed2..6e94e5135b 100755
--- a/test/plugin/test_in_tcp.rb
+++ b/test/plugin/test_in_tcp.rb
@@ -11,14 +11,14 @@ def setup
     port #{PORT}
     bind 127.0.0.1
     tag tcp
-    format json
+    format none
   ]
 
   IPv6_CONFIG = %[
     port #{PORT}
     bind ::1
     tag tcp
-    format json
+    format none
   ]
 
   def create_driver(conf = CONFIG)
@@ -71,8 +71,8 @@ def test_msg_size_with_same_connection
 
   def create_test_case
     [
-      {'msg' => '{"k":"v","message":"tcptest1"}' + "\n", 'expected' => 'tcptest1'},
-      {'msg' => '{"k":"v","message":"tcptest2"}' + "\n", 'expected' => 'tcptest2'},
+      {'msg' => "tcptest1\n", 'expected' => 'tcptest1'},
+      {'msg' => "tcptest2\n", 'expected' => 'tcptest2'},
     ]
   end
 

From 0d10768a6aa427f93ff73cbbee264b956dd7d13d Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Mon, 30 Jun 2014 17:34:07 +0900
Subject: [PATCH 6/6] Add other format tests to udp and tcp inputs

---
 test/plugin/test_in_tcp.rb | 79 ++++++++++++++++++++------------------
 test/plugin/test_in_udp.rb | 59 +++++++++++++++-------------
 2 files changed, 74 insertions(+), 64 deletions(-)

diff --git a/test/plugin/test_in_tcp.rb b/test/plugin/test_in_tcp.rb
index 6e94e5135b..326a191e99 100755
--- a/test/plugin/test_in_tcp.rb
+++ b/test/plugin/test_in_tcp.rb
@@ -7,21 +7,20 @@ def setup
   end
 
   PORT = unused_port
-  CONFIG = %[
+  BASE_CONFIG = %[
     port #{PORT}
-    bind 127.0.0.1
     tag tcp
+  ]
+  CONFIG = BASE_CONFIG + %[
+    bind 127.0.0.1
     format none
   ]
-
-  IPv6_CONFIG = %[
-    port #{PORT}
+  IPv6_CONFIG = BASE_CONFIG + %[
     bind ::1
-    tag tcp
     format none
   ]
 
-  def create_driver(conf = CONFIG)
+  def create_driver(conf)
     Fluent::Test::InputTestDriver.new(Fluent::TcpInput).configure(conf)
   end
 
@@ -37,44 +36,48 @@ def test_configure
     }
   end
 
-  def test_msg_size
-    d = create_driver
-    tests = create_test_case
-
-    d.run do
-      tests.each {|test|
-        TCPSocket.open('127.0.0.1', PORT) do |s|
-          s.send(test['msg'], 0)
-        end
-      }
-      sleep 1
-    end
-
-    compare_test_result(d.emits, tests)
-  end
-
-  def test_msg_size_with_same_connection
-    d = create_driver
-    tests = create_test_case
+  {
+    'none' => [
+      {'msg' => "tcptest1\n", 'expected' => 'tcptest1'},
+      {'msg' => "tcptest2\n", 'expected' => 'tcptest2'},
+    ],
+    'json' => [
+      {'msg' => {'k' => 123, 'message' => 'tcptest1'}.to_json + "\n", 'expected' => 'tcptest1'},
+      {'msg' => {'k' => 'tcptest2', 'message' => 456}.to_json + "\n", 'expected' => 456},
+    ]
+  }.each { |format, test_cases|
+    define_method("test_msg_size_#{format}") do
+      d = create_driver(BASE_CONFIG + "format #{format}")
+      tests = test_cases
 
-    d.run do
-      TCPSocket.open('127.0.0.1', PORT) do |s|
+      d.run do
         tests.each {|test|
-          s.send(test['msg'], 0)
+          TCPSocket.open('127.0.0.1', PORT) do |s|
+            s.send(test['msg'], 0)
+          end
         }
+        sleep 1
       end
-      sleep 1
+
+      compare_test_result(d.emits, tests)
     end
 
-    compare_test_result(d.emits, tests)
-  end
+    define_method("test_msg_size_with_same_connection_#{format}") do
+      d = create_driver(BASE_CONFIG + "format #{format}")
+      tests = test_cases
 
-  def create_test_case
-    [
-      {'msg' => "tcptest1\n", 'expected' => 'tcptest1'},
-      {'msg' => "tcptest2\n", 'expected' => 'tcptest2'},
-    ]
-  end
+      d.run do
+        TCPSocket.open('127.0.0.1', PORT) do |s|
+          tests.each {|test|
+            s.send(test['msg'], 0)
+          }
+        end
+        sleep 1
+      end
+
+      compare_test_result(d.emits, tests)
+    end
+  }
 
   def compare_test_result(emits, tests)
     assert_equal(2, emits.size)
diff --git a/test/plugin/test_in_udp.rb b/test/plugin/test_in_udp.rb
index df91597e54..11f17f18be 100755
--- a/test/plugin/test_in_udp.rb
+++ b/test/plugin/test_in_udp.rb
@@ -7,21 +7,20 @@ def setup
   end
 
   PORT = unused_port
-  CONFIG = %!
+  BASE_CONFIG = %[
     port #{PORT}
-    bind 127.0.0.1
     tag udp
+  ]
+  CONFIG = BASE_CONFIG + %!
+    bind 127.0.0.1
     format /^\\[(?<time>[^\\]]*)\\] (?<message>.*)/
   !
-
-  IPv6_CONFIG = %!
-    port #{PORT}
+  IPv6_CONFIG = BASE_CONFIG + %!
     bind ::1
-    tag udp
     format /^\\[(?<time>[^\\]]*)\\] (?<message>.*)/
   !
 
-  def create_driver(conf = CONFIG)
+  def create_driver(conf)
     Fluent::Test::InputTestDriver.new(Fluent::UdpInput).configure(conf)
   end
 
@@ -65,28 +64,36 @@ def test_time_format
     }
   end
 
-  def test_msg_size
-    d = create_driver
-    tests = create_test_case
-
-    d.run do
-      u = UDPSocket.new
-      u.connect('127.0.0.1', PORT)
-      tests.each { |test|
-        u.send(test['msg'], 0)
-      }
-      sleep 1
-    end
-
-    compare_test_result(d.emits, tests)
-  end
-
-  def create_test_case
-    [
+  {
+    'none' => [
+      {'msg' => "tcptest1\n", 'expected' => 'tcptest1'},
+      {'msg' => "tcptest2\n", 'expected' => 'tcptest2'},
+    ],
+    'json' => [
+      {'msg' => {'k' => 123, 'message' => 'tcptest1'}.to_json + "\n", 'expected' => 'tcptest1'},
+      {'msg' => {'k' => 'tcptest2', 'message' => 456}.to_json + "\n", 'expected' => 456},
+    ],
+    '/^\\[(?<time>[^\\]]*)\\] (?<message>.*)/' => [
       {'msg' => '[Sep 10 00:00:00] localhost: ' + 'x' * 100 + "\n", 'expected' => 'localhost: ' + 'x' * 100},
       {'msg' => '[Sep 10 00:00:00] localhost: ' + 'x' * 1024 + "\n", 'expected' => 'localhost: ' + 'x' * 1024},
     ]
-  end
+  }.each { |format, test_cases|
+    define_method("test_msg_size_#{format[0] == '/' ? 'regexp' : format}") do
+      d = create_driver(BASE_CONFIG + "format #{format}")
+      tests = test_cases
+
+      d.run do
+        u = UDPSocket.new
+        u.connect('127.0.0.1', PORT)
+        tests.each { |test|
+          u.send(test['msg'], 0)
+        }
+        sleep 1
+      end
+
+      compare_test_result(d.emits, tests)
+    end
+  }
 
   def compare_test_result(emits, tests)
     assert_equal(2, emits.size)