Class: Lapine::Consumer::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/lapine/consumer/runner.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(argv) ⇒ Runner

Returns a new instance of Runner.



18
19
20
21
22
 
# File 'lib/lapine/consumer/runner.rb', line 18

def initialize(argv)
  @argv = argv
  @message_count = 0
  @running_message_count = 0
end

Instance Attribute Details

#argv ⇒ Object (readonly)

Returns the value of attribute argv.



16
17
18
 
# File 'lib/lapine/consumer/runner.rb', line 16

def argv
  @argv
end

Instance Method Details

#config ⇒ Object 



73
74
75
 
# File 'lib/lapine/consumer/runner.rb', line 73

def config
  @config ||= Lapine::Consumer::Config.new.load(argv)
end

#handle_signals! ⇒ Object 



96
97
98
99
100
 
# File 'lib/lapine/consumer/runner.rb', line 96

def handle_signals!
  $STOP_LAPINE_CONSUMER = false
  Signal.trap('INT') { EventMachine.stop }
  Signal.trap('TERM') { $STOP_LAPINE_CONSUMER = true }
end

#logger ⇒ Object 



81
82
83
 
# File 'lib/lapine/consumer/runner.rb', line 81

def logger
  @logger ||= config.logfile ? ::Lapine::AnnotatedLogger.new(config.logfile) : ::Lapine::AnnotatedLogger.new(STDOUT)
end

#queue_properties ⇒ Object 



85
86
87
88
89
90
 
# File 'lib/lapine/consumer/runner.rb', line 85

def queue_properties
  {}.tap do |props|
    props.merge!(auto_delete: true) if config.transient?
    props.merge!(durable: true) unless config.transient?
  end
end

#run ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/lapine/consumer/runner.rb', line 24

def run
  handle_signals!
  Environmenter::Loader.new(config).load!
  logger.info 'starting Lapine::Consumer'

  @queue_properties = queue_properties
  EventMachine.run do
    topology.each_binding do |q, conn, routing_key, handlers|
      queue = conn.channel.queue(q, @queue_properties).bind(conn.exchange, routing_key: routing_key)
      queue.subscribe(ack: true) do |metadata, payload|
        process(metadata, payload, handlers)
        EventMachine.stop_event_loop if should_exit?
      end
      queues << queue
    end

    topology.each_queue_to_delete do |q, conn, routing_key, handlers|
      # if queue does not exist in RabbitMQ, skip processing
      # else
      queue = conn.channel.queue(q, @queue_properties)
      queues_to_delete << queue

      queue.subscribe(ack: true) do |metadata, payload|
        process(metadata, payload, handlers)
      end

      EventMachine.add_timer(0.5) do
        logger.info "Lapine::Consumer unbinding #{queue.name} from exchange: #{conn.exchange.name}, routing_key: #{routing_key}"
        queue.unbind(conn.exchange, routing_key: routing_key)
      end
    end

    if config.debug?
      EventMachine.add_periodic_timer(10) do
        logger.info "Lapine::Consumer messages processed=#{@message_count} running_count=#{@running_message_count}"
        @message_count = 0
      end
    end

    EventMachine.add_periodic_timer(5) do
      EventMachine.stop_event_loop if should_exit?
    end

    schedule_queue_deletion
  end

  logger.warn 'exiting Lapine::Consumer'
end

#should_exit? ⇒ Boolean

Returns:

  • (Boolean) 


92
93
94
 
# File 'lib/lapine/consumer/runner.rb', line 92

def should_exit?
  $STOP_LAPINE_CONSUMER
end

#topology ⇒ Object 



77
78
79
 
# File 'lib/lapine/consumer/runner.rb', line 77

def topology
  @topology ||= ::Lapine::Consumer::Topology.new(config, logger)
end