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 



66
67
68
 
# File 'lib/lapine/consumer/runner.rb', line 66

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

#handle_signals! ⇒ Object 



88
89
90
91
92
 
# File 'lib/lapine/consumer/runner.rb', line 88

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

#logger ⇒ Object 



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

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

#queue_properties ⇒ Object 



78
79
80
81
82
 
# File 'lib/lapine/consumer/runner.rb', line 78

def queue_properties
  {}.tap do |props|
    props.merge!(auto_delete: true) if 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
 
# File 'lib/lapine/consumer/runner.rb', line 24

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

  @queue_properties = queue_properties
  EventMachine.run do
    topology.each_binding do |q, conn, routing_key, classes|
      queue = conn.channel.queue(q, @queue_properties).bind(conn.exchange, routing_key: routing_key)
      queue.subscribe(ack: true) do |metadata, payload|

        message = Consumer::Message.new(payload, metadata, logger)
        Middleware.app.call(message) do |message|
          classes.each do |clazz|
            Lapine::Consumer::Dispatcher.new(clazz, message).dispatch
          end

          if config.debug?
            @message_count += 1
            @running_message_count += 1
          end
        end

        EventMachine.stop_event_loop if should_exit?
      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
  end

  logger.warn 'exiting Lapine::Consumer'
end

#should_exit? ⇒ Boolean

Returns:

  • (Boolean) 


84
85
86
 
# File 'lib/lapine/consumer/runner.rb', line 84

def should_exit?
  $STOP_LAPINE_CONSUMER
end

#topology ⇒ Object 



70
71
72
 
# File 'lib/lapine/consumer/runner.rb', line 70

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