clone codec per worker

added line codec tests missing spec

Author: colinsurprenant

lib/logstash/inputs/udp.rb

@@ -37,6 +37,8 @@ class LogStash::Inputs::Udp < LogStash::Inputs::Base
  # before packets will start dropping.
  config :queue_size, :validate => :number, :default => 2000
 
+ HOST_FIELD = "host".freeze
+
  public
  def initialize(params)
  super
@@ -86,7 +88,7 @@ def udp_listener(output_queue)
 
  @input_workers = @workers.times do |i|
  @logger.debug("Starting UDP worker thread", :worker => i)
- Thread.new { inputworker(i) }
+ Thread.new(i, @codec.clone) { |i, codec| inputworker(i, codec) }
  end
 
  while !stop?
@@ -109,17 +111,15 @@ def udp_listener(output_queue)
  end
  end # def udp_listener
 
- def inputworker(number)
+ def inputworker(number, codec)
  LogStash::Util::set_thread_name("<udp.#{number}")
  begin
  while true
  payload, client = @input_to_worker.pop
+ host = client[3]
 
- @codec.decode(payload) do |event|
- decorate(event)
- event.set("host", client[3]) if event.get("host").nil?
- @output_queue.push(event)
- end
+ codec.decode(payload) { |event| push_decoded_event(host, event) }
+ codec.flush { |event| push_decoded_event(host, event) }
  end
  rescue => e
  @logger.error("Exception in inputworker", "exception" => e, "backtrace" => e.backtrace)
@@ -136,4 +136,12 @@ def stop
  @udp.close rescue nil
  end
 
+ private
+
+ def push_decoded_event(host, event)
+ decorate(event)
+ event.set(HOST_FIELD, host) if event.get(HOST_FIELD).nil?
+ @output_queue.push(event)
+ end
+
 end # class LogStash::Inputs::Udp

logstash-input-udp.gemspec

@@ -24,6 +24,7 @@ Gem::Specification.new do |s|
 
  s.add_runtime_dependency 'logstash-codec-plain'
  s.add_runtime_dependency 'stud', '~> 0.0.22'
+ s.add_development_dependency 'logstash-codec-line'
  s.add_development_dependency 'logstash-devutils'
 end
 

spec/inputs/udp_spec.rb

@@ -9,8 +9,10 @@
  end
 
  let!(:helper) { UdpHelpers.new }
- let(:port) { rand(1024..65535) }
- subject { LogStash::Plugin.lookup("input", "udp").new({ "port" => port }) }
+ let(:client) { LogStash::Inputs::Test::UDPClient.new(port) }
+ let(:port) { rand(1024..65535) }
+ let(:config) { { "port" => port } }
+ subject { LogStash::Plugin.lookup("input","udp").new(config) }
 
  after :each do
  subject.close rescue nil
@@ -23,8 +25,6 @@
  end
 
  describe "receive" do
-
- let(:client) { LogStash::Inputs::Test::UDPClient.new(port) }
  let(:nevents) { 10 }
 
  let(:events) do
@@ -46,10 +46,32 @@
  expect(message).to match(/msg \d+/)
  end
  end
+ end
+
+ describe "multiple lines per datagram using line codec" do
+ # 3 workers for 3 datagrams send below
+ let(:config) { { "port" => port, "workers" => 3, "codec" => "line" } }
+
+ let(:events) do
+ helper.input(subject, 8) do
+ client.send("line1\nline2")
+ client.send("line3\nline4")
+ client.send("line5\nline6\nline7\nline8")
+ end
+ end
 
+ before(:each) do
+ subject.register
+ end
+
+ it "should receive events been generated" do
+ expect(events.size).to be(8)
+ messages = events.map { |event| event.get("message") }.sort # important to sort here because order is unpredictable
+ messages.each_index {|i| expect(messages[i]).to match("line#{i + 1}")}
+ end
  end
 
  it_behaves_like "an interruptible input plugin" do
- let(:config) { { "port" => port } }
+ # see https://github.com/elastic/logstash-devutils/blob/9c4a1fbf2b0c4547e428c5a40ed84f60aad17f97/lib/logstash/devutils/rspec/shared_examples.rb
  end
-end
+end

spec/support/client.rb

@@ -4,7 +4,6 @@
 module LogStash::Inputs::Test
 
  class UDPClient
-
  attr_reader :host, :port, :socket
 
  def initialize(port)
@@ -14,24 +13,18 @@ def initialize(port)
  socket.connect(host, port)
  end
 
- def send(msg="")
+ def send(msg)
  begin
- send(msg)
- rescue Exception => e
- puts "send.exception", e
+ socket.connect(host, port) if socket.closed?
+ socket.send(msg, 0)
+ rescue => e
+ puts("send exception, retrying", e.inspect)
  retry
  end
  end
 
- def send(msg)
- socket.connect(host, port) if socket.closed?
- socket.send(msg, 0)
- end
-
  def close
- socket.close
+ socket.close unless socket.closed?
  end
-
  end
-
 end