Merge branch 'master' of git://github.com/tmm1/amqp

conflicts resolved

Conflicts:

	lib/amqp.rb
	lib/amqp/protocol.rb
	lib/mq.rb
	lib/mq/exchange.rb
	lib/mq/queue.rb
	lib/mq/rpc.rb

amqp.gemspec

@@ -32,6 +32,7 @@ spec = Gem::Specification.new do |s|
              "lib/ext/em.rb",
              "lib/ext/emfork.rb",
              "lib/mq/exchange.rb",
+             "lib/mq/header.rb",
              "lib/mq/logger.rb",
              "lib/mq/queue.rb",
              "lib/mq/rpc.rb",

examples/mq/simple.rb

@@ -5,7 +5,7 @@
 EM.run do
 
   # connect to the amqp server
-  connection = AMQP.connect(:host => 'dev.rabbitmq.com', :logging => false)
+  connection = AMQP.connect(:host => 'localhost', :logging => false)
 
   # open a channel on the AMQP connection
   channel = MQ.new(connection)

lib/mq.rb

@@ -5,7 +5,7 @@
 require 'amqp'
 
 class MQ
-  %w[ exchange queue rpc ].each do |file|
+  %w[ exchange queue rpc header ].each do |file|
     require "mq/#{file}"
   end
 

lib/mq/exchange.rb

@@ -22,6 +22,172 @@ class MQ
   class Exchange
     include AMQP
 
+    # Defines, intializes and returns an Exchange to act as an ingress
+    # point for all published messages.
+    #
+    # There are three (3) supported Exchange types: direct, fanout and topic.
+    #
+    # As part of the standard, the server _must_ predeclare the direct exchange
+    # 'amq.direct' and the fanout exchange 'amq.fanout' (all exchange names 
+    # starting with 'amq.' are reserved). Attempts to declare an exchange using
+    # 'amq.' as the name will raise an MQ:Error and fail. In practice these
+    # default exchanges are never used directly by client code.
+    #
+    # == 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. This is because an exchange
+    # defined with the empty string uses the default pre-declared exchange
+    # called 'amq.direct'. In this case it needs to use :key to do its matching.
+    #
+    #  # exchange is named 'foo'
+    #  exchange = MQ::Exchange.new(MQ.new, :direct, 'foo')
+    #
+    #  # or, the exchange can use the default name (amq.direct) and perform
+    #  # routing comparisons using the :key
+    #  exchange = MQ::Exchange.new(MQ.new, :direct, "", :key => 'foo')
+    #  exchange.publish('some data') # will be delivered to queue bound to 'foo'
+    #
+    #  queue = MQ::Queue.new(MQ.new, 'foo')
+    #  # can receive data since the queue name and the exchange key match exactly
+    #  queue.pop { |data| puts "received data [#{data}]" }
+    #
+    # == 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.
+    #
+    # 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. Fanout exchanges
+    # defined with the empty string as the name use the default 'amq.fanout'.
+    # In this case it needs to use :key to do its matching.
+    #
+    #  EM.run do
+    #    clock = MQ::Exchange.new(MQ.new, :fanout, 'clock')
+    #    EM.add_periodic_timer(1) do
+    #      puts "\npublishing #{time = Time.now}"
+    #      clock.publish(Marshal.dump(time))
+    #    end
+    #
+    #    # one way of defining a queue
+    #    amq = MQ::Queue.new(MQ.new, 'every second')
+    #    amq.bind(MQ.fanout('clock')).subscribe do |time|
+    #      puts "every second received #{Marshal.load(time)}"
+    #    end
+    #
+    #    # defining a queue using the convenience method
+    #    # 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
+    #
+    # == 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 parameter.
+    #
+    # As part of the AMQP standard, each server _should_ predeclare a topic 
+    # exchange called 'amq.topic' (this is not required by the standard).
+    #
+    # 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::Exchange.new(MQ.new, :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
+    # therefore it is a good choice for high-performance and low-latency
+    # message publishing.
+    #
+    # Durable exchanges cause all messages to be written to non-volatile
+    # backing store (i.e. disk) prior to routing to any bound queues.
+    #
+    # * :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 initialize mq, type, name, opts = {}
       @mq = mq
       @type, @name = type, name

lib/mq/header.rb

@@ -0,0 +1,25 @@
+class MQ
+  class Header
+    include AMQP
+
+    def initialize(mq, header_obj)
+      @mq = mq
+      @header = header_obj
+    end
+
+    # Acknowledges the receipt of this message with the server.
+    def ack
+      @mq.callback do
+        @mq.send Protocol::Basic::Ack.new(:delivery_tag => properties[:delivery_tag])
+      end
+    end
+
+    def properties
+      @header.properties
+    end
+
+    def inspect
+      @header.inspect
+    end
+  end
+end

lib/mq/queue.rb

@@ -2,6 +2,65 @@ class MQ
   class Queue
     include AMQP
 
+    # 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).
+    #
+    # When a queue is created without a name, the server will generate a 
+    # unique name internally (not currently supported in this library).
+    #
+    # == 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 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).
+    #
+    # 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 an exchange to it.
+    #
+    # If the queue has been previously declared, this option is ignored
+    # on subsequent declarations.
+    #
+    # * :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 initialize mq, name, opts = {}
       @mq = mq
       @mq.queues[@name = name] ||= self
@@ -149,14 +208,14 @@ def delete opts = {}
     # method it will raise a channel or connection exception.
     #
     def pop opts = {}, &blk
-      @ack = opts[:no_ack] === false
+      @ack = generate_ack?(opts)
 
       @on_pop = blk if blk
 
       @mq.callback{
         @mq.send Protocol::Basic::Get.new({ :queue => name,
                                             :consumer_tag => name,
-                                            :no_ack => true,
+                                            :no_ack => no_ack?(opts),
                                             :nowait => true }.merge(opts))
         @mq.get_queue{ |q|
           q.push(self)
@@ -221,12 +280,13 @@ def subscribe opts = {}, &blk
       raise Error, 'already subscribed to the queue' if @on_msg
 
       @on_msg = blk
-      @ack = opts[:no_ack] === false
+      @on_msg_opts = opts
+      @ack = generate_ack?(opts)
 
       @mq.callback{
         @mq.send Protocol::Basic::Consume.new({ :queue => name,
                                                 :consumer_tag => @consumer_tag,
-                                                :no_ack => true,
+                                                :no_ack => no_ack?(opts),
                                                 :nowait => true }.merge(opts))
       }
       self
@@ -267,7 +327,7 @@ def receive headers, body
       end
 
       if cb = (@on_msg || @on_pop)
-        cb.call *(cb.arity == 1 ? [body] : [headers, body])
+        cb.call *(cb.arity == 1 ? [body] : [MQ::Header.new(@mq, headers), body])
       end
 
       if @ack && headers && !AMQP.closing
@@ -306,5 +366,18 @@ def cancelled
     def exchange
       @exchange ||= Exchange.new(@mq, :direct, '', :key => name)
     end
+
+    # Returns true if the options specified indicate that the AMQP
+    # library should autogenerate an Ack response after processing.
+    def generate_ack?(options)
+      options[:no_ack] === false && !options[:ack]
+    end
+
+    # Returns true if the options specified indicate that our
+    # request to the AMQP server should indicate that no Ack is required
+    # after delivering. (ie. no_ack == true)
+    def no_ack?(options)
+      !options[:ack]
+    end
   end
 end

lib/mq/rpc.rb

@@ -21,6 +21,25 @@ class MQ
   #  end
   #
   class RPC < BlankSlate
+    # 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.
+    #
     def initialize mq, queue, obj = nil
       @mq = mq
       @mq.rpcs[queue] ||= self