Extracted template method for Lookup class used in filters (#59)

In class LogStash::Filters::Jdbc::Lookup the methods fetch(local, event), call_prepared(local, event) shared a lot of logic.
Extracted common part in template method name 'retrieve_local_data(local, event, load_method_ref)' and introduced two methods containing the customization logic:
- load_data_from_prepared, used to load data from prepared statement rows
- load_data_from_local, used to load data from normal statement rows

The selection of which method to use is done at LookupProcessor creation time

Author: andsel

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 5.y.z
+  - Refactored Lookup used in jdbc_streaming and jdbc_static to avoid code duplication. [#59](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/59)
+
 ## 5.0.6
   - DOC:Replaced plugin_header file with plugin_header-integration file. [#40](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/40)
 

lib/logstash/filters/jdbc/lookup.rb

@@ -66,6 +66,7 @@ def initialize(options, globals, default_id)
       @prepared_statement = nil
       @symbol_parameters = nil
       parse_options
+      @load_method_ref = method(:load_data_from_local)
     end
 
     def id_used_as_target?
@@ -81,11 +82,7 @@ def formatted_errors
     end
 
     def enhance(local, event)
-      if @prepared_statement
-        result = call_prepared(local, event)
-      else
-        result = fetch(local, event) # should return a LookupResult
-      end
+      result = retrieve_local_data(local, event, &@load_method_ref) # return a LookupResult
       if result.failed? || result.parameters_invalid?
         tag_failure(event)
       end
@@ -112,6 +109,7 @@ def prepare(local)
       @prepared_parameters.each_with_index { |v, i| hash[:"$p#{i}"] = v }
       @prepared_param_placeholder_map = hash
       @prepared_statement = local.prepare(query, hash.keys)
+      @load_method_ref = method(:load_data_from_prepared)
     end
 
     private
@@ -128,34 +126,23 @@ def tag_default(event)
       end
     end
 
-    def fetch(local, event)
-      result = LookupResult.new()
-      if @parameters_specified
-        params = prepare_parameters_from_event(event, result)
-        if result.parameters_invalid?
-          logger.warn? && logger.warn("Parameter field not found in event", :lookup_id => @id, :invalid_parameters => result.invalid_parameters)
-          return result
-        end
-      else
-        params = {}
+    def load_data_from_local(local, query, params, result)
+      local.fetch(query, params).each do |row|
+        stringified = row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} #Stringify row keys
+        result.push(stringified)
       end
-      begin
-        logger.debug? && logger.debug("Executing Jdbc query", :lookup_id => @id, :statement => query, :parameters => params)
-        local.fetch(query, params).each do |row|
-          stringified = row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} #Stringify row keys
-          result.push(stringified)
-        end
-      rescue ::Sequel::Error => e
-        # all sequel errors are a subclass of this, let all other standard or runtime errors bubble up
-        result.failed!
-        logger.warn? && logger.warn("Exception when executing Jdbc query", :lookup_id => @id, :exception => e.message, :backtrace => e.backtrace.take(8))
+    end
+
+    def load_data_from_prepared(_local, _query, params, result)
+      @prepared_statement.call(params).each do |row|
+        stringified = row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} #Stringify row keys
+        result.push(stringified)
       end
-      # if either of: no records or a Sequel exception occurs the payload is
-      # empty and the default can be substituted later.
-      result
     end
 
-    def call_prepared(local, event)
+    # the &block is invoked with 4 arguments: local, query[String], params[Hash], result[LookupResult]
+    # the result is used as accumulator return variable
+    def retrieve_local_data(local, event, &proc)
       result = LookupResult.new()
       if @parameters_specified
         params = prepare_parameters_from_event(event, result)
@@ -168,10 +155,7 @@ def call_prepared(local, event)
       end
       begin
         logger.debug? && logger.debug("Executing Jdbc query", :lookup_id => @id, :statement => query, :parameters => params)
-        @prepared_statement.call(params).each do |row|
-          stringified = row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} #Stringify row keys
-          result.push(stringified)
-        end
+        proc.call(local, query, params, result)
       rescue ::Sequel::Error => e
         # all sequel errors are a subclass of this, let all other standard or runtime errors bubble up
         result.failed!

spec/filters/integration/jdbc_static_spec.rb

@@ -59,7 +59,7 @@ module LogStash module Filters
 
     let(:plugin) { JdbcStatic.new(settings) }
 
-    let(:event)      { ::LogStash::Event.new("message" => "some text", "ip" => ipaddr) }
+    let(:event) { ::LogStash::Event.new("message" => "some text", "ip" => ipaddr) }
 
     let(:ipaddr) { ".3.1.1" }