@@ -6,7 +6,8 @@ spec = Gem::Specification.new do |s| s.email = "amqp@tmm1.net" s.homepage = "http://amqp.rubyforge.org/" s.description = "AMQP client implementation in Ruby/EventMachine" - s.has_rdoc = false + s.has_rdoc = true + s.extra_rdoc_files = ['README'] s.authors = ["Aman Gupta"] s.add_dependency('eventmachine', '>= 0.12.2') amqp.gemspec @@ -41,6 +41,36 @@ module AMQP } end + # Must be called to startup the connection to the AMQP server. + # + # The method takes several arguments and an optional block. + # + # This takes any option that is also accepted by EventMachine::connect. + # Additionally, there are several AMQP-specific options. + # + # * :user => String (default 'guest') + # The username as defined by the AMQP server. + # * :pass => String (default 'guest') + # The password for the associated :user as defined by the AMQP server. + # * :vhost => String (default '/') + # The virtual host as defined by the AMQP server. + # * :timeout => Numeric (default nil) + # Measured in seconds. + # * :logging => true | false (default false) + # Toggle the extremely verbose logging of all protocol communications + # between the client and the server. Extremely useful for debugging. + # + # AMQP.start do + # # default to connecting to localhost:5672 + # + # # define queues, exchanges and bindings here. + # # also define all subscriptions and/or publishers + # # here. + # + # # this block never exits unless EM.stop_event_loop + # # is called. + # end + # def self.start *args, &blk EM.run{ @conn ||= connect *args lib/amqp.rb @@ -1,5 +1,5 @@ if [].map.respond_to? :with_index - class Array + class Array #:nodoc: def enum_with_index each.with_index end @@ -9,7 +9,7 @@ else end module AMQP - class Buffer + class Buffer #:nodoc: all class Overflow < StandardError; end class InvalidType < StandardError; end lib/amqp/buffer.rb @@ -115,10 +115,12 @@ module AMQP send_data data.to_s end + #:stopdoc: # def send_data data # log 'send_data', data # super # end + #:startdoc: def close &on_disconnect @on_disconnect = on_disconnect if on_disconnect lib/amqp/client.rb @@ -3,7 +3,7 @@ require 'amqp/buffer' require 'amqp/protocol' module AMQP - class Frame + class Frame #:nodoc: all def initialize payload = nil, channel = 0 @channel, @payload = channel, payload end lib/amqp/frame.rb @@ -3,6 +3,7 @@ require 'amqp/buffer' module AMQP module Protocol + #:stopdoc: class Class::Method def initialize *args opts = args.pop if args.last.is_a? Hash @@ -64,6 +65,28 @@ module AMQP end end + #:startdoc: + # Contains a properties hash that holds some potentially interesting + # information. + # * :delivery_mode + # 1 equals transient. + # 2 equals persistent. Unconsumed persistent messages will survive + # a server restart when they are stored in a durable queue. + # * :redelivered + # True or False + # * :routing_key + # The routing string used for matching this message to this queue. + # * :priority + # An integer in the range of 0 to 9 inclusive. + # * :content_type + # Always "application/octet-stream" (byte stream) + # * :exchange + # The source exchange which published this message. + # * :message_count + # The number of unconsumed messages contained in the queue. + # * :delivery_tag + # A monotonically increasing integer. This number should not be trusted + # as a sequence number. There is no guarantee it won't get reset. class Header def initialize *args opts = args.pop if args.last.is_a? Hash @@ -132,6 +155,7 @@ module AMQP class_id, method_id = buf.read(:short, :short) classes[class_id].methods[method_id].new(buf) end + #:stopdoc: end end lib/amqp/protocol.rb @@ -1,6 +1,7 @@ -# this file was autogenerated on Fri Aug 01 15:08:30 -0700 2008 -# using amqp-0.8.json (mtime: Fri Aug 01 15:08:22 -0700 2008) +#:stopdoc: +# this file was autogenerated on Sat Jan 03 14:05:57 -0600 2009 +# using amqp-0.8.json (mtime: Sat Jan 03 08:58:13 -0600 2009) # # DO NOT EDIT! (edit protocol/codegen.rb instead, and run `rake codegen`) lib/amqp/spec.rb @@ -1,7 +1,7 @@ unless defined?(BlankSlate) class BlankSlate < BasicObject; end if defined?(BasicObject) - class BlankSlate + class BlankSlate #:nodoc: instance_methods.each { |m| undef_method m unless m =~ /^__/ } end end \ No newline at end of file lib/ext/blankslate.rb @@ -5,6 +5,8 @@ rescue LoadError require 'eventmachine' end +#:stopdoc: + if EM::VERSION < '0.12.2' def EventMachine::run blk=nil, tail=nil, &block lib/ext/em.rb @@ -6,6 +6,8 @@ end require 'eventmachine' +#:stopdoc: + # helper to fork off EM reactors def EM.fork num = 1, &blk unless @forks lib/ext/emfork.rb @@ -1,3 +1,6 @@ +#:main: README +# + $:.unshift File.expand_path(File.dirname(File.expand_path(__FILE__))) require 'amqp' @@ -11,13 +14,31 @@ class MQ attr_accessor :logging end + # Raised whenever an illegal operation is attempted. class Error < StandardError; 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/forwards to the appropriate subclass method. class MQ include AMQP include EM::Deferrable + # 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 EventMachine::connect. + # + # EM.run do + # channel = MQ.new + # end + # + # EM.run do + # channel = MQ.new connect + # end + # def initialize connection = nil raise 'MQ can only be used from within EM.run{}' unless EM.reactor_running? @@ -30,6 +51,15 @@ class MQ end attr_reader :channel + # 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 @@ -117,18 +147,63 @@ class MQ } end - %w[ direct topic fanout ].each do |type| - class_eval %[ - def #{type} name = 'amq.#{type}', opts = {} - exchanges[name] ||= Exchange.new(self, :#{type}, name, opts) - end - ] + # A convenience method for defining a direct exchange. See + # MQ::Exchange.new for details and available options. + # + # direct_exch = MQ.direct('foo') + # # equivalent to + # direct_exch = MQ::Exchange.new(MQ.new, :direct, 'foo') + # + def direct name = 'amq.direct', opts = {} + exchanges[name] ||= Exchange.new(self, :direct, name, opts) + end + + # A convenience method for defining a fanout exchange. See + # MQ::Exchange.new for details and available options. + # + # fanout_exch = MQ.fanout('foo') + # # equivalent to + # fanout_exch = MQ::Exchange.new(MQ.new, :fanout, 'foo') + # + def fanout name = 'amq.fanout', opts = {} + exchanges[name] ||= Exchange.new(self, :fanout, name, opts) end + # A convenience method for defining a topic exchange. See + # MQ::Exchange.new for details and available options. + # + # topic_exch = MQ.topic('foo', :key => 'stocks.us') + # # equivalent to + # topic_exch = MQ::Exchange.new(MQ.new, :topic, 'foo', :key => 'stocks.us') + # + def topic name = 'amq.topic', opts = {} + exchanges[name] ||= Exchange.new(self, :topic, name, opts) + end + + # Convenience method for creating or retrieving a queue reference. Wraps + # calls to MQ::Queue. See the MQ::Queue class definition for the + # allowable options. + # + # queue = MQ.queue('bar', :durable => true) + # + # Equivalent to writing: + # channel = MQ.new + # queue = MQ::Queue.new(channel, 'bar', :durable => true) + # def queue name, opts = {} queues[name] ||= Queue.new(self, name, opts) end + # Convenience method for creating or retrieving an RPC (remote procedure + # call) reference. Wraps calls to MQ::RPC. See the MQ::RPC class definition + # for the allowable options. + # + # remote_proc = MQ.rpc('bar', Hash.new) + # + # Equivalent to writing: + # channel = MQ.new + # remote_proc = MQ::RPC.new(channel, 'bar', Hash.new) + # def rpc name, obj = nil rpcs[name] ||= RPC.new(self, name, obj) end @@ -144,8 +219,8 @@ class MQ end end - # error callback - + # Define a message and callback block to be executed on all + # errors. def self.error msg = nil, &blk if blk @error_callback = blk @@ -154,12 +229,16 @@ class MQ end end - # keep track of proxy objects - + # Returns a hash of all the exchange proxy objects. + # + # Not typically called by client code. def exchanges @exchanges ||= {} end + # Returns a hash of all the queue proxy objects. + # + # Not typically called by client code. def queues @queues ||= {} end @@ -172,12 +251,14 @@ class MQ 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 - + # Queue objects keyed on their consumer tags. def consumers @consumers ||= {} end @@ -194,11 +275,11 @@ class MQ alias :conn :connection end -# convenience wrapper (read: HACK) for thread-local MQ object +#-- convenience wrapper (read: HACK) for thread-local MQ object class MQ def MQ.default - # XXX clear this when connection is closed + #-- XXX clear this when connection is closed Thread.current[:mq] ||= MQ.new end @@ -207,8 +288,8 @@ class MQ end end -# unique identifier class MQ + # unique identifier def MQ.id Thread.current[:mq_id] ||= "#{`hostname`.strip}-#{Process.pid}-#{Thread.current.object_id}" end lib/mq.rb @@ -1,7 +1,181 @@ class MQ + # An Exchange acts as an ingress point for all published messages. An + # exchange may also be described as a router or a matcher. Every + # published message is received by an exchange which, depending on its + # type (described below), determines how to deliver the message. + # + # It determines the next delivery hop by examining the bindings associated + # with the exchange. + # 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 @@ -16,6 +190,42 @@ class MQ end attr_reader :name, :type, :key + # This method publishes a staged file message to a specific exchange. + # The file message will be routed to queues as defined by the exchange + # configuration and distributed to any active consumers when the + # transaction, if any, is committed. + # + # channel = MQ.new + # exchange = MQ::Exchange.new(channel, :direct, 'direct', :key => 'foo.bar') + # exchange.publish("some data") + # + # The method takes several hash key options which modify the behavior or + # lifecycle of the message. + # + # * :routing_key => 'string' + # + # Specifies the routing key for the message. The routing key is + # used for routing messages depending on the exchange configuration. + # + # * :mandatory => true | false (default false) + # + # This flag tells the server how to react if the message cannot be + # routed to a queue. If this flag is set, the server will return an + # unroutable message with a Return method. If this flag is zero, the + # server silently drops the message. + # + # * :immediate => true | false (default false) + # + # This flag tells the server how to react if the message cannot be + # routed to a queue consumer immediately. If this flag is set, the + # server will return an undeliverable message with a Return method. + # If this flag is zero, the server will queue the message, but with + # no guarantee that it will ever be consumed. + # + # * :persistent + # True or False. When true, this message will remain in the queue until + # it is consumed. When false, the message will be deleted. + # def publish data, opts = {} @mq.callback{ out = [] @@ -37,6 +247,28 @@ class MQ self end + # This method deletes an exchange. When an exchange is deleted all queue + # bindings on the exchange are cancelled. + # + # Further attempts to publish messages to a deleted exchange will raise + # an MQ::Error due to a channel close exception. + # + # exchange = MQ::Exchange.new(channel, :direct, 'direct', :key => 'foo.bar') + # exchange.delete + # + # == Options + # * :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. + # + # exchange.delete(:nowait => false) + # + # * :if_unused => true | false (default false) + # If set, the server will only delete the exchange if it has no queue + # bindings. If the exchange has queue bindings the server does not + # delete it but raises a channel exception instead (MQ:Error). + # def delete opts = {} @mq.callback{ @mq.send Protocol::Exchange::Delete.new({ :exchange => name, lib/mq/exchange.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 @@ -12,6 +71,32 @@ class MQ end attr_reader :name + # 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. Both of these are valid: + # exch = MQ::Exchange.new(MQ.new, :direct, 'foo exchange') + # queue = MQ::Queue.new(MQ.new, 'bar queue') + # queue.bind('foo.exchange') # OR + # queue.bind(exch) + # + # == Options + # * :key => 'some string' + # 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. + # + # * :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 bind exchange, opts = {} @mq.callback{ @mq.send Protocol::Queue::Bind.new({ :queue => name, @@ -32,6 +117,26 @@ class MQ self 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. + # + # == Options + # * :if_unused => true | false (default 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. + # + # * :if_empty => true | false (default 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. + # + # * :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 delete opts = {} @mq.callback{ @mq.send Protocol::Queue::Delete.new({ :queue => name, @@ -41,6 +146,58 @@ class MQ 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. + # + # The provided block is passed a single message each time pop is called. + # + # EM.run do + # exchange = MQ::Exchange.new(MQ.new, :direct, "foo queue")#, :key => 'foo queue') + # EM.add_periodic_timer(1) do + # exchange.publish("random number #{rand(1000)}") + # end + # + # queue = MQ::Queue.new(MQ.new, '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 headers and the body will + # be passed in for processing. The headers object is defined by + # AMQP::Protocol::Header. + # + # EM.run do + # exchange = MQ::Exchange.new(MQ.new, :direct, "foo queue")#, :key => 'foo queue') + # EM.add_periodic_timer(1) do + # exchange.publish("random number #{rand(1000)}") + # end + # + # queue = MQ::Queue.new(MQ.new, 'foo queue') + # queue.pop do |header, body| + # p header + # puts "received payload [#{body}]" } + # end + # + # EM.add_periodic_timer(1) { queue.pop } + # end + # + # == Options + # * :no_ack => true | false (default true) + # If this field is set 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. + # + # * :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 pop opts = {}, &blk @ack = opts[:no_ack] === false @@ -59,6 +216,52 @@ class MQ self end + # Subscribes to asynchronous message delivery. + # + # The provided block is passed a single message each time the + # exchange matches a message to this queue. + # + # EM.run do + # exchange = MQ::Exchange.new(MQ.new, :direct, "foo queue")#, :key => 'foo queue') + # EM.add_periodic_timer(1) do + # exchange.publish("random number #{rand(1000)}") + # end + # + # queue = MQ::Queue.new(MQ.new, 'foo queue') + # queue.subscribe { |body| puts "received payload [#{body}]" } + # end + # + # If the block takes 2 parameters, both the headers and the body will + # be passed in for processing. The headers object is defined by + # AMQP::Protocol::Header. + # + # EM.run do + # exchange = MQ::Exchange.new(MQ.new, :direct, "foo queue")#, :key => 'foo queue') + # EM.add_periodic_timer(1) do + # exchange.publish("random number #{rand(1000)}") + # end + # + # queue = MQ::Queue.new(MQ.new, 'foo queue') + # queue.subscribe do |header, body| + # p header + # puts "received payload [#{body}]" } + # end + # end + # + # == Options + # * :no_ack => true | false (default true) + # If this field is set 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. + # + # * :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 subscribe opts = {}, &blk @consumer_tag = "#{name}-#{Kernel.rand(999_999_999_999)}" @mq.consumers[@consumer_tag] = self @@ -90,6 +293,11 @@ class MQ exchange.publish(data, opts) end + # Passes the message to the block passed to pop or subscribe. + # + # Performs an arity check on the block's parameters. If arity == 1, + # pass only the message body. If arity != 1, pass the headers and + # the body to the block. def receive headers, body if AMQP.closing #You don't need this if your using ack, and if you aren't it doesn't do much good either lib/mq/queue.rb @@ -1,5 +1,45 @@ class MQ + # Basic RPC (remote procedure call) facility. + # + # Needs more detail and explanation. + # + # EM.run do + # server = MQ::RPC.new(MQ.new, 'hash table node', Hash) + # + # client = MQ::RPC.new(MQ.new, '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 + # 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 @@ -34,6 +74,20 @@ class MQ end end + # Calling MQ::RPC.new(*args) returns a proxy object without any methods beyond + # those in Object. All calls to the proxy are handled by #method_missing which + # works to marshal and unmarshal all method calls and their arguments. + # + # EM.run do + # server = MQ::RPC.new(MQ.new, 'hash table node', Hash) + # client = MQ::RPC.new(MQ.new, 'hash table node') + # + # # calls #method_missing on #[] which marshals the method name and + # # arguments to publish them to the remote + # client[:now] = Time.now + # .... + # end + # def method_missing meth, *args, &blk # XXX use uuids instead message_id = "random message id #{::Kernel.rand(999_999_999_999)}" lib/mq/rpc.rb @@ -12,6 +12,7 @@ s = JSON.parse(File.read(path)) require 'erb' puts ERB.new(%q[ + #:stopdoc: # this file was autogenerated on <%= Time.now.to_s %> # using <%= name.ljust(16) %> (mtime: <%= File.mtime(path) %>) # protocol/codegen.rb 809eba443246f1fd044a16ccf8539af14bdc93dd Chuck Remes cremes.devlist@mac.com http://github.com/dougbarth/amqp/commit/c4d207146d81c7289f79785191f0ae1932ceba5d c4d207146d81c7289f79785191f0ae1932ceba5d 2009-01-03T14:41:00-08:00 2009-01-03T14:25:30-08:00 Added rdoc to the main classes and modules used for building clients. - modified the gemspec to build the rdoc during installation and use the README file as the default index - added :nodoc: or :stopdoc: rdoc directives to all files in lib/ext - added :stopdoc: directives to codegen.rb so the spec.rb file does not get indexed by rdoc - added descriptive documentation to most of the major classes. A good chunk of the descriptions were copied verbatim from the amqp-0.8.xml file. Some code examples were repurposed from the examples directory to add color. More code examples needed. - removed some metaprogramming from mq.rb so that a few convenience methods could be properly documented. Signed-off-by: Aman Gupta <aman@tmm1.net> 5d656035e30f7e8db158f72769df62ae317836ae Aman Gupta aman@tmm1.net