Class: Carrot::AMQP::Queue
- Inherits:
-
Object
- Object
- Carrot::AMQP::Queue
- Defined in:
- lib/amqp/queue.rb
Instance Attribute Summary collapse
-
#carrot ⇒ Object
readonly
Returns the value of attribute carrot.
-
#delivery_tag ⇒ Object
Returns the value of attribute delivery_tag.
-
#name ⇒ Object
readonly
Returns the value of attribute name.
Instance Method Summary collapse
- #ack ⇒ Object
- #bind(exchange, opts = {}) ⇒ Object
- #consumer_count ⇒ Object
- #delete(opts = {}) ⇒ Object
-
#initialize(carrot, name, opts = {}) ⇒ Queue
constructor
A new instance of Queue.
- #message_count ⇒ Object
- #pop(opts = {}) ⇒ Object
- #publish(data, opts = {}) ⇒ Object
- #purge(opts = {}) ⇒ Object
- #server ⇒ Object
- #status(opts = {}, &blk) ⇒ Object
- #unbind(exchange, opts = {}) ⇒ Object
Constructor Details
#initialize(carrot, name, opts = {}) ⇒ Queue
Returns a new instance of Queue.
6 7 8 9 10 11 12 13 |
# File 'lib/amqp/queue.rb', line 6 def initialize(carrot, name, opts = {}) @opts = opts @name = name @carrot = carrot server.send_frame( Protocol::Queue::Declare.new({ :queue => name, :nowait => true }.merge(opts)) ) end |
Instance Attribute Details
#carrot ⇒ Object (readonly)
Returns the value of attribute carrot.
3 4 5 |
# File 'lib/amqp/queue.rb', line 3 def carrot @carrot end |
#delivery_tag ⇒ Object
Returns the value of attribute delivery_tag.
4 5 6 |
# File 'lib/amqp/queue.rb', line 4 def delivery_tag @delivery_tag end |
#name ⇒ Object (readonly)
Returns the value of attribute name.
3 4 5 |
# File 'lib/amqp/queue.rb', line 3 def name @name end |
Instance Method Details
#ack ⇒ Object
35 36 37 38 39 |
# File 'lib/amqp/queue.rb', line 35 def ack server.send_frame( Protocol::Basic::Ack.new(:delivery_tag => delivery_tag) ) end |
#bind(exchange, opts = {}) ⇒ Object
63 64 65 66 67 68 69 |
# File 'lib/amqp/queue.rb', line 63 def bind(exchange, opts = {}) exchange = exchange.respond_to?(:name) ? exchange.name : exchange bindings[exchange] = opts server.send_frame( Protocol::Queue::Bind.new({ :queue => name, :exchange => exchange, :routing_key => opts.delete(:key), :nowait => true }.merge(opts)) ) end |
#consumer_count ⇒ Object
49 50 51 |
# File 'lib/amqp/queue.rb', line 49 def consumer_count status.last end |
#delete(opts = {}) ⇒ Object
82 83 84 85 86 87 |
# File 'lib/amqp/queue.rb', line 82 def delete(opts = {}) server.send_frame( Protocol::Queue::Delete.new({ :queue => name, :nowait => true }.merge(opts)) ) carrot.queues.delete(name) end |
#message_count ⇒ Object
45 46 47 |
# File 'lib/amqp/queue.rb', line 45 def status.first end |
#pop(opts = {}) ⇒ Object
15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
# File 'lib/amqp/queue.rb', line 15 def pop(opts = {}) self.delivery_tag = nil server.send_frame( Protocol::Basic::Get.new({ :queue => name, :consumer_tag => name, :no_ack => !opts.delete(:ack), :nowait => true }.merge(opts)) ) method = server.next_method return unless method.kind_of?(Protocol::Basic::GetOk) self.delivery_tag = method.delivery_tag header = server.next_payload msg = '' while msg.length < header.size msg << server.next_payload end msg end |
#publish(data, opts = {}) ⇒ Object
41 42 43 |
# File 'lib/amqp/queue.rb', line 41 def publish(data, opts = {}) exchange.publish(data, opts) end |
#purge(opts = {}) ⇒ Object
89 90 91 92 93 |
# File 'lib/amqp/queue.rb', line 89 def purge(opts = {}) server.send_frame( Protocol::Queue::Purge.new({ :queue => name, :nowait => true }.merge(opts)) ) end |
#server ⇒ Object
95 96 97 |
# File 'lib/amqp/queue.rb', line 95 def server carrot.server end |
#status(opts = {}, &blk) ⇒ Object
53 54 55 56 57 58 59 60 61 |
# File 'lib/amqp/queue.rb', line 53 def status(opts = {}, &blk) server.send_frame( Protocol::Queue::Declare.new({ :queue => name, :passive => true }.merge(opts)) ) method = server.next_method return [nil, nil] if method.kind_of?(Protocol::Connection::Close) [method., method.consumer_count] end |
#unbind(exchange, opts = {}) ⇒ Object
71 72 73 74 75 76 77 78 79 80 |
# File 'lib/amqp/queue.rb', line 71 def unbind(exchange, opts = {}) exchange = exchange.respond_to?(:name) ? exchange.name : exchange bindings.delete(exchange) server.send_frame( Protocol::Queue::Unbind.new({ :queue => name, :exchange => exchange, :routing_key => opts.delete(:key), :nowait => true }.merge(opts) ) ) end |