Class: Sensu::Transport::RabbitMQ

Inherits:
Base
  • Object
show all
Defined in:
lib/sensu/transport/rabbitmq.rb

Instance Attribute Summary

Attributes inherited from Base

#logger

Instance Method Summary collapse

Methods inherited from Base

#ack, #after_reconnect, #before_reconnect, descendants, #on_error

Constructor Details

#initialize ⇒ RabbitMQ 



10
11
12
13
 
# File 'lib/sensu/transport/rabbitmq.rb', line 10

def initialize
  super
  @queues = {}
end

Instance Method Details

#acknowledge(info, &callback) ⇒ Object 



87
88
89
90
 
# File 'lib/sensu/transport/rabbitmq.rb', line 87

def acknowledge(info, &callback)
  info.ack
  callback.call(info) if callback
end

#close ⇒ Object 



46
47
48
 
# File 'lib/sensu/transport/rabbitmq.rb', line 46

def close
  @connection.close
end

#connect(options = {}) ⇒ Object 



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/sensu/transport/rabbitmq.rb', line 15

def connect(options={})
  create_connection_timeout
  @connection = AMQP.connect(options, {
    :on_tcp_connection_failure => on_connection_failure,
    :on_possible_authentication_failure => on_connection_failure
  })
  @connection.logger = @logger
  @connection.on_open do
    @connection_timeout.cancel
  end
  reconnect_callback = Proc.new { reconnect }
  @connection.on_tcp_connection_loss(&reconnect_callback)
  @connection.on_skipped_heartbeats(&reconnect_callback)
  @connection.on_recovery do
    @after_reconnect.call
  end
  setup_channel(options)
end

#connected? ⇒ Boolean 



42
43
44
 
# File 'lib/sensu/transport/rabbitmq.rb', line 42

def connected?
  @connection.connected?
end

#publish(exchange_type, exchange_name, message, options = {}, &callback) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/sensu/transport/rabbitmq.rb', line 50

def publish(exchange_type, exchange_name, message, options={}, &callback)
  begin
    @channel.method(exchange_type.to_sym).call(exchange_name, options).publish(message) do
      info = {}
      callback.call(info) if callback
    end
  rescue => error
    info = {:error => error}
    callback.call(info) if callback
  end
end

#reconnect ⇒ Object 



34
35
36
37
38
39
40
 
# File 'lib/sensu/transport/rabbitmq.rb', line 34

def reconnect
  unless @connection.reconnecting?
    @connection_timeout.cancel
    @before_reconnect.call
    @connection.periodically_reconnect(5)
  end
end

#stats(queue_name, options = {}, &callback) ⇒ Object 



92
93
94
95
96
97
98
99
100
101
 
# File 'lib/sensu/transport/rabbitmq.rb', line 92

def stats(queue_name, options={}, &callback)
  options = options.merge(:auto_delete => true)
  @channel.queue(queue_name, options).status do |messages, consumers|
    info = {
      :messages => messages,
      :consumers => consumers
    }
    callback.call(info)
  end
end

#subscribe(exchange_type, exchange_name, queue_name = "", options = {}, &callback) ⇒ Object 



62
63
64
65
66
67
68
69
70
 
# File 'lib/sensu/transport/rabbitmq.rb', line 62

def subscribe(exchange_type, exchange_name, queue_name="", options={}, &callback)
  previously_declared = @queues.has_key?(queue_name)
  @queues[queue_name] ||= @channel.queue!(queue_name, :auto_delete => true)
  queue = @queues[queue_name]
  queue.bind(@channel.method(exchange_type.to_sym).call(exchange_name))
  unless previously_declared
    queue.subscribe(options, &callback)
  end
end

#unsubscribe(&callback) ⇒ Object 



72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/sensu/transport/rabbitmq.rb', line 72

def unsubscribe(&callback)
  @queues.values.each do |queue|
    if connected?
      queue.unsubscribe
    else
      queue.before_recovery do
        queue.unsubscribe
      end
    end
  end
  @queues = {}
  @channel.recover if connected?
  super
end