# encoding: utf-8 require "amq/client/queue" module AMQP # h2. What are AMQP queues? # # Queues store and forward messages to consumers. They similar to mailboxes in SMTP. # Messages flow from producing applications to {Exchange exchanges} that route them # to queues and finally queues deliver them to consumer applications (or consumer # applications fetch messages as needed). # # # h2. Concept of bindings # # Queues must be bound to at least one exchange in order to receive messages from publishers. # Learn more about bindings in {Exchange Exchange class documentation}. # # # h2. Queue names. Server-named queues. Predefined queues. # # Like an Exchange, queue names starting with 'amq.' are reserved for # internal use. Attempts to create queue names in violation of this # reservation will raise AMQP::Error (ACCESS_REFUSED). # # When a queue is created without a name, the server will generate a # unique name internally (not currently supported in this library). # # h2. Queue life-cycles. When use of server-named queues is optimal and when it isn't. # # To quote AMQP 0.9.1 spec, there are two common message queue life-cycles: # # * Durable message queues that are shared by many consumers and have an independent existence: i.e. they # will continue to exist and collect messages whether or not there are consumers to receive them. # * Temporary message queues that are private to one consumer and are tied to that consumer. When the # consumer disconnects, the message queue is deleted. # # There are some variations on these, such as shared message queues that are deleted when the last of # many consumers disconnects. # # One example of durable message queues is well-known services like event collectors (event loggers). # They are usually up whether there are services to log anything or not. Other applications know what # queues they use and can rely on those queues being around all the time, survive broker restarts and # in general be available should an application in the network need to use them. In this case, # explicitly named durable queues are optimal and coupling it creates between applications is not # an issue. Another scenario of a well-known long-lived service is distributed metadata/directory/locking server # like Apache Zookeeper, Google's Chubby or DNS. Services like this benefit from using well-known, not generated # queue names, and so do other applications that use them. # # Different scenario is in "a cloud settings" when some kind of workers/instances may come online and # go down basically any time and other applications cannot rely on them being available. Using well-known # queue names in this case is possible but server-generated, short-lived queues that are bound to # topic or fanout exchanges to receive relevant messages is a better idea. # # Imagine a service that processes an endless stream of events (Twitter is one example). When traffic goes # up, development operations may spin up additional applications instances in the cloud to handle the load. # Those new instances want to subscribe to receive messages to process but the rest of the system doesn't # know anything about them, rely on them being online or try to address them directly: they process events # from a shared stream and are not different from their peers. In a case like this, there is no reason for # message consumers to not use queue names generated by the broker. # # In general, use of explicitly named or server-named queues depends on messaging pattern your application needs. # {http://www.eaipatterns.com/ Enterprise Integration Patters} discusses many messaging patterns in depth. # RabbitMQ FAQ also has a section on {http://www.rabbitmq.com/faq.html#scenarios use cases}. # # # h2. Queue durability and persistence of messages. # # AMQP separates concept of durability of entities (queues, exchanges) from messages persistence. # Queues can be durable or transient. Durable queues survive broker restart, transient queues don't (they # have to be redeclared when broker comes back online). Long-living queues (see Queue life-cycle section above) # are usually durable, short-lived queues don't have to be. # # The concept of messages persistence is separate: messages may be published as persistent. That makes # AMQP broker persist them to disk. If the server is restarted, the system ensures that received persistent messages # are not lost. Simply publishing message to a durable exchange or the fact that queue(s) they are routed to # is durable doesn't make messages persistent: it all depends on persistence mode of the messages itself. # Publishing messages as persistent affects performance (just like with data stores, durability comes at a certain cost # in performance and vise versa). Pass :persistent => true to {Exchange#publish} to publish your message as persistent. # # Note that *only durable queues can be bound to durable exchanges*. # # # h2. Message ordering # # RabbitMQ FAQ explains {http://www.rabbitmq.com/faq.html#message-ordering ordering of messages in AMQP queues} # # # h2. Key methods # # Key methods of Queue class are # # * {Queue#bind} # * {Queue#subscribe} # * {Queue#pop} # * {Queue#delete} # * {Queue#purge} # * {Queue#unbind} # # # @note Please make sure you read a section on queue durability vs. messages # persistence. # # # @see http://bit.ly/hw2ELX AMQP 0.9.1 specification (Section 2.1.1) # @see AMQP::Exchange class Queue < AMQ::Client::Queue # # API # # Name of this queue attr_reader :name attr_reader :sync_bind # Options this queue object was instantiated with attr_accessor :opts attr_accessor :on_declare attr_accessor :on_bind # @option opts [Boolean] :passive (false) If set, the server will not create the queue if it does not # already exist. The client can use this to check whether the queue # exists without modifying the server state. # # @option opts [Boolean] :durable (false) If set when creating a new queue, the queue will be marked as # durable. Durable queues remain active when a server restarts. # Non-durable queues (transient queues) are purged if/when a # server restarts. Note that durable queues do not necessarily # hold persistent messages, although it does not make sense to # send persistent messages to a transient queue (though it is # allowed). # # @option opts [Boolean] :exclusive (false) Exclusive queues may only be consumed from by the current connection. # Setting the 'exclusive' flag always implies 'auto-delete'. Only a # single consumer is allowed to remove messages from this queue. # The default is a shared queue. Multiple clients may consume messages # from this queue. # # @option opts [Boolean] :auto_delete (false) If set, the queue is deleted when all consumers have finished # using it. Last consumer can be cancelled either explicitly or because # its channel is closed. If there was no consumer ever on the queue, it # won't be deleted. # # @option opts [Boolean] :nowait (true) If set, the server will not respond to the method. The client should # not wait for a reply method. If the server could not complete the # method it will raise a channel or connection exception. # # # @yield [queue, declare_ok] Yields successfully declared queue instance and AMQP method (queue.declare-ok) instance. The latter is optional. # @yieldparam [Queue] queue Queue that is successfully declared and is ready to be used. # @yieldparam [AMQP::Protocol::Queue::DeclareOk] declare_ok AMQP queue.declare-ok) instance. # # @api public def initialize(channel, name = AMQ::Protocol::EMPTY_STRING, opts = {}, &block) @channel = channel name = AMQ::Protocol::EMPTY_STRING if name.nil? @name = name unless name.empty? @opts = self.class.add_default_options(name, opts, block) @bindings = Hash.new if @opts[:nowait] @status = :opened block.call(self) if block else @status = :opening end super(channel.connection, channel, name) shim = Proc.new do |q, declare_ok| case block.arity when 1 then block.call(q) else block.call(q, declare_ok) end end @channel.once_open do if block self.declare(@opts[:passive], @opts[:durable], @opts[:exclusive], @opts[:auto_delete], @opts[:nowait], nil, &shim) else self.declare(@opts[:passive], @opts[:durable], @opts[:exclusive], @opts[:auto_delete], @opts[:nowait], nil) end end end # This method binds a queue to an exchange. Until a queue is # bound it will not receive any messages. In a classic messaging # model, store-and-forward queues are bound to a dest exchange # and subscription queues are bound to a dest_wild exchange. # # A valid exchange name (or reference) must be passed as the first # parameter. # @example Both of these are valid # exch = AMQP::Channel.direct('foo exchange') # queue = AMQP::Channel.queue('bar queue') # queue.bind('foo.exchange') # OR # queue.bind(exch) # # It is not valid to call #bind without the +exchange+ parameter. # # It is unnecessary to call #bind when the exchange name and queue # name match exactly (for :direct and :fanout exchanges only). # There is an implicit bind which will deliver the messages from # the exchange to the queue. # # @param [Exchange] Exchange to bind to. # # @option opts [String] :routing_key Specifies the routing key for the binding. The routing key is # used for routing messages depending on the exchange configuration. # Not all exchanges use a routing key! Refer to the specific # exchange documentation. If the routing key is empty and the queue # name is empty, the routing key will be the current queue for the # channel, which is the last declared queue. # # @option opts [Boolean] :nowait (true) If set, the server will not respond to the method. The client should # not wait for a reply method. If the server could not complete the # method it will raise a channel or connection exception. # @return [Queue] Self # # # @yield [] Since queue.bind-ok carries no attributes, no parameters are yielded to the block. # # @api public # @see Queue#unbind def bind(exchange, opts = {}, &block) @status = :unbound @sync_bind = !opts[:nowait] # amq-client's Queue already does exchange.respond_to?(:name) ? exchange.name : exchange # for us exchange = exchange @bindings[exchange] = opts @channel.once_open do super(exchange, (opts[:key] || opts[:routing_key] || AMQ::Protocol::EMPTY_STRING), (opts[:nowait] || block.nil?), opts[:arguments], &block) end self end # Remove the binding between the queue and exchange. The queue will # not receive any more messages until it is bound to another # exchange. # # Due to the asynchronous nature of the protocol, it is possible for # "in flight" messages to be received after this call completes. # Those messages will be serviced by the last block used in a # {Queue#subscribe} or {Queue#pop} call. # # @param [Exchange] Exchange to unbind from. # # @option opts [Boolean] :nowait (true) If set, the server will not respond to the method. The client should # not wait for a reply method. If the server could not complete the # method it will raise a channel or connection exception. # # # @yield [] Since queue.unbind-ok carries no attributes, no parameters are yielded to the block. # # @api public # @see Queue#bind def unbind(exchange, opts = {}, &block) @channel.once_open do super(exchange, (opts[:key] || opts[:routing_key] || AMQ::Protocol::EMPTY_STRING), opts[:arguments], &block) end end # This method deletes a queue. When a queue is deleted any pending # messages are sent to a dead-letter queue if this is defined in the # server configuration, and all consumers on the queue are cancelled. # # @return [NilClass] nil (for v0.7 compatibility) # # @option opts [Boolean] :if_unused (false) If set, the server will only delete the queue if it has no # consumers. If the queue has consumers the server does does not # delete it but raises a channel exception instead. # # @option opts [Boolean] :if_empty (false) If set, the server will only delete the queue if it has no # messages. If the queue is not empty the server raises a channel # exception. # # @option opts [Boolean] :nowait (false) If set, the server will not respond to the method. The client should # not wait for a reply method. If the server could not complete the # method it will raise a channel or connection exception. # # # @return [NilClass] nil (for v0.7 compatibility) # # @yield [delete_ok] Yields AMQP method (queue.delete-ok) instance. # @yieldparam [AMQP::Protocol::Queue::DeleteOk] delete_ok AMQP queue.delete-ok) instance. Carries number of messages that were in the queue. # # @api public # @see Queue#purge # @see Queue#unbind def delete(opts = {}, &block) @channel.once_open do super(opts.fetch(:if_unused, false), opts.fetch(:if_empty, false), opts.fetch(:nowait, false), &block) end # backwards compatibility nil end # This method removes all messages from a queue which are not awaiting acknowledgment. # # @option opts [Boolean] :nowait (false) If set, the server will not respond to the method. The client should # not wait for a reply method. If the server could not complete the # method it will raise a channel or connection exception. # # @return [NilClass] nil (for v0.7 compatibility) # # # @yield [purge_ok] Yields AMQP method (queue.purge-ok) instance. # @yieldparam [AMQP::Protocol::Queue::PurgeOk] purge_ok AMQP queue.purge-ok) instance. Carries number of messages that were purged. # # @api public # @see Queue#delete # @see Queue#unbind def purge(opts = {}, &block) @channel.once_open do super(opts.fetch(:nowait, false), &block) end # backwards compatibility nil end # This method provides a direct access to the messages in a queue # using a synchronous dialogue that is designed for specific types of # application where synchronous functionality is more important than # performance. # # If provided block takes one argument, it is passed message payload every time {Queue#pop} is called. # # @example Use of callback with a single argument # # EM.run do # exchange = AMQP::Channel.direct("foo queue") # EM.add_periodic_timer(1) do # exchange.publish("random number #{rand(1000)}") # end # # # note that #bind is never called; it is implicit because # # the exchange and queue names match # queue = AMQP::Channel.queue('foo queue') # queue.pop { |body| puts "received payload [#{body}]" } # # EM.add_periodic_timer(1) { queue.pop } # end # # If the block takes 2 parameters, both the header and the body will # be passed in for processing. The header object is defined by # AMQP::Protocol::Header. # # @example Use of callback with two arguments # # EM.run do # exchange = AMQP::Channel.direct("foo queue") # EM.add_periodic_timer(1) do # exchange.publish("random number #{rand(1000)}") # end # # queue = AMQP::Channel.queue('foo queue') # queue.pop do |header, body| # p header # puts "received payload [#{body}]" # end # # EM.add_periodic_timer(1) { queue.pop } # end # # @option opts [Boolean] :ack (false) If this field is set to false the server does not expect acknowledgments # for messages. That is, when a message is delivered to the client # the server automatically and silently acknowledges it on behalf # of the client. This functionality increases performance but at # the cost of reliability. Messages can get lost if a client dies # before it can deliver them to the application. # # # @return [Qeueue] Self # # # @yield [headers, payload] When block only takes one argument, yields payload to it. In case of two arguments, yields headers and payload. # @yieldparam [AMQP::Header] headers Headers (metadata) associated with this message (for example, routing key). # @yieldparam [String] payload Message body (content). On Ruby 1.9, you may want to check or enforce content encoding. # # @api public def pop(opts = {}, &block) if block # We have to maintain this multiple arities jazz # because older versions this gem are used in examples in at least 3 # books published by O'Reilly :(. MK. shim = Proc.new { |method, headers, payload| case block.arity when 1 then block.call(payload) when 2 then h = Header.new(@channel, method, headers ? headers.decode_payload : nil) block.call(h, payload) else h = Header.new(@channel, method, headers ? headers.decode_payload : nil) block.call(h, payload, method.delivery_tag, method.redelivered, method.exchange, method.routing_key) end } @channel.once_open do # see AMQ::Client::Queue#get in amq-client self.get(!opts.fetch(:ack, false), &shim) end else @channel.once_open { self.get(!opts.fetch(:ack, false)) } end end # Subscribes to asynchronous message delivery. # # The provided block is passed a single message each time the # exchange matches a message to this queue. # # # @example Use of callback with a single argument # # EM.run do # exchange = AMQP::Channel.direct("foo queue") # EM.add_periodic_timer(1) do # exchange.publish("random number #{rand(1000)}") # end # # queue = AMQP::Channel.queue('foo queue') # queue.subscribe { |body| puts "received payload [#{body}]" } # end # # If the block takes 2 parameters, both the header and the body will # be passed in for processing. The header object is defined by # AMQP::Protocol::Header. # # @example Use of callback with two arguments # # EM.run do # exchange = AMQP::Channel.direct("foo queue") # EM.add_periodic_timer(1) do # exchange.publish("random number #{rand(1000)}") # end # # # note that #bind is never called; it is implicit because # # the exchange and queue names match # queue = AMQP::Channel.queue('foo queue') # queue.subscribe do |header, body| # p header # puts "received payload [#{body}]" # end # end # # # @option opts [Boolean ]:ack (false) If this field is set to false the server does not expect acknowledgments # for messages. That is, when a message is delivered to the client # the server automatically and silently acknowledges it on behalf # of the client. This functionality increases performance but at # the cost of reliability. Messages can get lost if a client dies # before it can deliver them to the application. # # @option opts [Boolean] :nowait (false) If set, the server will not respond to the method. The client should # not wait for a reply method. If the server could not complete the # method it will raise a channel or connection exception. # # @option opts [#call] :confirm (nil) If set, this proc will be called when the server confirms subscription # to the queue with a ConsumeOk message. Setting this option will # automatically set :nowait => false. This is required for the server # to send a confirmation. # # # @yield [headers, payload] When block only takes one argument, yields payload to it. In case of two arguments, yields headers and payload. # @yieldparam [AMQP::Header] headers Headers (metadata) associated with this message (for example, routing key). # @yieldparam [String] payload Message body (content). On Ruby 1.9, you may want to check or enforce content encoding. # # @return [Queue] Self # @api public def subscribe(opts = {}, &block) raise Error, 'already subscribed to the queue' if @consumer_tag # having initial value for @consumer_tag makes a lot of obscure issues # go away. It is set to real value once we receive consume-ok (it is handled by # AMQ::Client::Queue we inherit from). @consumer_tag = "for now" opts[:nowait] = false if (@on_confirm_subscribe = opts[:confirm]) # We have to maintain this multiple arities jazz # because older versions this gem are used in examples in at least 3 # books published by O'Reilly :(. MK. delivery_shim = Proc.new { |method, headers, payload| case block.arity when 1 then block.call(payload) when 2 then h = Header.new(@channel, method, headers.decode_payload) block.call(h, payload) else h = Header.new(@channel, method, headers.decode_payload) block.call(h, payload, method.consumer_tag, method.delivery_tag, method.redelivered, method.exchange, method.routing_key) end } @channel.once_open do @consumer_tag = nil # consumer_tag is set by AMQ::Client::Queue once we receive consume-ok, this takes a while. self.consume(!opts[:ack], opts[:exclusive], (opts[:nowait] || block.nil?), opts[:no_local], nil, &opts[:confirm]) end self.on_delivery(&delivery_shim) self end # Removes the subscription from the queue and cancels the consumer. # New messages will not be received by this queue instance. # # Due to the asynchronous nature of the protocol, it is possible for # "in flight" messages to be received after this call completes. # Those messages will be serviced by the last block used in a # {Queue#subscribe} or {Queue#pop} call. # # Additionally, if the queue was created with _autodelete_ set to # true, the server will delete the queue after its wait period # has expired unless the queue is bound to an active exchange. # # The method accepts a block which will be executed when the # unsubscription request is acknowledged as complete by the server. # # @option opts [Boolean] :nowait (false) If set, the server will not respond to the method. The client should # not wait for a reply method. If the server could not complete the # method it will raise a channel or connection exception. # # @yield [cancel_ok] # @yieldparam [AMQP::Protocol::Basic::CancelOk] cancel_ok AMQP method basic.cancel-ok. You can obtain consumer tag from it. # # # @api public def unsubscribe(opts = {}, &block) # @consumer_tag is nillified for us by AMQ::Client::Queue, that is, # our superclass. MK. @channel.once_open { self.cancel(opts.fetch(:nowait, true), &block) } end # Get the number of messages and active consumers (with active channel flow) on a queue. # # @example Getting number of messages and active consumers for a queue # # AMQP::Channel.queue('name').status { |number_of_messages, number_of_active_consumers| # puts number_of_messages # } # # @yield [number_of_messages, number_of_active_consumers] # @yieldparam [Fixnum] number_of_messages Number of messages in the queue # @yieldparam [Fixnum] number_of_active_consumers Number of active consumers for the queue. Note that consumers can suspend activity (Channel.Flow) in which case they do not appear in this count. # # @api public def status(opts = {}, &block) raise ArgumentError, "AMQP::Queue#status does not make any sense without a block" unless block shim = Proc.new { |q, declare_ok| block.call(declare_ok.message_count, declare_ok.consumer_count) } @channel.once_open { self.declare(true, @durable, @exclusive, @auto_delete, false, nil, &shim) } end # Boolean check to see if the current queue has already subscribed # to messages delivery. # # Attempts to {Queue#subscribe} multiple times to the same exchange will raise an # Exception. Only a single block at a time can be associated with any # queue instance for processing incoming messages. # # @return [Boolean] true if there is a consumer tag associated with this Queue instance # @api public def subscribed? !!@consumer_tag end # Compatibility alias for #on_declare. # # @api public # @deprecated def callback @on_declare end # Compatibility alias for #on_bind. # # @api public # @deprecated def bind_callback @on_bind end # Don't use this method. It is a leftover from very early days and # it ruins the whole point of exchanges/queue separation. # # @note This method will be removed before 1.0 release # @deprecated # @api public def publish(data, opts = {}) exchange.publish(data, opts) end # Resets queue state. Useful for error handling. # @api plugin def reset initialize(@channel, @name, @opts) end protected # @private def self.add_default_options(name, opts, block) { :queue => name, :nowait => block.nil? }.merge(opts) end private # Default direct exchange that we use to publish messages directly to this queue. # This is a leftover from very early days and will be removed before version 1.0. # # @deprecated def exchange @exchange ||= Exchange.new(@channel, :direct, AMQ::Protocol::EMPTY_STRING, :key => name) end end # Queue end # AMQP