test

Author: Ahmed Ammar

lib/logstash/inputs/dynamodb.rb

@@ -182,18 +182,20 @@ def register
 
   public
   def run(logstash_queue)
-    begin
-      run_with_catch(logstash_queue)
+    $exit = false;
+    run_with_catch(logstash_queue)
+    
+    while !stop?
+      Stud.stoppable_sleep(@interval) { stop? }
     end
   end
 
   public
   def stop
-    @logger.warn("STOPPING!")
-
+    $exit = true
     exit_threads
     until @queue.empty?
-      @logger.warn("Flushing rest of events in logstash queue")
+      @logger.log("Flushing rest of events in logstash queue")
       event = @queue.pop()
       queue_event(@parser.parse_stream(event), logstash_queue, @host)
     end # until [email protected]?
@@ -283,7 +285,7 @@ def scan(logstash_queue)
     start_table_copy_thread
 
     scan_queue = @logstash_writer.getQueue()
-    while !stop?
+    while !$exit
       event = scan_queue.take()
       if event.getEntry().nil? and event.getSize() == -1
         break
@@ -294,10 +296,10 @@ def scan(logstash_queue)
 
   private
   def stream(logstash_queue)
-    @logger.warn("Starting stream...")
+    @logger.log("Starting stream...")
     start_kcl_thread
 
-    while !stop?
+    while !$exit
       event = @queue.pop()
       queue_event(@parser.parse_stream(event), logstash_queue, @host)
     end # while true