Class: LogStash::Inputs::Redis

Inherits:
Threadable
  • Object
show all
Defined in:
lib/logstash/inputs/redis.rb

Constant Summary collapse

BATCH_EMPTY_SLEEP = 
0.25

Instance Method Summary collapse

Instance Method Details

#add_external_redis_builder(builder) ⇒ Object

public API use to store a proc that can provide a redis instance or mock



55
56
57
58
 
# File 'lib/logstash/inputs/redis.rb', line 55

def add_external_redis_builder(builder) #callable
  @redis_builder = builder
  self
end

#new_redis_instance ⇒ Object 



66
67
68
 
# File 'lib/logstash/inputs/redis.rb', line 66

def new_redis_instance
  @redis_builder.call
end

#register ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/logstash/inputs/redis.rb', line 70

def register
  @redis_url = "redis://#{@password}@#{@host}:#{@port}/#{@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)
    @stop_method = method(:list_stop)
  elsif @data_type == 'channel'
    @run_method = method(:channel_runner)
    @stop_method = method(:subscribe_stop)
  elsif @data_type == 'pattern_channel'
    @run_method = method(:pattern_channel_runner)
    @stop_method = method(:subscribe_stop)
  end

  @list_method = batched? ? method(:list_batch_listener) : method(:list_single_listener)

  @identity = "#{@redis_url} #{@data_type}:#{@key}"
  @logger.info("Registering Redis", :identity => @identity)
end

#run(output_queue) ⇒ Object

def register



93
94
95
96
97
 
# File 'lib/logstash/inputs/redis.rb', line 93

def run(output_queue)
  @run_method.call(output_queue)
rescue LogStash::ShutdownSignal
  # ignore and quit
end

#stop ⇒ Object

def run



99
100
101
 
# File 'lib/logstash/inputs/redis.rb', line 99

def stop
  @stop_method.call
end

#use_redis(instance) ⇒ Object

use to apply an instance directly and bypass the builder



61
62
63
64
 
# File 'lib/logstash/inputs/redis.rb', line 61

def use_redis(instance)
  @redis = instance
  self
end