Fix: resolve crash when commands_map is set

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.6.1
+  - Fix: resolve crash when commands_map is set [#86](https://github.com/logstash-plugins/logstash-input-redis/pull/86)
+
 ## 3.6.0
   - Remove ruby pipeline dependency. Starting from Logstash 8, Ruby execution engine is not available. All pipelines should use Java pipeline [#84](https://github.com/logstash-plugins/logstash-input-redis/pull/84)
 

lib/logstash/inputs/redis.rb

@@ -61,28 +61,10 @@ module LogStash module Inputs class Redis < LogStash::Inputs::Threadable
   config :command_map, :validate => :hash, :default => {}
 
   public
-  # public API
-  # use to store a proc that can provide a Redis instance or mock
-  def add_external_redis_builder(builder) #callable
-    @redis_builder = builder
-    self
-  end
-
-  # use to apply an instance directly and bypass the builder
-  def use_redis(instance)
-    @redis = instance
-    self
-  end
-
-  def new_redis_instance
-    @redis_builder.call
-  end
 
   def register
     @redis_url = @path.nil? ? "redis://#{@password}@#{@host}:#{@port}/#{@db}" : "#{@password}@#{@path}/#{@db}"
 
-    @redis_builder ||= method(:internal_redis_builder)
-
     # just switch on data_type once
     if @data_type == 'list' || @data_type == 'dummy'
       @run_method = method(:list_runner)
@@ -147,8 +129,7 @@ def redis_params
     return connectionParams.merge(baseParams)
   end
 
-  # private
-  def internal_redis_builder
+  def new_redis_instance
     ::Redis.new(redis_params)
   end
 
@@ -157,14 +138,12 @@ def connect
     redis = new_redis_instance
 
     # register any renamed Redis commands
-    if @command_map.any?
-      client_command_map = redis.client.command_map
-      @command_map.each do |name, renamed|
-        client_command_map[name.to_sym] = renamed.to_sym
-      end
+    @command_map.each do |name, renamed|
+      redis._client.command_map[name.to_sym] = renamed.to_sym
     end
 
     load_batch_script(redis) if batched? && is_list_type?
+
     redis
   end # def connect
 
@@ -208,7 +187,9 @@ def list_runner(output_queue)
         @redis ||= connect
         @list_method.call(@redis, output_queue)
       rescue ::Redis::BaseError => e
-        @logger.warn("Redis connection problem", :exception => e)
+        info = { message: e.message, exception: e.class }
+        info[:backtrace] = e.backtrace if @logger.debug?
+        @logger.warn("Redis connection problem", info)
         # Reset the redis variable to trigger reconnect
         @redis = nil
         # this sleep does not need to be stoppable as its
@@ -270,14 +251,14 @@ def subscribe_stop
     return if @redis.nil? || !@redis.connected?
     # if its a SubscribedClient then:
     # it does not have a disconnect method (yet)
-    if @redis.client.is_a?(::Redis::SubscribedClient)
+    if @redis.subscribed?
       if @data_type == 'pattern_channel'
-        @redis.client.punsubscribe
+        @redis.punsubscribe
       else
-        @redis.client.unsubscribe
+        @redis.unsubscribe
       end
     else
-      @redis.client.disconnect
+      @redis.disconnect!
     end
     @redis = nil
   end

logstash-input-redis.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-redis'
-  s.version         = '3.6.0'
+  s.version         = '3.6.1'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads events from a Redis instance"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
@@ -23,7 +23,7 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
 
   s.add_runtime_dependency 'logstash-codec-json'
-  s.add_runtime_dependency 'redis', '~> 4'
+  s.add_runtime_dependency 'redis', '>= 4.0.1', '< 5'
 
   s.add_development_dependency 'logstash-devutils'
 end