Make usage of the new shutdown api + small rspec refactor

lib/logstash/inputs/pipe.rb

@@ -6,6 +6,7 @@
 require "logstash/inputs/base"
 require "logstash/namespace"
 require "socket" # for Socket.gethostname
+require "stud/interval"
 
 # Stream events from a long running command pipe.
 #
@@ -28,7 +29,6 @@ class LogStash::Inputs::Pipe < LogStash::Inputs::Base
 
   def initialize(params)
     super
-    @shutdown_requested = false
     @pipe = nil
   end # def initialize
 
@@ -39,9 +39,9 @@ def register
 
   public
   def run(queue)
-    while !@shutdown_requested
+    while !stop?
       begin
-        @pipe = IO.popen(@command, mode = "r")
+        @pipe = IO.popen(@command, "r")
         hostname = Socket.gethostname
 
         @pipe.each do |line|
@@ -56,25 +56,22 @@ def run(queue)
         end
         @pipe.close
         @pipe = nil
-      rescue LogStash::ShutdownSignal => e
-        break
       rescue Exception => e
         @logger.error("Exception while running command", :e => e, :backtrace => e.backtrace)
       end
 
       # Keep running the command forever.
-      sleep(10)
+      Stud.stoppable_sleep(10) do
+        stop?
+      end
     end
   end # def run
 
-  def teardown
-    @shutdown_requested = true
+  def stop
     if @pipe
       Process.kill("KILL", @pipe.pid) rescue nil
       @pipe.close rescue nil
       @pipe = nil
     end
-    finished
   end
-
 end # class LogStash::Inputs::Pipe

logstash-input-pipe.gemspec

@@ -21,7 +21,7 @@ Gem::Specification.new do |s|
 
   # Gem dependencies
   s.add_runtime_dependency "logstash-core", "~> 2.0.0.snapshot"
-
+  s.add_runtime_dependency 'stud', '~> 0.0.22'
   s.add_runtime_dependency 'logstash-codec-plain'
   s.add_development_dependency 'logstash-devutils'
 end

spec/inputs/pipe_spec.rb

@@ -1,43 +1,68 @@
 # encoding: utf-8
 require "logstash/devutils/rspec/spec_helper"
+require "logstash/inputs/pipe"
 require "tempfile"
 
-describe "inputs/pipe", :unix => true do
+describe LogStash::Inputs::Pipe, :unix => true do
 
-  # rince and repeat a few times to stress the shutdown sequence
-  5.times.each do
-    it "should pipe from echo" do
-      conf = <<-CONFIG
+  it "should register" do
+    input = LogStash::Plugin.lookup("input", "pipe").new("command" => "echo 'world'")
+
+    # register will try to load jars and raise if it cannot find jars or if org.apache.log4j.spi.LoggingEvent class is not present
+    expect {input.register}.to_not raise_error
+  end
+
+  context "when interrupting the plugin" do
+
+    it_behaves_like "an interruptible input plugin" do
+      let(:config) { { "command" => "echo ☹" } }
+    end
+
+    it_behaves_like "an interruptible input plugin" do
+      let(:config) { { "command" => "echo foo" } }
+    end
+
+  end
+
+  describe "pipe from echo" do
+
+    let(:config) do <<-CONFIG
       input {
         pipe {
           command => "echo ☹"
         }
       }
-      CONFIG
+    CONFIG
+    end
 
-      event = input(conf) do |pipeline, queue|
+    let(:event) do
+      input(config) do |pipeline, queue|
         queue.pop
       end
+    end
 
-      insist { event["message"] } == "☹"
+    it "should receive the pipe" do
+      expect(event["message"]).to eq("☹")
     end
+
   end
 
-  # rince and repeat a few times to stress the shutdown sequence
-  5.times.each do
-    it "should pipe from tail -f" do
-      event_count = 10
-      tmp_file = Tempfile.new('logstash-spec-input-pipe')
+  describe "pipe from tail" do
 
-      conf = <<-CONFIG
-      input {
+    let(:tmp_file)    { Tempfile.new('logstash-spec-input-pipe') }
+    let(:event_count) { 10 }
+
+    let(:config) do <<-CONFIG
+    input {
         pipe {
           command => "tail -n +0 -f #{tmp_file.path}"
         }
       }
-      CONFIG
+    CONFIG
+    end
 
-      events = input(conf) do |pipeline, queue|
+    let(:events) do
+      input(config) do |pipeline, queue|
         File.open(tmp_file, "a") do |fd|
           event_count.times do |i|
             # unicode smiley for testing unicode support!
@@ -46,9 +71,11 @@
         end
         event_count.times.map { queue.pop }
       end
+    end
 
+    it "should receive all piped elements" do
       event_count.times do |i|
-        insist { events[i]["message"] } == "#{i} ☹"
+        expect(events[i]["message"]).to eq("#{i} ☹")
       end
     end
   end