# encoding: utf-8 #:main: README # $:.unshift File.expand_path(File.dirname(File.expand_path(__FILE__))) require 'amqp' require 'mq/collection' class MQ %w[ exchange queue rpc header ].each do |file| require "mq/#{file}" end class << self @logging = false attr_accessor :logging end # Raised whenever an illegal operation is attempted. class Error < StandardError; end class IncompatibleOptionsError < Error def initialize(name, opts_1, opts_2) super("There is already an instance called #{name} with options #{opts_1.inspect}, you can't define the same instance with different options (#{opts_2.inspect})!") end end class ChannelClosedError < Error def initialize(instance) super("The channel #{instance.channel} was closed, you can't use it anymore!") end end end # The top-level class for building AMQP clients. This class contains several # convenience methods for working with queues and exchanges. Many calls # delegate/forward to subclasses, but this is the preferred API. The subclass # API is subject to change while this high-level API will likely remain # unchanged as the library evolves. All code examples will be written using # the MQ API. # # Below is a somewhat complex example that demonstrates several capabilities # of the library. The example starts a clock using a +fanout+ exchange which # is used for 1 to many communications. Each consumer generates a queue to # receive messages and do some operation (in this case, print the time). # One consumer prints messages every second while the second consumer prints # messages every 2 seconds. After 5 seconds has elapsed, the 1 second # consumer is deleted. # # Of interest is the relationship of EventMachine to the process. All MQ # operations must occur within the context of an EM.run block. We start # EventMachine in its own thread with an empty block; all subsequent calls # to the MQ API add their blocks to the EM.run block. This demonstrates how # the library could be used to build up and tear down communications outside # the context of an EventMachine block and/or integrate the library with # other synchronous operations. See the EventMachine documentation for # more information. # # require 'rubygems' # require 'mq' # # thr = Thread.new { EM.run } # # # turns on extreme logging # #AMQP.logging = true # # def log *args # p args # end # # def publisher # clock = MQ.fanout('clock') # EM.add_periodic_timer(1) do # puts # # log :publishing, time = Time.now # clock.publish(Marshal.dump(time)) # end # end # # def one_second_consumer # MQ.queue('every second').bind(MQ.fanout('clock')).subscribe do |time| # log 'every second', :received, Marshal.load(time) # end # end # # def two_second_consumer # MQ.queue('every 2 seconds').bind('clock').subscribe do |time| # time = Marshal.load(time) # log 'every 2 seconds', :received, time if time.sec % 2 == 0 # end # end # # def delete_one_second # EM.add_timer(5) do # # delete the 'every second' queue # log 'Deleting [every second] queue' # MQ.queue('every second').delete # end # end # # publisher # one_second_consumer # two_second_consumer # delete_one_second # thr.join # # __END__ # # [:publishing, Tue Jan 06 22:46:14 -0600 2009] # ["every second", :received, Tue Jan 06 22:46:14 -0600 2009] # ["every 2 seconds", :received, Tue Jan 06 22:46:14 -0600 2009] # # [:publishing, Tue Jan 06 22:46:16 -0600 2009] # ["every second", :received, Tue Jan 06 22:46:16 -0600 2009] # ["every 2 seconds", :received, Tue Jan 06 22:46:16 -0600 2009] # # [:publishing, Tue Jan 06 22:46:17 -0600 2009] # ["every second", :received, Tue Jan 06 22:46:17 -0600 2009] # # [:publishing, Tue Jan 06 22:46:18 -0600 2009] # ["every second", :received, Tue Jan 06 22:46:18 -0600 2009] # ["every 2 seconds", :received, Tue Jan 06 22:46:18 -0600 2009] # ["Deleting [every second] queue"] # # [:publishing, Tue Jan 06 22:46:19 -0600 2009] # # [:publishing, Tue Jan 06 22:46:20 -0600 2009] # ["every 2 seconds", :received, Tue Jan 06 22:46:20 -0600 2009] # class MQ # # Behaviors # include AMQP include EM::Deferrable # # API # # Returns a new channel. A channel is a bidirectional virtual # connection between the client and the AMQP server. Elsewhere in the # library the channel is referred to in parameter lists as +mq+. # # Optionally takes the result from calling AMQP::connect. # # Rarely called directly by client code. This is implicitly called # by most instance methods. See #method_missing. # # EM.run do # channel = MQ.new # end # # EM.run do # channel = MQ.new AMQP::connect # end # def initialize(connection = nil) raise 'MQ can only be used from within EM.run {}' unless EM.reactor_running? @_send_mutex = Mutex.new @get_queue_mutex = Mutex.new @connection = connection || AMQP.start conn.callback { |c| @channel = c.add_channel(self) send Protocol::Channel::Open.new } end attr_reader :channel, :connection, :status alias :conn :connection def closed? @status.eql?(:closed) end # Defines, intializes and returns an Exchange to act as an ingress # point for all published messages. # # == Direct # A direct exchange is useful for 1:1 communication between a publisher and # subscriber. Messages are routed to the queue with a binding that shares # the same name as the exchange. Alternately, the messages are routed to # the bound queue that shares the same name as the routing key used for # defining the exchange. This exchange type does not honor the +:key+ option # when defining a new instance with a name. It _will_ honor the +:key+ option # if the exchange name is the empty string. # Allocating this exchange without a name _or_ with the empty string # will use the internal 'amq.direct' exchange. # # Any published message, regardless of its persistence setting, is thrown # away by the exchange when there are no queues bound to it. # # # exchange is named 'foo' # exchange = MQ.direct('foo') # # # or, the exchange can use the default name (amq.direct) and perform # # routing comparisons using the :key # exchange = MQ.direct("", :key => 'foo') # exchange.publish('some data') # will be delivered to queue bound to 'foo' # # queue = MQ.queue('foo') # # can receive data since the queue name and the exchange key match exactly # queue.pop { |data| puts "received data [#{data}]" } # # == Options # * :passive => true | false (default false) # If set, the server will not create the exchange if it does not # already exist. The client can use this to check whether an exchange # exists without modifying the server state. # # * :durable => true | false (default false) # If set when creating a new exchange, the exchange will be marked as # durable. Durable exchanges remain active when a server restarts. # Non-durable exchanges (transient exchanges) are purged if/when a # server restarts. # # A transient exchange (the default) is stored in memory-only. The # exchange and all bindings will be lost on a server restart. # It makes no sense to publish a persistent message to a transient # exchange. # # Durable exchanges and their bindings are recreated upon a server # restart. Any published messages not routed to a bound queue are lost. # # * :auto_delete => true | false (default false) # If set, the exchange is deleted when all queues have finished # using it. The server waits for a short period of time before # determining the exchange is unused to give time to the client code # to bind a queue to it. # # If the exchange has been previously declared, this option is ignored # on subsequent declarations. # # * :internal => true | false (default false) # If set, the exchange may not be used directly by publishers, but # only when bound to other exchanges. Internal exchanges are used to # construct wiring that is not visible to applications. # # * :nowait => true | false (default 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. # # == Exceptions # Doing any of these activities are illegal and will raise MQ:Error. # * redeclare an already-declared exchange to a different type # * :passive => true and the exchange does not exist (NOT_FOUND) # def direct(name = 'amq.direct', opts = {}, &block) if exchange = self.exchanges.find { |exchange| exchange.name == name } extended_opts = Exchange.add_default_options(:direct, name, opts, block) validate_parameters_match!(exchange, extended_opts) exchange else self.exchanges << Exchange.new(self, :direct, name, opts, &block) end end # Defines, intializes and returns an Exchange to act as an ingress # point for all published messages. # # == Fanout # A fanout exchange is useful for 1:N communication where one publisher # feeds multiple subscribers. Like direct exchanges, messages published # to a fanout exchange are delivered to queues whose name matches the # exchange name (or are bound to that exchange name). Each queue gets # its own copy of the message. # # Any published message, regardless of its persistence setting, is thrown # away by the exchange when there are no queues bound to it. # # Like the direct exchange type, this exchange type does not honor the # +:key+ option when defining a new instance with a name. It _will_ honor # the +:key+ option if the exchange name is the empty string. # Allocating this exchange without a name _or_ with the empty string # will use the internal 'amq.fanout' exchange. # # EM.run do # clock = MQ.fanout('clock') # EM.add_periodic_timer(1) do # puts "\npublishing #{time = Time.now}" # clock.publish(Marshal.dump(time)) # end # # amq = MQ.queue('every second') # amq.bind(MQ.fanout('clock')).subscribe do |time| # puts "every second received #{Marshal.load(time)}" # end # # # note the string passed to #bind # MQ.queue('every 5 seconds').bind('clock').subscribe do |time| # time = Marshal.load(time) # puts "every 5 seconds received #{time}" if time.strftime('%S').to_i%5 == 0 # end # end # # == Options # * :passive => true | false (default false) # If set, the server will not create the exchange if it does not # already exist. The client can use this to check whether an exchange # exists without modifying the server state. # # * :durable => true | false (default false) # If set when creating a new exchange, the exchange will be marked as # durable. Durable exchanges remain active when a server restarts. # Non-durable exchanges (transient exchanges) are purged if/when a # server restarts. # # A transient exchange (the default) is stored in memory-only. The # exchange and all bindings will be lost on a server restart. # It makes no sense to publish a persistent message to a transient # exchange. # # Durable exchanges and their bindings are recreated upon a server # restart. Any published messages not routed to a bound queue are lost. # # * :auto_delete => true | false (default false) # If set, the exchange is deleted when all queues have finished # using it. The server waits for a short period of time before # determining the exchange is unused to give time to the client code # to bind a queue to it. # # If the exchange has been previously declared, this option is ignored # on subsequent declarations. # # * :internal => true | false (default false) # If set, the exchange may not be used directly by publishers, but # only when bound to other exchanges. Internal exchanges are used to # construct wiring that is not visible to applications. # # * :nowait => true | false (default 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. # # == Exceptions # Doing any of these activities are illegal and will raise MQ:Error. # * redeclare an already-declared exchange to a different type # * :passive => true and the exchange does not exist (NOT_FOUND) # def fanout(name = 'amq.fanout', opts = {}, &block) if exchange = self.exchanges.find { |exchange| exchange.name == name } extended_opts = Exchange.add_default_options(:fanout, name, opts, block) validate_parameters_match!(exchange, extended_opts) exchange else self.exchanges << Exchange.new(self, :fanout, name, opts, &block) end end # Defines, intializes and returns an Exchange to act as an ingress # point for all published messages. # # == Topic # A topic exchange allows for messages to be published to an exchange # tagged with a specific routing key. The Exchange uses the routing key # to determine which queues to deliver the message. Wildcard matching # is allowed. The topic must be declared using dot notation to separate # each subtopic. # # This is the only exchange type to honor the +key+ hash key for all # cases. # # Any published message, regardless of its persistence setting, is thrown # away by the exchange when there are no queues bound to it. # # As part of the AMQP standard, each server _should_ predeclare a topic # exchange called 'amq.topic' (this is not required by the standard). # Allocating this exchange without a name _or_ with the empty string # will use the internal 'amq.topic' exchange. # # The classic example is delivering market data. When publishing market # data for stocks, we may subdivide the stream based on 2 # characteristics: nation code and trading symbol. The topic tree for # Apple Computer would look like: # 'stock.us.aapl' # For a foreign stock, it may look like: # 'stock.de.dax' # # When publishing data to the exchange, bound queues subscribing to the # exchange indicate which data interests them by passing a routing key # for matching against the published routing key. # # EM.run do # exch = MQ.topic("stocks") # keys = ['stock.us.aapl', 'stock.de.dax'] # # EM.add_periodic_timer(1) do # every second # puts # exch.publish(10+rand(10), :routing_key => keys[rand(2)]) # end # # # match against one dot-separated item # MQ.queue('us stocks').bind(exch, :key => 'stock.us.*').subscribe do |price| # puts "us stock price [#{price}]" # end # # # match against multiple dot-separated items # MQ.queue('all stocks').bind(exch, :key => 'stock.#').subscribe do |price| # puts "all stocks: price [#{price}]" # end # # # require exact match # MQ.queue('only dax').bind(exch, :key => 'stock.de.dax').subscribe do |price| # puts "dax price [#{price}]" # end # end # # For matching, the '*' (asterisk) wildcard matches against one # dot-separated item only. The '#' wildcard (hash or pound symbol) # matches against 0 or more dot-separated items. If none of these # symbols are used, the exchange performs a comparison looking for an # exact match. # # == Options # * :passive => true | false (default false) # If set, the server will not create the exchange if it does not # already exist. The client can use this to check whether an exchange # exists without modifying the server state. # # * :durable => true | false (default false) # If set when creating a new exchange, the exchange will be marked as # durable. Durable exchanges remain active when a server restarts. # Non-durable exchanges (transient exchanges) are purged if/when a # server restarts. # # A transient exchange (the default) is stored in memory-only. The # exchange and all bindings will be lost on a server restart. # It makes no sense to publish a persistent message to a transient # exchange. # # Durable exchanges and their bindings are recreated upon a server # restart. Any published messages not routed to a bound queue are lost. # # * :auto_delete => true | false (default false) # If set, the exchange is deleted when all queues have finished # using it. The server waits for a short period of time before # determining the exchange is unused to give time to the client code # to bind a queue to it. # # If the exchange has been previously declared, this option is ignored # on subsequent declarations. # # * :internal => true | false (default false) # If set, the exchange may not be used directly by publishers, but # only when bound to other exchanges. Internal exchanges are used to # construct wiring that is not visible to applications. # # * :nowait => true | false (default 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. # # == Exceptions # Doing any of these activities are illegal and will raise MQ:Error. # * redeclare an already-declared exchange to a different type # * :passive => true and the exchange does not exist (NOT_FOUND) # def topic(name = 'amq.topic', opts = {}, &block) if exchange = self.exchanges.find { |exchange| exchange.name == name } extended_opts = Exchange.add_default_options(:topic, name, opts, block) validate_parameters_match!(exchange, extended_opts) exchange else self.exchanges << Exchange.new(self, :topic, name, opts, &block) end end # Defines, intializes and returns an Exchange to act as an ingress # point for all published messages. # # == Headers # A headers exchange allows for messages to be published to an exchange # # Any published message, regardless of its persistence setting, is thrown # away by the exchange when there are no queues bound to it. # # As part of the AMQP standard, each server _should_ predeclare a headers # exchange called 'amq.match' (this is not required by the standard). # Allocating this exchange without a name _or_ with the empty string # will use the internal 'amq.match' exchange. # # TODO: The classic example is ... # # When publishing data to the exchange, bound queues subscribing to the # exchange indicate which data interests them by passing arguments # for matching against the headers in published messages. The # form of the matching can be controlled by the 'x-match' argument, which # may be 'any' or 'all'. If unspecified (in RabbitMQ at least), it defaults # to "all". # # A value of 'all' for 'x-match' implies that all values must match (i.e. # it does an AND of the headers ), while a value of 'any' implies that # at least one should match (ie. it does an OR). # # TODO: document behavior when either the binding or the message is missing # a header present in the other # # TODO: insert example # # == Options # * :passive => true | false (default false) # If set, the server will not create the exchange if it does not # already exist. The client can use this to check whether an exchange # exists without modifying the server state. # # * :durable => true | false (default false) # If set when creating a new exchange, the exchange will be marked as # durable. Durable exchanges remain active when a server restarts. # Non-durable exchanges (transient exchanges) are purged if/when a # server restarts. # # A transient exchange (the default) is stored in memory-only. The # exchange and all bindings will be lost on a server restart. # It makes no sense to publish a persistent message to a transient # exchange. # # Durable exchanges and their bindings are recreated upon a server # restart. Any published messages not routed to a bound queue are lost. # # * :auto_delete => true | false (default false) # If set, the exchange is deleted when all queues have finished # using it. The server waits for a short period of time before # determining the exchange is unused to give time to the client code # to bind a queue to it. # # If the exchange has been previously declared, this option is ignored # on subsequent declarations. # # * :internal => true | false (default false) # If set, the exchange may not be used directly by publishers, but # only when bound to other exchanges. Internal exchanges are used to # construct wiring that is not visible to applications. # # * :nowait => true | false (default 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. # # == Exceptions # Doing any of these activities are illegal and will raise MQ:Error. # * redeclare an already-declared exchange to a different type # * :passive => true and the exchange does not exist (NOT_FOUND) # * using a value other than "any" or "all" for "x-match" def headers(name = 'amq.match', opts = {}, &block) if exchange = self.exchanges.find { |exchange| exchange.name == name } extended_opts = Exchange.add_default_options(:headers, name, opts, block) validate_parameters_match!(exchange, extended_opts) exchange else self.exchanges << Exchange.new(self, :headers, name, opts, &block) end end # Queues store and forward messages. Queues can be configured in the server # or created at runtime. Queues must be attached to at least one exchange # in order to receive messages from publishers. # # 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 MQ:Error (ACCESS_REFUSED). # # It is not supported to create a queue without a name; some string # (even the empty string) must be passed in the +name+ parameter. # # == Options # * :passive => true | false (default 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. # # * :durable => true | false (default 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). # # Again, note the durability property on a queue has no influence on # the persistence of published messages. A durable queue containing # transient messages will flush those messages on a restart. # # If the queue has already been declared, any redeclaration will # ignore this setting. A queue may only be declared durable the # first time when it is created. # # * :exclusive => true | false (default 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. # # Attempting to redeclare an already-declared queue as :exclusive => true # will raise MQ:Error. # # * :auto_delete = true | false (default 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. # # The server waits for a short period of time before # determining the queue is unused to give time to the client code # to bind a queue to it. # # If the queue has been previously declared, this option is ignored # on subsequent declarations. # # Any remaining messages in the queue will be purged when the queue # is deleted regardless of the message's persistence setting. # # * :nowait => true | false (default 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. # def queue(name, opts = {}, &block) if queue = self.queues.find { |queue| queue.name == name } extended_opts = Queue.add_default_options(name, opts, block) validate_parameters_match!(queue, extended_opts) queue else self.queues << Queue.new(self, name, opts, &block) end end def queue!(name, opts = {}, &block) self.queues.add! Queue.new(self, name, opts, &block) end # Takes a channel, queue and optional object. # # The optional object may be a class name, module name or object # instance. When given a class or module name, the object is instantiated # during this setup. The passed queue is automatically subscribed to so # it passes all messages (and their arguments) to the object. # # Marshalling and unmarshalling the objects is handled internally. This # marshalling is subject to the same restrictions as defined in the # Marshal[http://ruby-doc.org/core/classes/Marshal.html] standard # library. See that documentation for further reference. # # When the optional object is not passed, the returned rpc reference is # used to send messages and arguments to the queue. See #method_missing # which does all of the heavy lifting with the proxy. Some client # elsewhere must call this method *with* the optional block so that # there is a valid destination. Failure to do so will just enqueue # marshalled messages that are never consumed. # # EM.run do # server = MQ.rpc('hash table node', Hash) # # client = MQ.rpc('hash table node') # client[:now] = Time.now # client[:one] = 1 # # client.values do |res| # p 'client', :values => res # end # # client.keys do |res| # p 'client', :keys => res # EM.stop_event_loop # end # end # def rpc(name, obj = nil) rpcs[name] ||= RPC.new(self, name, obj) end def close(&block) @on_close = block if @deferred_status == :succeeded send Protocol::Channel::Close.new(:reply_code => 200, :reply_text => 'bye', :method_id => 0, :class_id => 0) else @closing = true end end # Define a message and callback block to be executed on all # errors. def self.error msg = nil, &blk if blk @error_callback = blk else @error_callback.call(msg) if @error_callback and msg end end def prefetch(size) @prefetch_size = size send Protocol::Basic::Qos.new(:prefetch_size => 0, :prefetch_count => size, :global => false) self end # Asks the broker to redeliver all unacknowledged messages on this # channel. # # * requeue (default false) # If this parameter is false, the message will be redelivered to the original recipient. # If this flag is true, the server will attempt to requeue the message, potentially then # delivering it to an alternative subscriber. # def recover(requeue = false) send Protocol::Basic::Recover.new(:requeue => requeue) self end # Returns a hash of all the exchange proxy objects. # # Not typically called by client code. def exchanges @exchanges ||= MQ::Collection.new end # Returns a hash of all the queue proxy objects. # # Not typically called by client code. def queues @queues ||= MQ::Collection.new end def get_queue if block_given? @get_queue_mutex.synchronize { yield( @get_queue ||= [] ) } end end # Returns a hash of all rpc proxy objects. # # Not typically called by client code. def rpcs @rcps ||= {} end # Queue objects keyed on their consumer tags. # # Not typically called by client code. def consumers @consumers ||= {} end def reset @deferred_status = nil @channel = nil initialize @connection @consumers = {} exs = @exchanges @exchanges = MQ::Collection.new exs.each { |e| e.reset } if exs qus = @queues @queues = MQ::Collection.new qus.each { |q| q.reset } if qus prefetch(@prefetch_size) if @prefetch_size end # # Implementation # # May raise a MQ::Error exception when the frame payload contains a # Protocol::Channel::Close object. # # This usually occurs when a client attempts to perform an illegal # operation. A short, and incomplete, list of potential illegal operations # follows: # * publish a message to a deleted exchange (NOT_FOUND) # * declare an exchange using the reserved 'amq.' naming structure (ACCESS_REFUSED) # def process_frame(frame) log :received, frame case frame when Frame::Header @header = frame.payload @body = '' check_content_completion when Frame::Body @body << frame.payload check_content_completion when Frame::Method case method = frame.payload when Protocol::Channel::OpenOk send Protocol::Access::Request.new(:realm => '/data', :read => true, :write => true, :active => true, :passive => true) when Protocol::Access::RequestOk @ticket = method.ticket callback { send Protocol::Channel::Close.new(:reply_code => 200, :reply_text => 'bye', :method_id => 0, :class_id => 0) } if @closing succeed when Protocol::Basic::CancelOk if @consumer = consumers[ method.consumer_tag ] @consumer.cancelled else MQ.error "Basic.CancelOk for invalid consumer tag: #{method.consumer_tag}" end when Protocol::Exchange::DeclareOk # We can't use exchanges[method.exchange] because if the name would # be an empty string, then AMQP broker generated a random one. exchanges = self.exchanges.select { |exchange| exchange.opts[:nowait].eql?(false) } exchange = exchanges.reverse.find { |exchange| exchange.status.eql?(:unfinished) } exchange.receive_response method when Protocol::Queue::DeclareOk # We can't use queues[method.queue] because if the name would # be an empty string, then AMQP broker generated a random one. queues = self.queues.select { |queue| queue.opts[:nowait].eql?(false) } queue = queues.reverse.find { |queue| queue.status.eql?(:unfinished) } queue.receive_status method when Protocol::Queue::BindOk # We can't use queues[method.queue] because if the name would # be an empty string, then AMQP broker generated a random one. queues = self.queues.select { |queue| queue.sync_bind } queue = queues.reverse.find { |queue| queue.status.eql?(:unbound) } queue.after_bind method when Protocol::Basic::Deliver, Protocol::Basic::GetOk @method = method @header = nil @body = '' if method.is_a? Protocol::Basic::GetOk @consumer = get_queue { |q| q.shift } MQ.error "No pending Basic.GetOk requests" unless @consumer else @consumer = consumers[ method.consumer_tag ] MQ.error "Basic.Deliver for invalid consumer tag: #{method.consumer_tag}" unless @consumer end when Protocol::Basic::GetEmpty if @consumer = get_queue { |q| q.shift } @consumer.receive nil, nil else MQ.error "Basic.GetEmpty for invalid consumer" end when Protocol::Channel::Close @status = :closed MQ.error "#{method.reply_text} in #{Protocol.classes[method.class_id].methods[method.method_id]} on #{@channel}" when Protocol::Channel::CloseOk @status = :closed @on_close && @on_close.call(self) @closing = false conn.callback { |c| c.channels.delete @channel c.close if c.channels.empty? } when Protocol::Basic::ConsumeOk if @consumer = consumers[ method.consumer_tag ] @consumer.confirm_subscribe else MQ.error "Basic.ConsumeOk for invalid consumer tag: #{method.consumer_tag}" end end end end # process_frame def send(*args) conn.callback { |c| @_send_mutex.synchronize do args.each do |data| unless self.closed? data.ticket = @ticket if @ticket and data.respond_to? :ticket= log :sending, data c.send data, :channel => @channel else unless data.class == AMQP::Protocol::Channel::CloseOk raise ChannelClosedError.new(self) end end end end } end # send def check_content_completion if @body.length >= @header.size @header.properties.update(@method.arguments) @consumer.receive @header, @body if @consumer @body = @header = @consumer = @method = nil end end # check_content_completion private def log(*args) return unless MQ.logging pp args puts end # log def validate_parameters_match!(entity, parameters) unless entity.opts == parameters || parameters[:passive] raise IncompatibleOptionsError.new(entity.name, entity.opts, parameters) end end # validate_parameters_match!(entity, parameters) end #-- convenience wrapper (read: HACK) for thread-local MQ object class MQ def MQ.default #-- XXX clear this when connection is closed Thread.current[:mq] ||= MQ.new end # Allows for calls to all MQ instance methods. This implicitly calls # MQ.new so that a new channel is allocated for subsequent operations. def MQ.method_missing meth, *args, &blk MQ.default.__send__(meth, *args, &blk) end end class MQ # unique identifier def MQ.id Thread.current[:mq_id] ||= "#{`hostname`.strip}-#{Process.pid}-#{Thread.current.object_id}" end end