Implement plugin stop behavior

lib/logstash/inputs/sqlite.rb

@@ -2,6 +2,8 @@
 require "logstash/inputs/base"
 require "logstash/namespace"
 require "socket"
+require "stud/interval"
+require "concurrent"
 
 
 # Read rows from an sqlite database.
@@ -146,7 +148,7 @@ def run(queue)
 
  begin
  @logger.debug("Tailing sqlite db", :path => @path)
- loop do
+ until stop?
  count = 0
  @table_data.each do |k, table|
  table_name = table[:name]
@@ -174,7 +176,8 @@ def run(queue)
  # sleep a bit
  @logger.debug("No new rows. Sleeping.", :time => sleeptime)
  sleeptime = [sleeptime * 2, sleep_max].min
- sleep(sleeptime)
+
+ Stud.stoppable_sleep(sleeptime) { stop? }
  else
  sleeptime = sleep_min
  end

logstash-input-sqlite.gemspec

@@ -26,5 +26,6 @@ Gem::Specification.new do |s|
  s.add_runtime_dependency 'jdbc-sqlite3'
 
  s.add_development_dependency 'logstash-devutils'
+ s.add_development_dependency 'logstash-codec-plain'
 end
 

spec/inputs/sqlite_spec.rb

@@ -1 +1,11 @@
+# encoding: utf-8
 require "logstash/devutils/rspec/spec_helper"
+require "logstash/inputs/sqlite"
+require "logstash/codecs/plain"
+
+describe LogStash::Inputs::Sqlite do
+ describe "stopping" do
+ let(:config) { {"path" => "testdb.sqlite3"} }
+ it_behaves_like "an interruptible input plugin"
+ end
+end