lib/bunny/subscription08.rb lib/bunny/subscription09.rb lib/qrack/subscription.rb @@ -38,9 +38,12 @@ Gem::Specification.new do |s| "lib/bunny/exchange09.rb", "lib/bunny/queue08.rb", "lib/bunny/queue09.rb", + "lib/bunny/subscription08.rb", + "lib/bunny/subscription09.rb", "lib/qrack/client.rb", "lib/qrack/channel.rb", "lib/qrack/queue.rb", + "lib/qrack/subscription.rb", "lib/qrack/protocol/protocol08.rb", "lib/qrack/protocol/protocol09.rb", "lib/qrack/protocol/spec08.rb", bunny.gemspec @@ -22,7 +22,7 @@ q = b.queue('test1') q.publish('Hello everybody!') # get message from the queue -msg = q.pop +msg = q.pop[:payload] puts 'This is the message: ' + msg + "\n\n" examples/simple_08.rb @@ -22,7 +22,7 @@ q = b.queue('test1') q.publish('Hello everybody!') # get message from the queue -msg = q.pop +msg = q.pop[:payload] puts 'This is the message: ' + msg + "\n\n" examples/simple_09.rb @@ -22,7 +22,7 @@ q = b.queue('test1') q.publish('Testing acknowledgements') # get message from the queue -msg = q.pop(:ack => true) +msg = q.pop(:ack => true)[:payload] # acknowledge receipt of message q.ack examples/simple_ack_08.rb @@ -22,7 +22,7 @@ q = b.queue('test1') q.publish('Testing acknowledgements') # get message from the queue -msg = q.pop(:ack => true) +msg = q.pop(:ack => true)[:payload] # acknowledge receipt of message q.ack examples/simple_ack_09.rb @@ -44,18 +44,10 @@ exch = b.exchange('sorting_room') # bind queue to exchange q.bind(exch, :key => 'fred') -# initialize counter -i = 1 - # subscribe to queue -begin - ret = q.subscribe(:consumer_tag => 'testtag1', :timeout => 30) do |msg| - puts "#{i.to_s}: #{msg}" - i+=1 - end -rescue Qrack::ClientTimeout - puts '==== simple_consumer_08.rb timed out - closing down ====' - q.unsubscribe(:consumer_tag => 'testtag1') - # close the connection - b.stop -end \ No newline at end of file +q.subscribe(:consumer_tag => 'testtag1', :timeout => 30) do |msg| + puts "#{q.subscription.message_count}: #{msg[:payload]}" +end + +# Close client +b.stop \ No newline at end of file examples/simple_consumer_08.rb @@ -44,18 +44,10 @@ exch = b.exchange('sorting_room') # bind queue to exchange q.bind(exch, :key => 'fred') -# initialize counter -i = 1 - # subscribe to queue -begin - ret = q.subscribe(:consumer_tag => 'testtag1', :timeout => 30) do |msg| - puts "#{i.to_s}: #{msg}" - i+=1 - end -rescue Qrack::ClientTimeout - puts '==== simple_consumer_09.rb timed out - closing down ====' - q.unsubscribe(:consumer_tag => 'testtag1') - # close the connection - b.stop -end \ No newline at end of file +q.subscribe(:consumer_tag => 'testtag1', :timeout => 30) do |msg| + puts "#{q.subscription.message_count}: #{msg[:payload]}" +end + +# Close client +b.stop \ No newline at end of file examples/simple_consumer_09.rb @@ -30,9 +30,9 @@ q2.bind(exch) exch.publish('This message will be fanned out') # get message from the queues -msg = q1.pop +msg = q1.pop[:payload] puts 'This is the message from q1: ' + msg + "\n\n" -msg = q2.pop +msg = q2.pop[:payload] puts 'This is the message from q2: ' + msg + "\n\n" # close the client connection examples/simple_fanout_08.rb @@ -30,9 +30,9 @@ q2.bind(exch) exch.publish('This message will be fanned out') # get message from the queues -msg = q1.pop +msg = q1.pop[:payload] puts 'This is the message from q1: ' + msg + "\n\n" -msg = q2.pop +msg = q2.pop[:payload] puts 'This is the message from q2: ' + msg + "\n\n" # close the client connection examples/simple_fanout_09.rb @@ -32,7 +32,7 @@ header_exch.publish('Headers test msg 2', :headers => {'h1'=>'z'}) # get messages from the queue - should only be msg 1 that got through msg = "" until msg == :queue_empty do - msg = q.pop + msg = q.pop[:payload] puts 'This is a message from the header_q1 queue: ' + msg + "\n" unless msg == :queue_empty end examples/simple_headers_08.rb @@ -32,7 +32,7 @@ header_exch.publish('Headers test msg 2', :headers => {'h1'=>'z'}) # get messages from the queue - should only be msg 1 that got through msg = "" until msg == :queue_empty do - msg = q.pop + msg = q.pop[:payload] puts 'This is a message from the header_q1 queue: ' + msg + "\n" unless msg == :queue_empty end examples/simple_headers_09.rb @@ -38,20 +38,20 @@ sports_results.publish('British Lions 15 : South Africa 12', :key => 'rugby.resu # get message from the queues # soccer queue got the soccer message -msg = soccer.pop +msg = soccer.pop[:payload] puts 'This is a message from the soccer q: ' + msg + "\n\n" # cricket queue got the cricket message -msg = cricket.pop +msg = cricket.pop[:payload] puts 'This is a message from the cricket q: ' + msg + "\n\n" # rugby queue got the rugby message -msg = rugby.pop +msg = rugby.pop[:payload] puts 'This is a message from the rugby q: ' + msg + "\n\n" # allsport queue got all of the messages until msg == :queue_empty do - msg = allsport.pop + msg = allsport.pop[:payload] puts 'This is a message from the allsport q: ' + msg + "\n\n" unless msg == :queue_empty end examples/simple_topic_08.rb @@ -38,20 +38,20 @@ sports_results.publish('British Lions 15 : South Africa 12', :key => 'rugby.resu # get message from the queues # soccer queue got the soccer message -msg = soccer.pop +msg = soccer.pop[:payload] puts 'This is a message from the soccer q: ' + msg + "\n\n" # cricket queue got the cricket message -msg = cricket.pop +msg = cricket.pop[:payload] puts 'This is a message from the cricket q: ' + msg + "\n\n" # rugby queue got the rugby message -msg = rugby.pop +msg = rugby.pop[:payload] puts 'This is a message from the rugby q: ' + msg + "\n\n" # allsport queue got all of the messages until msg == :queue_empty do - msg = allsport.pop + msg = allsport.pop[:payload] puts 'This is a message from the allsport q: ' + msg + "\n\n" unless msg == :queue_empty end examples/simple_topic_09.rb @@ -7,12 +7,13 @@ end module Bunny - class ProtocolError < StandardError; end - class ServerDownError < StandardError; end class ConnectionError < StandardError; end - class MessageError < StandardError; end - class ForcedConnectionCloseError < StandardError; end class ForcedChannelCloseError < StandardError; end + class ForcedConnectionCloseError < StandardError; end + class MessageError < StandardError; end + class ProtocolError < StandardError; end + class ServerDownError < StandardError; end + class UnsubscribeError < StandardError; end VERSION = '0.5.4' @@ -62,6 +63,7 @@ module Bunny require 'bunny/exchange08' require 'bunny/queue08' require 'bunny/channel08' + require 'bunny/subscription08' @client = Bunny::Client.new(opts) else @@ -71,6 +73,7 @@ module Bunny require 'bunny/exchange09' require 'bunny/queue09' require 'bunny/channel09' + require 'bunny/subscription09' @client = Bunny::Client09.new(opts) end lib/bunny.rb @@ -18,6 +18,7 @@ Queues must be attached to at least one exchange in order to receive messages fr @client = client @opts = opts @delivery_tag = nil + @subscription = nil # Queues without a given name are named by the server and are generally # bound to the process that created them. @@ -65,7 +66,7 @@ ask to confirm a single message or a set of messages up to and including a speci def ack(opts={}) # If delivery tag is nil then set it to 1 to prevent errors - self.delivery_tag = 1 if self.delivery_tag.nil? + self.delivery_tag = opts[:delivery_tag] || 1 client.send_frame( Qrack::Protocol::Basic::Ack.new({:delivery_tag => delivery_tag, :multiple => false}.merge(opts)) @@ -164,9 +165,7 @@ from queues if successful. If an error occurs raises _Bunny_::_ProtocolError_. Gets a message from a queue in a synchronous way. If error occurs, raises _Bunny_::_ProtocolError_. ==== OPTIONS: - -* <tt>:header => true or false (_default_)</tt> - If set to _true_, - hash <tt>{:header, :delivery_details, :payload}</tt> is returned. + * <tt>:no_ack => true (_default_) or false</tt> - If set to _true_, the server does not expect an acknowledgement message from the client. If set to _false_, the server expects an acknowledgement message from the client and will re-queue the message if it does not receive one within a time specified @@ -174,17 +173,13 @@ Gets a message from a queue in a synchronous way. If error occurs, raises _Bunny ==== RETURNS: -If <tt>:header => true</tt> returns hash <tt>{:header, :delivery_details, :payload}</tt>. <tt>:delivery_details</tt> is -a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count}</tt>. If -<tt>:header => false</tt> only the message payload is returned. +Hash <tt>{:header, :delivery_details, :payload}</tt>. <tt>:delivery_details</tt> is +a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count}</tt>. =end def pop(opts = {}) - # do we want the message header? - hdr = opts.delete(:header) - # do we want to have to provide an acknowledgement? ack = opts.delete(:ack) @@ -198,7 +193,7 @@ a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count method = client.next_method if method.is_a?(Qrack::Protocol::Basic::GetEmpty) then - return :queue_empty + return {:header => nil, :payload => :queue_empty, :delivery_details => nil} elsif !method.is_a?(Qrack::Protocol::Basic::GetOk) raise Bunny::ProtocolError, "Error getting message from queue #{name}" end @@ -216,7 +211,7 @@ a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count end # Return message with additional info if requested - hdr ? {:header => header, :payload => msg, :delivery_details => method.arguments} : msg + {:header => header, :payload => msg, :delivery_details => method.arguments} end @@ -270,99 +265,14 @@ Returns hash {:message_count, :consumer_count}. {:message_count => method.message_count, :consumer_count => method.consumer_count} end -=begin rdoc - -=== DESCRIPTION: - -Asks the server to start a "consumer", which is a transient request for messages from a specific -queue. Consumers last as long as the channel they were created on, or until the client cancels them -with an _unsubscribe_. Every time a message reaches the queue it is passed to the _blk_ for -processing. If error occurs, _Bunny_::_ProtocolError_ is raised. - -==== OPTIONS: -* <tt>:header => true or false (_default_)</tt> - If set to _true_, hash is delivered for each message - <tt>{:header, :delivery_details, :payload}</tt>. -* <tt>:consumer_tag => '_tag_'</tt> - Specifies the identifier for the consumer. The consumer tag is - local to a connection, so two clients can use the same consumer tags. If this field is empty the - queue name is used. -* <tt>:no_ack=> true (_default_) or false</tt> - If set to _true_, the server does not expect an - acknowledgement message from the client. If set to _false_, the server expects an acknowledgement - message from the client and will re-queue the message if it does not receive one within a time specified - by the server. -* <tt>:exclusive => true or false (_default_)</tt> - Request exclusive consumer access, meaning - only this consumer can access the queue. -* <tt>:nowait => true or false (_default_)</tt> - Ignored by Bunny, always _false_. -* <tt>:timeout => number of seconds - The subscribe loop will continue to wait for - messages until terminated (Ctrl-C or kill command) or this timeout interval is reached. -* <tt>:message_max => max number messages to process</tt> - When the required number of messages - is processed subscribe loop is exited. - -==== RETURNS: - -If <tt>:header => true</tt> returns hash <tt>{:header, :delivery_details, :payload}</tt> for each message. -<tt>:delivery_details</tt> is a hash <tt>{:consumer_tag, :delivery_tag, :redelivered, :exchange, :routing_key}</tt>. -If <tt>:header => false</tt> only message payload is returned. -If <tt>:timeout => > 0</tt> is reached Qrack::ClientTimeout is raised - -=end - def subscribe(opts = {}, &blk) - # Get maximum amount of messages to process - message_max = opts[:message_max] || nil - return if message_max == 0 - - # If a consumer tag is not passed in the server will generate one - consumer_tag = opts[:consumer_tag] || nil - - # ignore the :nowait option if passed, otherwise program will hang waiting for a - # response from the server causing an error. - opts.delete(:nowait) - - # do we want the message header? - hdr = opts.delete(:header) - - # do we want to have to provide an acknowledgement? - ack = opts.delete(:ack) - - client.send_frame( - Qrack::Protocol::Basic::Consume.new({ :queue => name, - :consumer_tag => consumer_tag, - :no_ack => !ack, - :nowait => false }.merge(opts)) - ) - - raise Bunny::ProtocolError, - "Error subscribing to queue #{name}" unless - client.next_method.is_a?(Qrack::Protocol::Basic::ConsumeOk) - - # Initialize message counter - counter = 0 - - loop do - method = client.next_method(:timeout => opts[:timeout]) - - # get delivery tag to use for acknowledge - self.delivery_tag = method.delivery_tag if ack - - header = client.next_payload - - # If maximum frame size is smaller than message payload body then message - # will have a message header and several message bodies - msg = '' - while msg.length < header.size - msg += client.next_payload - end - - # pass the message and related info, if requested, to the block for processing - blk.call(hdr ? {:header => header, :payload => msg, :delivery_details => method.arguments} : msg) - - # Increment message counter - counter += 1 - - # Exit loop if message_max condition met - break if !message_max.nil? and counter == message_max - end + def subscribe(opts = {}, &blk) + # Create subscription + s = Bunny::Subscription.new(client, self, opts) + s.start(&blk) + # Reset when subscription finished + @subscription = nil end =begin rdoc @@ -384,20 +294,27 @@ the server will not send any more messages for that consumer. =end def unsubscribe(opts = {}) - consumer_tag = opts[:consumer_tag] || name + # Default consumer_tag from subscription if not passed in + consumer_tag = subscription ? subscription.consumer_tag : opts[:consumer_tag] - # ignore the :nowait option if passed, otherwise program will hang waiting for a - # response from the server causing an error - opts.delete(:nowait) + # Must have consumer tag to tell server what to unsubscribe + raise Bunny::UnsubscribeError, + "No consumer tag received" if !consumer_tag - client.send_frame( Qrack::Protocol::Basic::Cancel.new({ :consumer_tag => consumer_tag }.merge(opts))) + # Cancel consumer + client.send_frame( Qrack::Protocol::Basic::Cancel.new(:consumer_tag => consumer_tag, + :nowait => false)) + + raise Bunny::UnsubscribeError, + "Error unsubscribing from queue #{name}" unless + client.next_method.is_a?(Qrack::Protocol::Basic::CancelOk) - raise Bunny::ProtocolError, - "Error unsubscribing from queue #{name}" unless - client.next_method.is_a?(Qrack::Protocol::Basic::CancelOk) + # Reset subscription + @subscription = nil - # return confirmation + # Return confirmation :unsubscribe_ok + end =begin rdoc lib/bunny/queue08.rb @@ -166,9 +166,7 @@ from queues if successful. If an error occurs raises _Bunny_::_ProtocolError_. Gets a message from a queue in a synchronous way. If error occurs, raises _Bunny_::_ProtocolError_. ==== OPTIONS: - -* <tt>:header => true or false (_default_)</tt> - If set to _true_, - hash <tt>{:header, :delivery_details, :payload}</tt> is returned. + * <tt>:no_ack => true (_default_) or false</tt> - If set to _true_, the server does not expect an acknowledgement message from the client. If set to _false_, the server expects an acknowledgement message from the client and will re-queue the message if it does not receive one within a time specified @@ -176,17 +174,13 @@ Gets a message from a queue in a synchronous way. If error occurs, raises _Bunny ==== RETURNS: -If <tt>:header => true</tt> returns hash <tt>{:header, :delivery_details, :payload}</tt>. <tt>:delivery_details</tt> is -a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count}</tt>. If -<tt>:header => false</tt> only the message payload is returned. +Hash <tt>{:header, :delivery_details, :payload}</tt>. <tt>:delivery_details</tt> is +a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count}</tt>. =end def pop(opts = {}) - # do we want the message header? - hdr = opts.delete(:header) - # do we want to have to provide an acknowledgement? ack = opts.delete(:ack) @@ -201,7 +195,7 @@ a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count method = client.next_method if method.is_a?(Qrack::Protocol09::Basic::GetEmpty) then - return :queue_empty + return {:header => nil, :payload => :queue_empty, :delivery_details => nil} elsif !method.is_a?(Qrack::Protocol09::Basic::GetOk) raise Bunny::ProtocolError, "Error getting message from queue #{name}" end @@ -219,7 +213,7 @@ a hash <tt>{:delivery_tag, :redelivered, :exchange, :routing_key, :message_count end # Return message with additional info if requested - hdr ? {:header => header, :payload => msg, :delivery_details => method.arguments} : msg + {:header => header, :payload => msg, :delivery_details => method.arguments} end @@ -273,100 +267,13 @@ Returns hash {:message_count, :consumer_count}. {:message_count => method.message_count, :consumer_count => method.consumer_count} end -=begin rdoc - -=== DESCRIPTION: - -Asks the server to start a "consumer", which is a transient request for messages from a specific -queue. Consumers last as long as the channel they were created on, or until the client cancels them -with an _unsubscribe_. Every time a message reaches the queue it is passed to the _blk_ for -processing. If error occurs, _Bunny_::_ProtocolError_ is raised. - -==== OPTIONS: -* <tt>:header => true or false (_default_)</tt> - If set to _true_, hash is delivered for each message - <tt>{:header, :delivery_details, :payload}</tt>. -* <tt>:consumer_tag => '_tag_'</tt> - Specifies the identifier for the consumer. The consumer tag is - local to a connection, so two clients can use the same consumer tags. If this field is empty the - queue name is used. -* <tt>:no_ack=> true (_default_) or false</tt> - If set to _true_, the server does not expect an - acknowledgement message from the client. If set to _false_, the server expects an acknowledgement - message from the client and will re-queue the message if it does not receive one within a time specified - by the server. -* <tt>:exclusive => true or false (_default_)</tt> - Request exclusive consumer access, meaning - only this consumer can access the queue. -* <tt>:nowait => true or false (_default_)</tt> - Ignored by Bunny, always _false_. -* <tt>:timeout => number of seconds - The subscribe loop will continue to wait for - messages until terminated (Ctrl-C or kill command) or this timeout interval is reached. -* <tt>:message_max => max number messages to process</tt> - When the required number of messages - is processed subscribe loop is exited. - -==== RETURNS: - -If <tt>:header => true</tt> returns hash <tt>{:header, :delivery_details, :payload}</tt> for each message. -<tt>:delivery_details</tt> is a hash <tt>{:consumer_tag, :delivery_tag, :redelivered, :exchange, :routing_key}</tt>. -If <tt>:header => false</tt> only message payload is returned. -If <tt>:timeout => > 0</tt> is reached Qrack::ClientTimeout is raised - -=end - def subscribe(opts = {}, &blk) - # Get maximum amount of messages to process - message_max = opts[:message_max] || nil - return if message_max == 0 - - # If a consumer tag is not passed in the server will generate one - consumer_tag = opts[:consumer_tag] || nil - - # ignore the :nowait option if passed, otherwise program will hang waiting for a - # response from the server causing an error. - opts.delete(:nowait) - - # do we want the message header? - hdr = opts.delete(:header) - - # do we want to have to provide an acknowledgement? - ack = opts.delete(:ack) - - client.send_frame( - Qrack::Protocol09::Basic::Consume.new({ :reserved_1 => 0, - :queue => name, - :consumer_tag => consumer_tag, - :no_ack => !ack, - :nowait => false }.merge(opts)) - ) - - raise Bunny::ProtocolError, - "Error subscribing to queue #{name}" unless - client.next_method.is_a?(Qrack::Protocol09::Basic::ConsumeOk) - - # Initialize message counter - counter = 0 - - loop do - method = client.next_method(:timeout => opts[:timeout]) - - # get delivery tag to use for acknowledge - self.delivery_tag = method.delivery_tag if ack - - header = client.next_payload - - # If maximum frame size is smaller than message payload body then message - # will have a message header and several message bodies - msg = '' - while msg.length < header.size - msg += client.next_payload - end - - # pass the message and related info, if requested, to the block for processing - blk.call(hdr ? {:header => header, :payload => msg, :delivery_details => method.arguments} : msg) - - # Increment message counter - counter += 1 - - # Exit loop if message_max condition met - break if !message_max.nil? and counter == message_max - end + # Create subscription + s = Bunny::Subscription09.new(client, self, opts) + s.start(&blk) + # Reset when subscription finished + @subscription = nil end =begin rdoc @@ -429,19 +336,25 @@ the server will not send any more messages for that consumer. =end def unsubscribe(opts = {}) - consumer_tag = opts[:consumer_tag] || name + # Default consumer_tag from subscription if not passed in + consumer_tag = subscription ? subscription.consumer_tag : opts[:consumer_tag] - # ignore the :nowait option if passed, otherwise program will hang waiting for a - # response from the server causing an error - opts.delete(:nowait) + # Must have consumer tag to tell server what to unsubscribe + raise Bunny::UnsubscribeError, + "No consumer tag received" if !consumer_tag - client.send_frame( Qrack::Protocol09::Basic::Cancel.new({ :consumer_tag => consumer_tag }.merge(opts))) + # Cancel consumer + client.send_frame( Qrack::Protocol09::Basic::Cancel.new(:consumer_tag => consumer_tag, + :nowait => false)) - raise Bunny::ProtocolError, - "Error unsubscribing from queue #{name}" unless - client.next_method.is_a?(Qrack::Protocol09::Basic::CancelOk) + raise Bunny::UnsubscribeError, + "Error unsubscribing from queue #{name}" unless + client.next_method.is_a?(Qrack::Protocol09::Basic::CancelOk) + + # Reset subscription + @subscription = nil - # return confirmation + # Return confirmation :unsubscribe_ok end lib/bunny/queue09.rb @@ -9,6 +9,7 @@ require 'transport/frame08' require 'qrack/client' require 'qrack/channel' require 'qrack/queue' +require 'qrack/subscription' module Qrack lib/qrack/qrack08.rb @@ -9,6 +9,7 @@ require 'transport/frame09' require 'qrack/client' require 'qrack/channel' require 'qrack/queue' +require 'qrack/subscription' module Qrack lib/qrack/qrack09.rb @@ -4,7 +4,7 @@ module Qrack class Queue attr_reader :name, :client - attr_accessor :delivery_tag + attr_accessor :delivery_tag, :subscription =begin rdoc lib/qrack/queue.rb @@ -63,7 +63,7 @@ describe Bunny do it "should be able to pop a message complete with header and delivery details" do q = @b.queue('test1') - msg = q.pop(:header => true) + msg = q.pop() msg.should be_an_instance_of(Hash) msg[:header].should be_an_instance_of(Bunny::Protocol::Header) msg[:payload].should == 'This is a test message' @@ -74,7 +74,7 @@ describe Bunny do it "should be able to pop a message and just get the payload" do q = @b.queue('test1') q.publish('This is another test message') - msg = q.pop + msg = q.pop[:payload] msg.should == 'This is another test message' q.message_count.should == 0 end @@ -83,7 +83,7 @@ describe Bunny do q = @b.queue('test1') lg_msg = 'z' * 142000 q.publish(lg_msg) - msg = q.pop + msg = q.pop[:payload] msg.should == lg_msg end @@ -105,7 +105,7 @@ describe Bunny do q = @b.queue('test1') q.publish('This is another test message') q.pop - msg = q.pop + msg = q.pop[:payload] msg.should == :queue_empty end @@ -113,9 +113,8 @@ describe Bunny do q = @b.queue('test1') 5.times {q.publish('Yet another test message')} q.message_count.should == 5 - q.subscribe(:message_max => 0){|msg| x = 1} + q.subscribe(:message_max => 0) q.message_count.should == 5 - q.unsubscribe.should == :unsubscribe_ok q.purge.should == :purge_ok end @@ -123,16 +122,60 @@ describe Bunny do q = @b.queue('test1') 5.times {q.publish('Yet another test message')} q.message_count.should == 5 - q.subscribe(:message_max => 5){|msg| x = 1} - q.unsubscribe.should == :unsubscribe_ok + q.subscribe(:message_max => 5) end - + + it "should stop subscription after processing message_max messages < total in queue" do + q = @b.queue('test1') + @b.qos() + 10.times {q.publish('Yet another test message')} + q.message_count.should == 10 + q.subscribe(:message_max => 5, :ack => true) + q.message_count.should == 5 + q.purge.should == :purge_ok + end + it "should raise an error when delete fails" do q = @b.queue('test1') lambda {q.delete(:queue => 'bogus')}.should raise_error(Bunny::ForcedChannelCloseError) @b.channel.active.should == false end + it "should pass correct block parameters through on subscribe" do + q = @b.queue('test1') + q.publish("messages pop\'n") + + q.subscribe do |msg| + msg[:header].should be_an_instance_of Qrack::Protocol::Header + msg[:payload].should == "messages pop'n" + msg[:delivery_details].should_not be_nil + + q.unsubscribe + break + end + + end + + it "should finish processing subscription messages if break is called in block" do + q = @b.queue('test1') + q.publish('messages in my quezen') + + q.subscribe do |msg| + msg[:payload].should == 'messages in my quezen' + q.unsubscribe + break + end + + 5.times {|i| q.publish("#{i}")} + q.subscribe do |msg| + if msg[:payload] == '4' + q.unsubscribe + break + end + end + + end + it "should be able to be deleted" do q = @b.queue('test1') res = q.delete spec/spec_08/queue_spec.rb @@ -31,7 +31,7 @@ describe Bunny do @b.channel.active.should == false end - it "should be able to bind to an exchange" do + it "should be able to bind to an existing exchange" do exch = @b.exchange('direct_exch') q = @b.queue('test1') q.bind(exch).should == :bind_ok @@ -63,7 +63,7 @@ describe Bunny do it "should be able to pop a message complete with header and delivery details" do q = @b.queue('test1') - msg = q.pop(:header => true) + msg = q.pop() msg.should be_an_instance_of(Hash) msg[:header].should be_an_instance_of(Bunny::Protocol09::Header) msg[:payload].should == 'This is a test message' @@ -74,7 +74,7 @@ describe Bunny do it "should be able to pop a message and just get the payload" do q = @b.queue('test1') q.publish('This is another test message') - msg = q.pop + msg = q.pop[:payload] msg.should == 'This is another test message' q.message_count.should == 0 end @@ -83,7 +83,7 @@ describe Bunny do q = @b.queue('test1') lg_msg = 'z' * 142000 q.publish(lg_msg) - msg = q.pop + msg = q.pop[:payload] msg.should == lg_msg end @@ -105,17 +105,16 @@ describe Bunny do q = @b.queue('test1') q.publish('This is another test message') q.pop - msg = q.pop + msg = q.pop[:payload] msg.should == :queue_empty end - + it "should stop subscription without processing messages if max specified is 0" do q = @b.queue('test1') 5.times {q.publish('Yet another test message')} q.message_count.should == 5 - q.subscribe(:message_max => 0){|msg| x = 1} + q.subscribe(:message_max => 0) q.message_count.should == 5 - q.unsubscribe.should == :unsubscribe_ok q.purge.should == :purge_ok end @@ -123,8 +122,17 @@ describe Bunny do q = @b.queue('test1') 5.times {q.publish('Yet another test message')} q.message_count.should == 5 - q.subscribe(:message_max => 5){|msg| x = 1} - q.unsubscribe.should == :unsubscribe_ok + q.subscribe(:message_max => 5) + end + + it "should stop subscription after processing message_max messages < total in queue" do + q = @b.queue('test1') + @b.qos() + 10.times {q.publish('Yet another test message')} + q.message_count.should == 10 + q.subscribe(:message_max => 5, :ack => true) + q.message_count.should == 5 + q.purge.should == :purge_ok end it "should raise an error when delete fails" do @@ -133,24 +141,58 @@ describe Bunny do @b.channel.active.should == false end + it "should pass correct block parameters through on subscribe" do + q = @b.queue('test1') + q.publish("messages pop\'n") + + q.subscribe do |msg| + msg[:header].should be_an_instance_of Qrack::Protocol09::Header + msg[:payload].should == "messages pop'n" + msg[:delivery_details].should_not be_nil + + q.unsubscribe + break + end + + end + + it "should finish processing subscription messages if break is called in block" do + q = @b.queue('test1') + q.publish('messages in my quezen') + + q.subscribe do |msg| + msg[:payload].should == 'messages in my quezen' + q.unsubscribe + break + end + + 5.times {|i| q.publish("#{i}")} + q.subscribe do |msg| + if msg[:payload] == '4' + q.unsubscribe + break + end + end + end + it "should be able to be deleted" do q = @b.queue('test1') res = q.delete res.should == :delete_ok @b.queues.has_key?('test1').should be(false) end - + it "should ignore the :nowait option when deleted" do q = @b.queue('test0') q.delete(:nowait => true) end - it "should support server named queues" do - q = @b.queue - q.name.should_not == nil + it "should support server named queues" do + q = @b.queue + q.name.should_not == nil - @b.queue(q.name).should == q - q.delete - end + @b.queue(q.name).should == q + q.delete + end end spec/spec_09/queue_spec.rb b8b39af3359000e2cdf82ee05c857c7a996a08ab Chris Duncan celldee@gmail.com http://github.com/celldee/bunny/commit/5166c1e1e5bc367e5b21da1ca0fb8d6184038dd4 5166c1e1e5bc367e5b21da1ca0fb8d6184038dd4 2009-09-13T17:38:24-07:00 2009-09-13T17:38:24-07:00 Add Subscription classes and improve Queue#pop/subscribe/unsubscribe. Update examples, gemspec and test specs. 7b8da8ac0126ded6659e48a72c16de166f5d1dcb Chris Duncan celldee@gmail.com