added purbon's changes, (changed === to == everywhere, added .freeze to constants, removed Jruby check, reorganized private methods)

Author: Brandon Tom

lib/logstash/inputs/DynamoDBLogParser.rb

@@ -58,7 +58,7 @@ def parse_scan(log, new_image_size)
     @key_schema.each { |x|
       @hash_template["dynamodb"]["keys"][x] = data_hash[x]
     }
-    unless @view_type === "keys_only"
+    unless @view_type == "keys_only"
       size_bytes += new_image_size
       @hash_template["dynamodb"]["newImage"] = data_hash
     end
@@ -69,6 +69,11 @@ def parse_scan(log, new_image_size)
     return parse_view_type(@hash_template)
   end
 
+  public
+  def parse_stream(log)
+    return parse_view_type(JSON.parse(@mapper.writeValueAsString(log))["internalObject"])
+  end
+
   private
   def calculate_key_size_in_bytes(record)
     key_size = 0
@@ -88,14 +93,9 @@ def calculate_key_size_in_bytes(record)
     return key_size
   end
 
-  public
-  def parse_stream(log)
-    return parse_view_type(JSON.parse(@mapper.writeValueAsString(log))["internalObject"])
-  end
-
   private
   def parse_view_type(hash)
-    if @log_format === LogStash::Inputs::DynamoDB::LF_PLAIN
+    if @log_format == LogStash::Inputs::DynamoDB::LF_PLAIN
       return hash.to_json
     end
     case @view_type
@@ -110,7 +110,7 @@ def parse_view_type(hash)
 
   private
   def parse_format(hash)
-    if @log_format === LogStash::Inputs::DynamoDB::LF_DYNAMODB
+    if @log_format == LogStash::Inputs::DynamoDB::LF_DYNAMODB
       return hash.to_json
     else
       return dynamodb_to_json(hash)
@@ -128,7 +128,7 @@ def formatAttributeValueMap(hash)
     hash.each do |k, v|
       dynamodb_key = v.keys.first
       dynamodb_value = v.values.first
-      if @log_format === LogStash::Inputs::DynamoDB::LF_JSON_NO_BIN and (dynamodb_key === "BS" or dynamodb_key === "B")
+      if @log_format == LogStash::Inputs::DynamoDB::LF_JSON_NO_BIN and (dynamodb_key == "BS" or dynamodb_key == "B")
         keys_to_delete.push(k) # remove binary values and binary sets
         next
       end

lib/logstash/inputs/dynamodb.rb

@@ -75,14 +75,14 @@ module KCL
 class LogStash::Inputs::DynamoDB < LogStash::Inputs::Base
   config_name "dynamodb"
 
-  LF_DYNAMODB = "dymamodb"
-  LF_JSON_NO_BIN = "json_drop_binary"
-  LF_PLAIN = "plain"
-  LF_JSON_BIN_AS_TEXT = "json_binary_as_text"
-  VT_KEYS_ONLY = "keys_only"
-  VT_OLD_IMAGE = "old_image"
-  VT_NEW_IMAGE = "new_image"
-  VT_ALL_IMAGES = "new_and_old_images"
+  LF_DYNAMODB = "dymamodb".freeze
+  LF_JSON_NO_BIN = "json_drop_binary".freeze
+  LF_PLAIN = "plain".freeze
+  LF_JSON_BIN_AS_TEXT = "json_binary_as_text".freeze
+  VT_KEYS_ONLY = "keys_only".freeze
+  VT_OLD_IMAGE = "old_image".freeze
+  VT_NEW_IMAGE = "new_image".freeze
+  VT_ALL_IMAGES = "new_and_old_images".freeze
 
   default :codec, 'json'
 
@@ -129,7 +129,6 @@ class LogStash::Inputs::DynamoDB < LogStash::Inputs::Base
   # Number of threads to use when scanning the specified table
   config :number_of_scan_threads, :validate => :number, :default => 1
 
-
   # Number of threads to write to the logstash queue when scanning the table
   config :number_of_write_threads, :validate => :number, :default => 1
 
@@ -155,11 +154,6 @@ def build_credentials
 
   public
   def register
-    if not LogStash::Environment.jruby?
-      raise(LogStash::PluginLoadingError, "KCL requires JRuby for streams processing")
-    end # if not LogStash::Environment.jruby?
-
-
     LogStash::Logger.setup_log4j(@logger)
 
     @host = Socket.gethostname
@@ -168,10 +162,10 @@ def register
     @credentials = build_credentials()
     @logger.info("Checkpointer: " + @checkpointer)
 
-    if @perform_scan and @view_type === VT_OLD_IMAGE
+    if @perform_scan and @view_type == VT_OLD_IMAGE
       raise(LogStash::ConfigurationError, "Cannot perform scan with view type: " + @view_type + " configuration")
     end
-    if @view_type === VT_ALL_IMAGES and (not @log_format === LF_PLAIN)
+    if @view_type == VT_ALL_IMAGES and (not @log_format == LF_PLAIN)
       raise(LogStash::ConfigurationError, "Cannot show view_type: " + @view_type + ", with log_format: " + @log_format)
     end
 
@@ -198,6 +192,34 @@ def register
     end # unless @perform_stream
   end # def register
 
+  public
+  def run(logstash_queue)
+    begin
+      run_with_catch(logstash_queue)
+    rescue LogStash::ShutdownSignal
+      exit_threads
+      until @queue.empty?
+        @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]?
+    end # begin
+  end # def run(logstash_queue)
+
+  # Starts KCL app in a background thread
+  # Starts parallel scan if need be in a background thread
+  private
+  def run_with_catch(logstash_queue)
+    if @perform_scan
+      scan(logstash_queue)
+    end # if @perform_scan
+
+    # Once scan is finished, start kcl thread to read from streams
+    if @perform_stream
+      stream(logstash_queue)
+    end # unless @perform_stream
+  end # def run
+
   private
   def setup_stream
     worker_id = SecureRandom.uuid()
@@ -222,21 +244,21 @@ def setup_stream
       stream_status = stream_description.getStreamStatus()
 
       stream_view_type = stream_description.getStreamViewType().to_s.downcase
-      unless (stream_view_type === @view_type or @view_type === VT_KEYS_ONLY or stream_view_type === VT_ALL_IMAGES)
+      unless (stream_view_type == @view_type or @view_type == VT_KEYS_ONLY or stream_view_type == VT_ALL_IMAGES)
         raise(LogStash::ConfigurationError, "Cannot stream " + @view_type + " when stream is setup for " + stream_view_type)
       end
 
-      while stream_status === "ENABLING"
-        if(stream_status === "ENABLING")
+      while stream_status == "ENABLING"
+        if(stream_status == "ENABLING")
           @logger.info("Sleeping until stream is enabled")
           sleep(1)
-        end # if stream_status === "ENABLING"
+        end # if stream_status == "ENABLING"
         stream_description = dynamodb_streams_client.describeStream(AmazonDynamoDB::DescribeStreamRequest.new() \
           .withStreamArn(stream_arn)).getStreamDescription()
         stream_status = stream_description.getStreamStatus()
       end # while not active
 
-      if not stream_status === "ENABLED"
+      if not stream_status == "ENABLED"
         raise(LogStash::PluginLoadingError, "No streams are enabled")
       end # if not active
       @logger.info("Stream Id: " + stream_arn)
@@ -259,34 +281,6 @@ def setup_stream
     @worker = KCL::Worker.new(LogStashRecordProcessorFactory.new(@queue), kcl_config, adapter, @dynamodb_client, cloudwatch_client)
   end # def setup_stream
 
-  public
-  def run(logstash_queue)
-    begin
-      run_with_catch(logstash_queue)
-    rescue LogStash::ShutdownSignal
-      exit_threads
-      until @queue.empty?
-        @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]?
-    end # begin
-  end # def run(logstash_queue)
-
-  # Starts KCL app in a background thread
-  # Starts parallel scan if need be in a background thread
-  private
-  def run_with_catch(logstash_queue)
-    if @perform_scan
-      scan(logstash_queue)
-    end # if @perform_scan
-
-    # Once scan is finished, start kcl thread to read from streams
-    if @perform_stream
-      stream(logstash_queue)
-    end # unless @perform_stream
-  end # def run
-
   private
   def scan(logstash_queue)
     @logger.info("Starting scan...")
@@ -298,7 +292,7 @@ def scan(logstash_queue)
     scan_queue = @logstash_writer.getQueue()
     while true
       event = scan_queue.take()
-      if event.getEntry().nil? and event.getSize() === -1
+      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)