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, #initialize, #on_error

Constructor Details

This class inherits a constructor from Sensu::Transport::Base

Instance Method Details

#acknowledge(info, &callback) ⇒ Object



72
73
74
75
# File 'lib/sensu/transport/rabbitmq.rb', line 72

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

#closeObject



30
31
32
33
# File 'lib/sensu/transport/rabbitmq.rb', line 30

def close
  callback = Proc.new { @connection.close }
  connected? ? callback.call : EM.next_tick(callback)
end

#connect(options = {}) ⇒ Object



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

def connect(options={})
  reset
  set_connection_options(options)
  create_connection_timeout
  connect_with_eligible_options
end

#connected?Boolean



26
27
28
# File 'lib/sensu/transport/rabbitmq.rb', line 26

def connected?
  @connection.connected?
end

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



35
36
37
38
39
40
41
42
43
44
45
# File 'lib/sensu/transport/rabbitmq.rb', line 35

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

#reconnectObject



17
18
19
20
21
22
23
24
# File 'lib/sensu/transport/rabbitmq.rb', line 17

def reconnect
  unless @reconnecting
    @reconnecting = true
    @before_reconnect.call
    reset
    periodically_reconnect
  end
end

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



77
78
79
80
81
82
83
84
85
86
# File 'lib/sensu/transport/rabbitmq.rb', line 77

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



47
48
49
50
51
52
53
54
55
# File 'lib/sensu/transport/rabbitmq.rb', line 47

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



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

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