Module: Qsagi::Queue::ClassMethods

Defined in:
lib/qsagi/queue.rb

Instance Method Summary collapse

Instance Method Details

#_exchange ⇒ Object 



64
65
66
 
# File 'lib/qsagi/queue.rb', line 64

def _exchange
  @exchange || ""
end

#_exchange_options ⇒ Object 



68
69
70
 
# File 'lib/qsagi/queue.rb', line 68

def _exchange_options
  @exchange_options || {}
end

#_message_class ⇒ Object 



72
73
74
 
# File 'lib/qsagi/queue.rb', line 72

def _message_class
  @message_class || Qsagi::Message
end

#_queue(options) ⇒ Object 



21
22
23
24
25
26
27
28
 
# File 'lib/qsagi/queue.rb', line 21

def _queue(options)
  standard_queue = StandardQueue.new(options)
  if options[:queue_type] == :confirmed
    ConfirmedQueue.new(standard_queue)
  else
    standard_queue
  end
end

#_serializer ⇒ Object 



76
77
78
 
# File 'lib/qsagi/queue.rb', line 76

def _serializer
  @serializer || Qsagi::DefaultSerializer
end

#connect(opts = {}, &block) ⇒ Object 



8
9
10
11
12
13
14
15
16
17
18
19
 
# File 'lib/qsagi/queue.rb', line 8

def connect(opts={}, &block)
  options = default_options.merge(opts)
  queue = _queue(options)

  begin
    queue.connect

    block.call(queue)
  ensure
    queue.disconnect
  end
end

#default_options ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/qsagi/queue.rb', line 30

def default_options
  {
    :host => host,
    :port => port,
    :queue_type => :standard,
    :heartbeat => heartbeat,
    :message_class => _message_class,
    :queue_name => queue_name,
    :durable => true,
    :queue_arguments => {"x-ha-policy" => "all"},
    :persistent => true,
    :mandatory => true,
    :serializer => _serializer,
    :exchange_options => _exchange_options,
    :exchange => _exchange,
    :connect_timeout => 5,
    :read_timeout => 5,
    :write_timeout => 5
  }
end

#exchange(exchange, options = {}) ⇒ Object 



51
52
53
54
 
# File 'lib/qsagi/queue.rb', line 51

def exchange(exchange, options = {})
  @exchange = exchange
  @exchange_options = {:type => :direct}.merge(options)
end

#message_class(message_class) ⇒ Object 



56
57
58
 
# File 'lib/qsagi/queue.rb', line 56

def message_class(message_class)
  @message_class = message_class
end

#serializer(serializer) ⇒ Object 



60
61
62
 
# File 'lib/qsagi/queue.rb', line 60

def serializer(serializer)
  @serializer = serializer
end