Class: LogStash::Inputs::Pulsar

Inherits:
Base
  • Object
show all
Defined in:
lib/logstash/inputs/pulsar.rb

Instance Method Summary collapse

Instance Method Details

#register ⇒ Object 



33
34
35
36
37
 
# File 'lib/logstash/inputs/pulsar.rb', line 33

def register
  logger.info("register logstash-input-pulsar")
  @runner_threads = []
  @runner_pulsar_clients = Array.new
end

#run(logstash_queue) ⇒ Object 



40
41
42
43
44
 
# File 'lib/logstash/inputs/pulsar.rb', line 40

def run(logstash_queue)
  @runner_consumers = consumer_threads.times.map { |i| create_consumer("#{client_id}-#{i}") }
  @runner_threads = @runner_consumers.map { |consumer| thread_runner(logstash_queue, consumer) }
  @runner_threads.each { |t| t.join }
end

#stop ⇒ Object 



48
49
50
51
52
53
54
55
56
 
# File 'lib/logstash/inputs/pulsar.rb', line 48

def stop
  logger.info("Stop all pulsar consumer !")
  @runner_consumers.each { |c| c.close }

  logger.info("Stop all pulsar client  !!!")
  @runner_pulsar_clients.each { |c| c.close }

  @runner_threads.each { |t| t.exit }
end