Module: Qusion

Defined in:
lib/qusion.rb,
lib/qusion/amqp_config.rb,
lib/qusion/channel_pool.rb

Defined Under Namespace

Classes: AmqpConfig, ChannelPool

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.thread ⇒ Object (readonly)

Returns the value of attribute thread.



10
11
12
 
# File 'lib/qusion.rb', line 10

def thread
  @thread
end

Class Method Details

.channel ⇒ Object 



44
45
46
 
# File 'lib/qusion.rb', line 44

def self.channel
  ChannelPool.instance.channel
end

.channel_pool_size(new_pool_size) ⇒ Object 



48
49
50
 
# File 'lib/qusion.rb', line 48

def self.channel_pool_size(new_pool_size)
  ChannelPool.pool_size = new_pool_size
end

.die_gracefully_on_signal ⇒ Object 



39
40
41
42
 
# File 'lib/qusion.rb', line 39

def self.die_gracefully_on_signal
  Signal.trap("INT")  { graceful_stop }
  Signal.trap("TERM") { graceful_stop }
end

.graceful_stop ⇒ Object 



67
68
69
70
71
72
73
 
# File 'lib/qusion.rb', line 67

def self.graceful_stop
  EM.schedule do
    @graceful_stop = true
    AMQP.stop { EM.stop }
  end
  thread && thread.join
end

.ready_to_dispatch? ⇒ Boolean

Returns:

  • (Boolean) 


75
76
77
 
# File 'lib/qusion.rb', line 75

def self.ready_to_dispatch?
  EM.reactor_running? && AMQP.conn && AMQP.conn.connected?
end

.start(*opts) ⇒ Object 



13
14
15
16
 
# File 'lib/qusion.rb', line 13

def self.start(*opts)
  amqp_opts = AmqpConfig.new(*opts).config_opts
  start_amqp_dispatcher(amqp_opts)
end

.start_amqp_dispatcher(amqp_settings = {}) ⇒ Object 



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/qusion.rb', line 18

def self.start_amqp_dispatcher(amqp_settings={})
  AMQP.settings.merge!(amqp_settings)

  if defined?(::PhusionPassenger) && ::PhusionPassenger.respond_to?(:on_event)
    ::PhusionPassenger.on_event(:starting_worker_process) do |forked| 
      if forked
        EM.stop if EM.reactor_running?
        thread && thread.join
        Thread.current[:mq] = nil
        AMQP.instance_variable_set(:@conn, nil)
      end
      start_in_background
      die_gracefully_on_signal
    end
    return
  end

  start_in_background
  die_gracefully_on_signal
end

.start_in_background ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/qusion.rb', line 52

def self.start_in_background
  if EM.reactor_running?
    raise ArgumentError, 'AMQP already connected' if ready_to_dispatch?
    AMQP.start
  else
    raise ArgumentError, 'Qusion already started' if thread && thread.alive?
    @thread = Thread.new do
      EM.run { AMQP.start }
      raise "Premature AMQP shutdown" unless @graceful_stop
    end
    thread.abort_on_exception = true
    thread.join(0.1) until ready_to_dispatch?
  end
end