test

Author: Ahmed Ammar

lib/logstash/inputs/dynamodb.rb

@@ -182,7 +182,7 @@ def register
 
   public
   def run(logstash_queue)
-   @something = false
+    $exit = false
     run_with_catch(logstash_queue)
 
     while !stop?
@@ -192,12 +192,10 @@ def run(logstash_queue)
 
   public
   def stop
-    @logger.log("exiting threads");
-    @something = true
+    $exit = true
     exit_threads
-    @logger.log("threads exited");
     until @queue.empty?
-      @logger.log("Flushing rest of events in logstash queue")
+      @logger.info("Flushing rest of events in logstash queue")
       event = @queue.pop()
       queue_event(@parser.parse_stream(event), logstash_queue, @host)
     end # until [email protected]?
@@ -287,32 +285,33 @@ def scan(logstash_queue)
     start_table_copy_thread
 
     scan_queue = @logstash_writer.getQueue()
-    while @something == false
-      event = scan_queue.take()
-      if event.getEntry().nil? and event.getSize() == -1
-        break
-      end # if event.isEmpty()
-      queue_event(@parser.parse_scan(event.getEntry(), event.getSize()), logstash_queue, @host)
+    while !$exit
+      if !@queue.empty?
+        event = scan_queue.take()
+        if event.getEntry().nil? and event.getSize() == -1
+          break
+        end # if event.isEmpty()
+        queue_event(@parser.parse_scan(event.getEntry(), event.getSize()), logstash_queue, @host)
+      else
+        sleep(1)
+      end
     end # while true
   end
 
   private
   def stream(logstash_queue)
-    @logger.log("Starting stream...")
     start_kcl_thread
 
-    @logger.warn("Started KCL thread")
-
-    while @something == false
-    #@logger.warn("Looping KCL thread #{@something}")
-
+    while !$exit
       if !@queue.empty?
         event = @queue.pop()
         queue_event(@parser.parse_stream(event), logstash_queue, @host)
+      else
+        sleep(1)
       end
     end # while true
 
-    @logger.warn("Stopping stream...")
+    @logger.log("Stopping stream...")
   end
 
   private