@@ -30,4 +30,10 @@ rescue LoadError end desc "Default task runs specs" -task :default => [:spec] \ No newline at end of file +task :default => [:spec] + +desc 'Removes trailing whitespace' +task :whitespace do + sh %{find . -name '*.rb' -exec sed -i '' 's/ *$//g' {} \\;} +end + Rakefile @@ -14,7 +14,7 @@ module RosettaQueue Destinations.define do |dest| dest.map :foo, "queue.foo" - end + end class MessageHandlerFoo include RosettaQueue::MessageHandler @@ -31,7 +31,7 @@ module RosettaQueue # consumer = RosettaQueue::Consumer.new(MessageHandlerFoo.new) # # Thread.new(consumer) do |cons| # consumer.receive -# # end +# # end # puts "sleeping for 10" # sleep 10 # puts "shutting consumer down" @@ -40,6 +40,6 @@ module RosettaQueue ThreadedManager.create do |m| m.add MessageHandlerFoo.new m.start - end + end end examples/sample_amqp_consumer.rb @@ -14,7 +14,7 @@ module RosettaQueue Destinations.define do |dest| dest.map :foo, "fanout.foo" - end + end class MessageHandlerFoo include RosettaQueue::MessageHandler @@ -47,6 +47,6 @@ module RosettaQueue m.add MessageHandlerFoo.new m.add MessageHandlerBar.new m.start - end + end end examples/sample_amqp_fanout_consumer.rb @@ -11,7 +11,7 @@ end RosettaQueue::Destinations.define do |dest| dest.map :foo, "fanout.foo" -end +end RosettaQueue::Producer.publish(:foo, "hello there") examples/sample_amqp_fanout_producer.rb @@ -11,6 +11,6 @@ end RosettaQueue::Destinations.define do |dest| dest.map :foo, "queue.foo" -end +end RosettaQueue::Producer.publish(:foo, "hello there") examples/sample_amqp_producer.rb @@ -7,7 +7,7 @@ end When /^the message on '(.+)' is consumed$/ do |queue_name| # TODO - # @consumed_message = queue(:foo_queue).pop + # @consumed_message = queue(:foo_queue).pop @consumed_message = consume_once(queue_name.to_sym) end features/step_definitions/filtering_steps.rb @@ -6,11 +6,11 @@ Given /^a consumer is listening to queue '(.*)'$/ do |queue| when /evented/ EM.run do RosettaQueue::Consumer.new(cons).receive - end + end else RosettaQueue::Consumer.new(cons).receive - end - end + end + end end Then /^the message should be consumed from '(.*)'$/ do |queue| features/step_definitions/point_to_point_steps.rb @@ -4,15 +4,15 @@ Given /^multiple consumers are listening to queue '(.*)'$/ do |queue| @managers << RosettaQueue::ThreadedManager.create do |m| m.add(eval_consumer_class(queue, "fooconsumer.log", "FooConsumer").new) m.start - end - end + end + end Thread.new do @managers << RosettaQueue::ThreadedManager.create do |m| m.add(eval_consumer_class(queue, "barconsumer.log", "BarConsumer").new) m.start - end - end + end + end sleep 5 end @@ -20,6 +20,6 @@ Then /^multiple messages should be consumed from '(\w+)'$/ do |queue| ["FooConsumer", "BarConsumer"].each do |class_name, value| file_path = "#{CONSUMER_LOG_DIR}/#{class_name.downcase}.log" File.readlines(file_path).last.should =~ /Hello World! from #{class_name}/ - end + end @managers.each {|m| m.stop_threads } end features/step_definitions/publish_subscribe_steps.rb @@ -4,14 +4,14 @@ class SampleConsumer include ::RosettaQueue::MessageHandler subscribes_to :foo options :durable => true - + attr_reader :msg - + def on_message(msg) @msg = msg puts "MESSAGE #{msg}" end - + end @@ -19,11 +19,11 @@ class SampleConsumerTwo include ::RosettaQueue::MessageHandler subscribes_to :foo options :durable => true - + attr_reader :msg - + def on_message(msg) @msg = msg end - + end features/support/sample_consumers.rb @@ -5,15 +5,15 @@ module RosettaQueue class << self attr_writer :user, :password, :host, :port, :options - + def define yield self end - + def reset @user, @password, @host, @port, @options, @adapter_class = nil, nil, nil, nil, nil, nil end - + def type=(adapter_prefix) require "rosetta_queue/adapters/#{adapter_prefix}" @adapter_class = RosettaQueue::Gateway.const_get("#{adapter_prefix.to_s.classify}Adapter") @@ -28,12 +28,12 @@ module RosettaQueue end private - + def opts raise AdapterException, "Adapter options should be a hash" unless @options.nil? || @options.is_a?(Hash) @options ||= {} end - + end - end + end end lib/rosetta_queue/adapter.rb @@ -7,7 +7,7 @@ module RosettaQueue raise AdapterException, "Unable to discover fanout exchange. Cannot bind queue to exchange!" unless fanout_name fanout_name end - end + end class Amqp < BaseAdapter @@ -18,7 +18,7 @@ module RosettaQueue def delete(destination, opts={}) exchange_strategy_for(destination, opts).delete(destination) - end + end def disconnect(message_handler) destination = destination_for(message_handler) lib/rosetta_queue/adapters/amqp.rb @@ -4,7 +4,7 @@ require File.expand_path(File.dirname(__FILE__) + "/amqp.rb") module RosettaQueue module Gateway - + class AmqpEventedAdapter < Amqp private @@ -21,35 +21,35 @@ module RosettaQueue end end end - + module EventedExchange class BaseExchange - + def initialize(adapter_settings, options={}) @adapter_settings, @options = adapter_settings, options end def delete(destination) conn.queue(destination).delete(@options) - end + end protected - + def channel @channel ||= MQ.new(conn) end - + def conn - vhost = @adapter_settings[:opts][:vhost] || "/" - @conn ||= AMQP.connect(:user => @adapter_settings[:user], - :pass => @adapter_settings[:password], - :host => @adapter_settings[:host], + vhost = @adapter_settings[:opts][:vhost] || "/" + @conn ||= AMQP.connect(:user => @adapter_settings[:user], + :pass => @adapter_settings[:password], + :host => @adapter_settings[:host], :vhost => vhost) end end - - + + class DirectExchange < BaseExchange def publish(destination, message, options={}) @@ -60,7 +60,7 @@ module RosettaQueue RosettaQueue.logger.info("Publishing to #{destination} :: #{message}") queue.unsubscribe end - + def receive(destination, message_handler) raise AdapterException, "Consumers need to run in an EventMachine 'run' block. Try wrapping them inside the evented consumer manager." unless EM.reactor_running? @@ -75,7 +75,7 @@ module RosettaQueue def receive_once(destination, options={}) raise AdapterException, "Consumers need to run in an EventMachine 'run' block. (e.g., EM.run { RosettaQueue::Consumer.receive }" unless EM.reactor_running? - + queue = channel.queue(destination, @options) ack = @options[:ack] queue.pop(@options) do |header, msg| @@ -85,8 +85,8 @@ module RosettaQueue end end end - - + + class FanoutExchange < BaseExchange include Fanout @@ -97,36 +97,36 @@ module RosettaQueue exchange.publish(msg, opts) RosettaQueue.logger.info("Publishing to fanout #{dest} :: #{msg}") end - + def receive(destination, message_handler) raise AdapterException, "Consumers need to run in an EventMachine 'run' block. Try wrapping them inside the evented consumer manager." unless EM.reactor_running? queue = channel.queue("queue_#{self.object_id}") exchange = channel.fanout(fanout_name_for(destination), @options) ack = @options[:ack] - + queue.bind(exchange).subscribe(@options) do |header, msg| RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") message_handler.on_message(Filters.process_receiving(msg)) header.ack if ack end end - + def receive_once(destination, opts={}) raise AdapterException, "Consumers need to run in an EventMachine 'run' block. (e.g., EM.run { RosettaQueue::Consumer.receive }" unless EM.reactor_running? queue = channel.queue("queue_#{self.object_id}") exchange = channel.fanout(fanout_name_for(destination), opts) ack = @options[:ack] - + queue.bind(exchange).pop(opts) do |header, msg| RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") header.ack if ack yield Filters.process_receiving(msg) end end - + end end - end + end end lib/rosetta_queue/adapters/amqp_evented.rb @@ -21,7 +21,7 @@ module RosettaQueue @exchange ||= SynchExchange::DirectExchange.new(@adapter_settings, options) end end - end + end module SynchExchange @@ -33,7 +33,7 @@ module RosettaQueue def delete(destination, options={}) conn.queue(destination).delete(options) - end + end def unsubscribe @queue.unsubscribe @@ -43,10 +43,10 @@ module RosettaQueue protected def conn - vhost = @adapter_settings[:opts][:vhost] || "/" - @conn ||= Bunny.new( :user => @adapter_settings[:user], - :pass => @adapter_settings[:password], - :host => @adapter_settings[:host], + vhost = @adapter_settings[:opts][:vhost] || "/" + @conn ||= Bunny.new( :user => @adapter_settings[:user], + :pass => @adapter_settings[:password], + :host => @adapter_settings[:host], :vhost => vhost) @conn.start unless @conn.status == :connected @conn @@ -61,7 +61,7 @@ module RosettaQueue queue = conn.queue(destination, options) queue.publish(message, options) conn.stop - end + end def receive(destination, message_handler) ack = @options[:ack] @@ -70,7 +70,7 @@ module RosettaQueue RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") message_handler.on_message(Filters.process_receiving(msg)) @queue.ack if ack - end + end end def receive_once(destination, options = {}) @@ -86,12 +86,12 @@ module RosettaQueue class FanoutExchange < BaseExchange include Fanout - + def publish(destination, message, options={}) exchange = conn.exchange(fanout_name_for(destination), options.merge({:type => :fanout})) exchange.publish(message, options) RosettaQueue.logger.info("Publishing to fanout #{destination} :: #{message}") - end + end def receive(destination, message_handler) ack = @options[:ack] @@ -102,7 +102,7 @@ module RosettaQueue RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") message_handler.on_message(Filters.process_receiving(msg)) @queue.ack if ack - end + end end def receive_once(destination, options={}) @@ -116,8 +116,8 @@ module RosettaQueue yield Filters.process_receiving(msg) end - end + end - end + end end end lib/rosetta_queue/adapters/amqp_synch.rb @@ -2,7 +2,7 @@ module RosettaQueue module Gateway class BaseAdapter - + protected def options_for(message_handler) @@ -13,7 +13,7 @@ module RosettaQueue raise DestinationNotFound.new("Missing destination!") unless message_handler.destination @dest ||= Destinations.lookup(message_handler.destination.to_sym) end - + def filter_receiving(msg) Filters.process_receiving(msg) end @@ -21,7 +21,7 @@ module RosettaQueue def filter_sending(msg) Filters.process_sending(msg) end - + end end end lib/rosetta_queue/adapters/base.rb @@ -2,7 +2,7 @@ require 'beanstalk-client' module RosettaQueue module Gateway - + class BeanstalkAdapter < BaseAdapter def ack(msg) @@ -22,7 +22,7 @@ module RosettaQueue msg.delete msg end - + def receive_once(destination=nil, opts={}) receive.body end @@ -39,14 +39,14 @@ module RosettaQueue message_handler.on_message(filter_receiving(msg)) end end - + def send_message(destination, message, options) - RosettaQueue.logger.info("Publishing to #{destination} :: #{message}") + RosettaQueue.logger.info("Publishing to #{destination} :: #{message}") @conn.put(message) end private - + def running(&block) loop(&block) end lib/rosetta_queue/adapters/beanstalk.rb @@ -1,6 +1,6 @@ module RosettaQueue module Gateway - + class FakeAdapter def initialize @@ -10,17 +10,17 @@ module RosettaQueue def send_message(queue, message, headers) @messages << {'queue' => queue, 'message' => RosettaQueue::Filters::process_receiving(message), 'headers' => headers} end - + def messages_sent_to(queue) (queue ? @messages.select{|message| message['queue'] == queue} : @messages).map{|m| m['message']} end - + def queues @messages.map {|message| message['queue']} end end - + end - + end lib/rosetta_queue/adapters/fake.rb @@ -1,8 +1,8 @@ module RosettaQueue module Gateway - + # The null adapter lets all send messages enter into the ether and so is ideal for modes - # when you do not want to incur the overhead of a real adapter. You can not consume with + # when you do not want to incur the overhead of a real adapter. You can not consume with # this adapter however. # # In your RosettaQueue definition block, and your using rails, you could base your adapter type on Rails.env: @@ -18,11 +18,11 @@ module RosettaQueue # a.type = "null" # end # end - # - # (if you follow this example and are using stories be sure + # + # (if you follow this example and are using stories be sure # to set ENV["RUNNING_STORIES"] = "true" in your helper.rb or env.rb file) class NullAdapter - + def initialize(adapter_settings) # no-op end @@ -30,15 +30,15 @@ module RosettaQueue def disconnect # no-op end - + def receive raise "Null Adpater is in use, you can not consume messages!" end - + def receive_with(message_handler) raise "Null Adpater is in use, you can not consume messages!" end - + def send_message(queue, message, options) # no-op end @@ -46,7 +46,7 @@ module RosettaQueue def subscribe(queue, options) # no-op end - + def unsubscribe(queue) # no-op end lib/rosetta_queue/adapters/null.rb @@ -2,7 +2,7 @@ require 'stomp' module RosettaQueue module Gateway - + class StompAdapter < BaseAdapter def ack(msg) @@ -12,10 +12,10 @@ module RosettaQueue def initialize(adapter_settings = {}) raise AdapterException, "Missing adapter settings" if adapter_settings.empty? - @conn = Stomp::Connection.open(adapter_settings[:user], - adapter_settings[:password], - adapter_settings[:host], - adapter_settings[:port], + @conn = Stomp::Connection.open(adapter_settings[:user], + adapter_settings[:password], + adapter_settings[:host], + adapter_settings[:port], true) end @@ -29,7 +29,7 @@ module RosettaQueue ack(msg) unless options[:ack].nil? msg end - + def receive_once(destination, opts) subscribe(destination, opts) msg = receive(opts).body @@ -51,22 +51,22 @@ module RosettaQueue Thread.current[:processing] = false end end - + def send_message(destination, message, options) - RosettaQueue.logger.info("Publishing to #{destination} :: #{message}") + RosettaQueue.logger.info("Publishing to #{destination} :: #{message}") @conn.send(destination, message, options) end def subscribe(destination, options) @conn.subscribe(destination, options) end - + def unsubscribe(destination) @conn.unsubscribe(destination) end - + private - + def running(&block) loop(&block) end @@ -74,15 +74,15 @@ module RosettaQueue end class StompAdapterProxy - + def initialize(adapter, msg) @adapter, @msg = adapter, msg end def ack @adapter.ack(@msg) - end - end + end + end end end lib/rosetta_queue/adapters/stomp.rb @@ -1,5 +1,5 @@ module RosettaQueue - + class Base def disconnect @@ -12,4 +12,4 @@ module RosettaQueue end end -end \ No newline at end of file +end lib/rosetta_queue/base.rb @@ -3,17 +3,17 @@ module RosettaQueue def self.receive(destination, options = {}) RosettaQueue::Adapter.instance.receive_once(Destinations.lookup(destination), options) - + rescue Exception=>e RosettaQueue.logger.error("Caught exception in Consumer.receive: #{$!}\n" + e.backtrace.join("\n\t")) end def self.delete(destination, options={}) RosettaQueue::Adapter.instance.delete(Destinations.lookup(destination), options) - + rescue Exception=>e RosettaQueue.logger.error("Caught exception in Consumer.delete: #{$!}\n" + e.backtrace.join("\n\t")) - end + end def initialize(message_handler) @message_handler = message_handler @@ -21,7 +21,7 @@ module RosettaQueue def receive connection.receive_with(@message_handler) - + rescue Exception=>e RosettaQueue.logger.error("Caught exception in Consumer#receive: #{$!}\n" + e.backtrace.join("\n\t")) end lib/rosetta_queue/consumer.rb @@ -1,5 +1,5 @@ module RosettaQueue - + class BaseManager attr_reader :consumers @@ -10,15 +10,15 @@ module RosettaQueue manager end end - + def initialize @consumers = {} end - + def add(message_handler) key = message_handler.class.to_s.underscore.to_sym @consumers[key] = Consumer.new(message_handler) end - - end + + end end lib/rosetta_queue/consumer_managers/base.rb @@ -2,9 +2,9 @@ require 'rosetta_queue/consumer_managers/base' require 'mq' module RosettaQueue - + class EventedManager < BaseManager - + def start EM.run { trap_interruptions @@ -19,20 +19,20 @@ module RosettaQueue end } end - + def stop RosettaQueue.logger.info("Shutting down event machine...") EM.stop end private - + def trap_interruptions trap("INT") { RosettaQueue.logger.warn("Interrupt received. Shutting down...") EM.stop } - + trap("TERM") { RosettaQueue.logger.warn("Interrupt received. Shutting down...") EM.stop lib/rosetta_queue/consumer_managers/evented.rb @@ -25,10 +25,10 @@ module RosettaQueue def join_threads @threads.each { |thread| thread.join } end - + def shutdown_requested RosettaQueue.logger.error "Shutdown requested, starting to prune threads..." - + while @threads.any? { |n, t| t.alive? } RosettaQueue.logger.info "Calling stop_threads" stop_threads @@ -73,7 +73,7 @@ module RosettaQueue Thread.pass end end - end + end end def stop_threads lib/rosetta_queue/consumer_managers/threaded.rb @@ -7,11 +7,11 @@ class String self.first.downcase + camelize(self)[1..-1] end end - + def classify camelize(self.sub(/.*\./, '')) end - + def underscore self.gsub(/::/, '/'). gsub(/([A-Z]+)([A-Z][a-z])/,'\1_\2'). @@ -19,4 +19,4 @@ class String tr("-", "_"). downcase end -end \ No newline at end of file +end lib/rosetta_queue/core_ext/string.rb @@ -17,4 +17,4 @@ class Time formatter.respond_to?(:call) ? formatter.call(self).to_s : strftime(formatter) end -end +end lib/rosetta_queue/core_ext/time.rb @@ -1,7 +1,7 @@ module RosettaQueue class Destinations - + @dest = {} class << self @@ -10,25 +10,25 @@ module RosettaQueue def define yield self end - + def clear @dest.clear end - + def lookup(dest_name) mapping = dest[dest_name.to_sym] raise "No destination mapping for '#{dest_name}' has been defined!" unless mapping return mapping end - + def map(key, dest) @dest[key] = dest end - + def queue_names @dest.values end end end -end \ No newline at end of file +end lib/rosetta_queue/destinations.rb @@ -7,36 +7,36 @@ module RosettaQueue class Filters - + class << self - + def define yield self end - + def reset @receiving = nil @sending = nil end - + def receiving(&receiving_filter) @receiving = receiving_filter end - + def sending(&sending_filter) @sending = sending_filter end - + def process_sending(message) return message unless @sending @sending.call(message) end - + def process_receiving(message) return message unless @receiving @receiving.call(message) end - + end end end lib/rosetta_queue/filters.rb @@ -6,7 +6,7 @@ module RosettaQueue def self.logger=(new_logger) @logger = new_logger end - + def self.logger return @logger if @logger raise MissingLogger, "No logger has been set for RosettaQueue. Please define one with RosettaQueue.logger=." lib/rosetta_queue/logger.rb @@ -1,6 +1,6 @@ module RosettaQueue module MessageHandler - + module ClassMethods attr_reader :destination, :options_hash @@ -17,22 +17,22 @@ module RosettaQueue @destination = destination end end - + def self.included(receiver) receiver.extend(ClassMethods) attr_accessor :adapter_proxy def destination - self.class.destination + self.class.destination end - + def options_hash self.class.options_hash end def ack adapter_proxy.ack unless adapter_proxy.nil? - end + end end end lib/rosetta_queue/message_handler.rb @@ -1,12 +1,12 @@ module RosettaQueue - + class Producer < Base include MessageHandler def self.publish(destination, message, options = {}) RosettaQueue::Adapter.instance.send_message(Destinations.lookup(destination), Filters.process_sending(message), options) - rescue Exception=>e + rescue Exception => e RosettaQueue.logger.error("Caught exception in Consumer.publish: #{$!}\n" + e.backtrace.join("\n\t")) end lib/rosetta_queue/producer.rb @@ -1,7 +1,7 @@ class Hash # To be used in conjuction with rspec's predicate matcher. # - # For example, in story/feature or a functional spec you could say: + # For example, in story/feature or a functional spec you could say: # # expected_message = {'name' => 'Advertiser'} # expected_message.should be_published_to(:advertiser_create) @@ -13,9 +13,9 @@ class Hash rescue Timeout::Error raise "#{destination} should have received a message but did not NOTE: make sure there are no other processes which are polling messages" end - + # calling should == is kinda wierd, I know.. but in order to get a decent error message it is needed received_message.should == self end - + end lib/rosetta_queue/spec_helpers/hash.rb @@ -4,18 +4,18 @@ module RosettaQueue # World {|world| world.extend RosettaQueue::SpecHelpers } module SpecHelpers require 'open-uri' - - # *Currently* only works with ActiveMQ being used as gateway. + + # *Currently* only works with ActiveMQ being used as gateway. # This will clear the queues defined in the RosettaQueue::Destinations mapping. # TODO: Figure out a better spot for this to allow for other gateways... def clear_queues - RosettaQueue::Destinations.queue_names.each do |name| + RosettaQueue::Destinations.queue_names.each do |name| queue = name.gsub('/queue/','') open("http://127.0.0.1:8161/admin/deleteDestination.action?JMSDestination=#{queue}&JMSDestinationType=queue") end end - - # Publishes a given hash as json to the specified destination. + + # Publishes a given hash as json to the specified destination. # Example: # publish_message(expected_message, :to => :client_status, :options => {...}) # The :options will be passed to the publisher and are optional. @@ -23,14 +23,14 @@ module RosettaQueue options[:options] ||= {:persistent => false} RosettaQueue::Producer.publish(options[:to], message, options[:options]) end - + # Consumes the first message on queue of consumer that is passed in and uses the consumer to handle it. # Example: # consume_once_with ClientStatusConsumer def consume_once_with(consumer) consumer.new.on_message(RosettaQueue::Consumer.receive(consumer.destination)) end - + # Consumes the first message on queue and returns it. # Example: # message = consume_once :foo_queue @@ -42,6 +42,6 @@ module RosettaQueue sleep 1 Messaging::Consumer.receive(destination, :persistent => false).to_hash_from_json end - + end end lib/rosetta_queue/spec_helpers/helpers.rb @@ -1,6 +1,6 @@ module RosettaQueue module Matchers - + class PublishAMessageTo def initialize(expected_queue_name, options=nil) @@ -10,14 +10,14 @@ module RosettaQueue @expected_queue = expected_queue_name.is_a?(Symbol) ? RosettaQueue::Destinations.lookup(expected_queue_name) : expected_queue_name end - def matches?(lambda_to_run) + def matches?(lambda_to_run) #given RosettaQueue::Adapter.stub!(:instance).and_return(fake_adapter = RosettaQueue::Gateway::FakeAdapter.new) #when lambda_to_run.call #then - @actual_queues = fake_adapter.queues - @number_of_messages_published = @actual_queues.select{ |q| q == @expected_queue}.size + @actual_queues = fake_adapter.queues + @number_of_messages_published = @actual_queues.select{ |q| q == @expected_queue}.size @number_of_messages_published == @how_many_messages_expected end @@ -32,7 +32,7 @@ module RosettaQueue def description "publish #{message_plural} to the '#{@expected_queue_name}' queue" end - + private def message_plural @how_many_messages_expected == 1 ? "a message" : "#{@how_many_messages_expected} messages" @@ -42,16 +42,16 @@ module RosettaQueue def publish_a_message_to(expected_queue) PublishAMessageTo.new(expected_queue) end - + alias :publish_message_to :publish_a_message_to - + def publish_messages_to(expected_queue, options) PublishAMessageTo.new(expected_queue, options) end - + class PublishMessageMatcher - - + + def matches?(lambda_to_run) #given RosettaQueue::Adapter.stub!(:instance).and_return(fake_adapter = RosettaQueue::Gateway::FakeAdapter.new) @@ -61,30 +61,30 @@ module RosettaQueue message = fake_adapter.messages_sent_to(@expected_queue).first || '' @actual_message = message end - + protected def extract_options(options) - if (expected_queue_name = options[:to]) + if (expected_queue_name = options[:to]) @expected_queue = expected_queue_name.is_a?(Symbol) ? RosettaQueue::Destinations.lookup(expected_queue_name) : expected_queue_name end end end - + class PublishMessageWith < PublishMessageMatcher - + def initialize(message_subset, options) @message_subset = message_subset extract_options(options) end - def matches?(lambda_to_run) + def matches?(lambda_to_run) super Spec::Mocks::ArgumentConstraints::HashIncludingConstraint.new(@message_subset) == @actual_message end def failure_message if @actual_message.blank? - "expected #{@message_subset.inspect} to be contained in a message but no message was published" + "expected #{@message_subset.inspect} to be contained in a message but no message was published" else "expected #{@message_subset.inspect} to be contained in the message: #{@actual_message.inspect}" end @@ -97,29 +97,29 @@ module RosettaQueue def description "publish a message with #{@message_subset.inspect}" end - + end - + def publish_message_with(message_subset, options={}) PublishMessageWith.new(message_subset, options) end - - + + class PublishMessage < PublishMessageMatcher - + def initialize(expected_message, options) @expected_message = expected_message extract_options(options) end - def matches?(lambda_to_run) + def matches?(lambda_to_run) super @actual_message == @expected_message end def failure_message if @actual_message.blank? - "expected #{@expected_message.inspect} to be published but no message was" + "expected #{@expected_message.inspect} to be published but no message was" else "expected #{@expected_message.inspect} to be published but the following was instead: #{@actual_message.inspect}" end @@ -132,13 +132,13 @@ module RosettaQueue def description "publish the message: #{@expected_message.inspect}" end - + end - + def publish_message(exact_expected_message, options={}) PublishMessage.new(exact_expected_message, options) end - + end end lib/rosetta_queue/spec_helpers/publishing_matchers.rb @@ -1,14 +1,14 @@ require File.dirname(__FILE__) + '/../spec_helper' module RosettaQueue - + describe Adapter do before(:each) do @stomp_adapter = mock("Gateway::StompAdapter") Adapter.reset end - + describe ".reset" do it "should clear all definitions" do Adapter.define { |a| a.type = "null" } @@ -17,38 +17,38 @@ module RosettaQueue running { Adapter.instance }.should raise_error(AdapterException) end end - + describe ".type=" do - + it "should raise error when adapter does not exist" do - running { + running { Adapter.define do |a| a.type = "foo" end }.should raise_error(AdapterException) end - + end - + describe "adapter not type set" do it "should raise an error when .instance is called" do # given Adapter.define { |a| } # then & when - running { Adapter.instance }.should raise_error(AdapterException) + running { Adapter.instance }.should raise_error(AdapterException) end end describe "adapter type set" do - + before(:each) do Adapter.define { |a| a.type = "null" } end - + it "should return adapter instance" do Adapter.instance.class.should == RosettaQueue::Gateway::NullAdapter end - + end describe "adapter instantiation" do @@ -61,41 +61,41 @@ module RosettaQueue a.port = "9000" a.type = "fake" end - end - + end + def do_process Adapter.instance end it "should set opts as an empty has unless variable is set" do - during_process { + during_process { RosettaQueue::Gateway::FakeAdapter.should_receive(:new).with({:user => "foo", :password => "bar", :host => "localhost", :port => "9000", :opts => {}}) } - end + end describe "when setting options" do before(:each) do Adapter.define { |a| a.options = {:vhost => "baz"} } - end - + end + it "should map adapter_settings to a hash" do - during_process { + during_process { RosettaQueue::Gateway::FakeAdapter.should_receive(:new).with({:user => "foo", :password => "bar", :host => "localhost", :port => "9000", :opts => {:vhost => "baz"}}) } - end - end + end + end describe "setting options incorrectly (options should always be set as a Hash)" do - + before(:each) do Adapter.define { |a| a.options = "baz" } - end - + end + it "should raise an adapter exception" do running { Adapter.instance }.should raise_error("Adapter options should be a hash") - end - end - - end + end + end + + end end end spec/rosetta_queue/adapter_spec.rb @@ -259,7 +259,7 @@ module RosettaQueue::Gateway # end # context "#ack" do - + # it "should delegate to AMQP queue object" do # # expect # @queue.should_receive(:ack) @@ -267,7 +267,7 @@ module RosettaQueue::Gateway # # when # @proxy.ack # end - + # end # end spec/rosetta_queue/adapters/amqp_synchronous_spec.rb @@ -23,7 +23,7 @@ module RosettaQueue def do_receiving_once @adapter.receive_once end - + it "should delete messages once received" do when_receiving_once { @msg_obj.should_receive(:delete) @@ -37,11 +37,11 @@ module RosettaQueue end it "should delete message during receive" do - when_receiving { + when_receiving { @msg_obj.should_receive(:delete) } end - end + end end end end spec/rosetta_queue/adapters/beanstalk_spec.rb @@ -3,7 +3,7 @@ require 'rosetta_queue/adapters/fake' module RosettaQueue module Gateway - + describe FakeAdapter do describe "#queues" do @@ -16,7 +16,7 @@ module RosettaQueue adapter.queues.should == ['queue 1', 'queue 2'] end end - + describe "#messages_sent_to" do it "should return the message bodies that were delivered to the specified queue" do @@ -41,7 +41,7 @@ module RosettaQueue # then adapter.messages_sent_to('queue').should == ['Filtered Message'] end - + it "should return all the message's bodies when nil is passed in at the queue" do # given adapter = FakeAdapter.new @@ -52,7 +52,7 @@ module RosettaQueue # then results.should == ['message 1', 'message 2'] end - + it "should return an empty array when no messages have been delivered" do # given adapter = FakeAdapter.new @@ -62,11 +62,11 @@ module RosettaQueue # then results.should == [] end - + end end - + end end spec/rosetta_queue/adapters/fake_spec.rb @@ -6,25 +6,25 @@ module RosettaQueue describe NullAdapter do - + def null_adapter NullAdapter.new({:user => 'user', :password => 'password', :host => 'host', :port => 'port'}) end - + %w[disconnect receive receive_with send_message subscribe unsubscribe].each do |adapter_method| it "should respond to ##{adapter_method}" do null_adapter.should respond_to(adapter_method) end end - + it "should raise an error when #receive is called" do running { null_adapter.receive }.should raise_error end - + it "should raise an error when #receive_with is called" do running { null_adapter.receive_with('consumer') }.should raise_error end - + end end spec/rosetta_queue/adapters/null_spec.rb @@ -14,7 +14,7 @@ module RosettaQueue def do_receiving_with_handler @adapter.receive_with(@handler) end - + describe "#receive_once" do it "should return the message from the connection" do @@ -22,7 +22,7 @@ module RosettaQueue end end - + describe "#receive_with" do it "should look up the destination defined on the class" do @@ -30,9 +30,9 @@ module RosettaQueue Destinations.should_receive(:lookup).with(:foo).and_return("foo") } end - + end - + end end end spec/rosetta_queue/adapters/shared_adapter_behavior.rb @@ -8,7 +8,7 @@ module RosettaQueue # @channel.should_receive(:fanout).with(@exchange).and_return(@bound_queue) # } # end - + it "should bind to fanout exchange" do when_receiving_exchange { @queue.should_receive(:bind).with(@exchange).and_return(@bound_queue) spec/rosetta_queue/adapters/shared_fanout_behavior.rb @@ -18,7 +18,7 @@ module RosettaQueue end it_should_behave_like "an adapter" - + describe "#send_message" do it "should delegate to the connection" do # need this hack since the stomp client overrides #send @@ -26,9 +26,9 @@ module RosettaQueue @args = args end def @conn.sent_args ; @args end - + after_publishing { - @conn.sent_args.should == ['queue', 'message', 'options'] + @conn.sent_args.should == ['queue', 'message', 'options'] } end end @@ -38,16 +38,16 @@ module RosettaQueue def do_receiving_once @adapter.receive_once("/queue/foo", {:persistent => false}) end - + it "should subscribe to queue" do - when_receiving_once { - @conn.should_receive("subscribe").with("/queue/foo", {:persistent => false}) + when_receiving_once { + @conn.should_receive("subscribe").with("/queue/foo", {:persistent => false}) } end - + it "should unsubscribe from queue" do - when_receiving_once { - @conn.should_receive("unsubscribe").with("/queue/foo") + when_receiving_once { + @conn.should_receive("unsubscribe").with("/queue/foo") } end end @@ -64,8 +64,8 @@ module RosettaQueue when_receiving_with_handler { @conn.should_receive(:ack) } - end - + end + describe "no ack" do before(:each) do @@ -74,15 +74,15 @@ module RosettaQueue it "should not acknowledge client" do when_receiving_with_handler { - @conn.should_not_receive(:ack) + @conn.should_not_receive(:ack) } - end + end - end + end end - + describe "disconnect" do - + def do_disconnecting @adapter.disconnect(@handler) end @@ -98,8 +98,8 @@ module RosettaQueue @conn.should_receive("disconnect") } end - - end + + end describe StompAdapterProxy do @@ -109,7 +109,7 @@ module RosettaQueue end context "#ack" do - + it "should delegate to AMQP queue object" do # expect @adapter.should_receive(:ack).with("foo") @@ -117,7 +117,7 @@ module RosettaQueue # when @proxy.ack end - + end end spec/rosetta_queue/adapters/stomp_spec.rb @@ -7,7 +7,7 @@ module RosettaQueue before(:each) do Stomp::Connection.stub!(:open).and_return(nil) - @consumer = mock("test_consumer_1", :receive => true, + @consumer = mock("test_consumer_1", :receive => true, :connection => mock("AmqpAdapter", :subscribe => true, :unsubscribe => nil, :disconnect => nil), :unsubscribe => true, :disconnect => true) Consumer.stub!(:new).and_return(@consumer) @@ -24,7 +24,7 @@ module RosettaQueue end describe "starting" do - + def do_process @manager.start end @@ -34,7 +34,7 @@ module RosettaQueue @manager.consumers.each_value { |cons| cons.should_receive(:receive) } } end - + end describe "stopping" do @@ -42,15 +42,15 @@ module RosettaQueue def do_process @manager.stop end - + it "should stop consumers" do during_process { EM.should_receive(:stop) } end - + end end end -end \ No newline at end of file +end spec/rosetta_queue/consumer_managers/evented_spec.rb @@ -3,9 +3,9 @@ module RosettaQueue describe "a consumer manager", :shared => true do def do_process - @manager.start + @manager.start end - + describe ".add" do def do_process @@ -23,4 +23,4 @@ module RosettaQueue end end -end \ No newline at end of file +end spec/rosetta_queue/consumer_managers/shared_manager_behavior.rb @@ -19,7 +19,7 @@ module RosettaQueue it_should_behave_like "a consumer manager" describe "threading" do - + before do @manager.stub!(:join_threads) @manager.stub!(:monitor_threads) @@ -31,9 +31,9 @@ module RosettaQueue it "should load subscriptions into threads on start" do during_process {Thread.should_receive(:new).with(:"spec/mocks/mock", @consumer).and_return(@thread)} end - + describe "shutting down" do - + def do_process @manager.start @manager.stop @@ -42,10 +42,10 @@ module RosettaQueue it "should shut threaded subscriptions down on stop" do during_process do @consumer.should_receive(:disconnect) - @thread.should_receive(:kill) + @thread.should_receive(:kill) end end end end end -end \ No newline at end of file +end spec/rosetta_queue/consumer_managers/threaded_spec.rb @@ -2,7 +2,7 @@ require File.dirname(__FILE__) + '/../spec_helper' module RosettaQueue describe Consumer do - + class TestConsumer include MessageHandler @@ -16,14 +16,14 @@ module RosettaQueue before(:each) do @message = mock("message", "headers" => "foo", "body" => "message body") - @adapter = mock("adpater", :subscribe => true, :unsubscribe => true, :disconnect => true, :receive_with => TestConsumer.new, + @adapter = mock("adpater", :subscribe => true, :unsubscribe => true, :disconnect => true, :receive_with => TestConsumer.new, :receive_once => @message.body, :ack => true) Adapter.stub!(:instance).and_return(@adapter) Destinations.stub!(:lookup).and_return("/queue/foo") end - + it_should_behave_like "a messaging gateway object" - + attr_reader :adapter def gateway @gateway ||= Consumer.new(TestConsumer.new) @@ -33,30 +33,30 @@ module RosettaQueue before(:each) do @consumer = Consumer.new( @message_handler = TestConsumer.new) end - + def when_receiving yield if block_given? @consumer.receive end - + it "should pass message handler onto the adpater with #receive" do - when_receiving { + when_receiving { @adapter.should_receive("receive_with").with(@message_handler) } end end - + describe ".delete" do - before(:each) do + before(:each) do @adapter.stub!(:delete) - Destinations.stub!(:lookup).and_return("/queue/foo") - end + Destinations.stub!(:lookup).and_return("/queue/foo") + end it "should look up the destination" do # expect - Destinations.should_receive(:lookup).with(:test_queue_passed_in).and_return("/queue/foo") + Destinations.should_receive(:lookup).with(:test_queue_passed_in).and_return("/queue/foo") # when Consumer.delete(:test_queue_passed_in) @@ -80,7 +80,7 @@ module RosettaQueue it "should look up the destination" do when_receiving { - Destinations.should_receive(:lookup).with(:test_queue_passed_in).and_return("/queue/foo") + Destinations.should_receive(:lookup).with(:test_queue_passed_in).and_return("/queue/foo") } end spec/rosetta_queue/consumer_spec.rb @@ -4,12 +4,12 @@ describe String do describe "#camelize" do it "should have code examples" end - + describe "#classify" do it "should have code examples" end - + describe "#underscore" do it "should have code examples" - end + end end spec/rosetta_queue/core_ext/string_spec.rb @@ -7,28 +7,28 @@ module RosettaQueue before(:each) do Destinations.clear end - + after(:each) do Destinations.clear end - + it "should map destination to hash" do - Destinations.define do |queue| + Destinations.define do |queue| queue.map :test_queue, "/queue/test_queue" end - + Destinations.lookup(:test_queue).should == "/queue/test_queue" end - + it "#queue_names should return an array of the actuual queue names" do - Destinations.define do |queue| + Destinations.define do |queue| queue.map :foo, "/queue/foo" queue.map :bar, "/queue/bar" end - + Destinations.queue_names.should include("/queue/foo") Destinations.queue_names.should include("/queue/bar") end end -end \ No newline at end of file +end spec/rosetta_queue/destinations_spec.rb @@ -1,44 +1,44 @@ require File.dirname(__FILE__) + '/../spec_helper' module RosettaQueue - + describe Filters do - + after(:each) do Filters.reset end - + describe "#process_receiving" do it "should process the passed in message with the defined receiving filter" do - Filters.define do |f| + Filters.define do |f| f.receiving {|message| "Foo #{message}"} end - + Filters.process_receiving("Bar").should == "Foo Bar" end - + it "should return the same message when no filter is defined" do Filters.process_receiving("Bar").should == "Bar" end end - - + + describe "#process_sending" do it "should process the passed in message with the defined sending filter" do - Filters.define do |f| + Filters.define do |f| f.sending {|message| "Foo #{message}"} end - + Filters.process_sending("Bar").should == "Foo Bar" end - + it "should return the same message when no filter is defined" do Filters.process_sending("Bar").should == "Bar" - end + end end - - - + + + end - + end spec/rosetta_queue/filters_spec.rb @@ -1,42 +1,42 @@ require File.dirname(__FILE__) + '/../spec_helper' module RosettaQueue - + class TestProducer < Producer publishes_to :test_queue options :persistent => false end - - describe Producer do - + + describe Producer do + before(:each) do @adapter = mock("adapter", :send_message => nil) RosettaQueue::Adapter.stub!(:instance).and_return(@adapter) @gateway = TestProducer.new - + Destinations.stub!(:lookup).and_return("/queue/test_queue") end - + it_should_behave_like "a messaging gateway object" attr_reader :adapter, :gateway - - + + describe "#publish" do - + before(:each) do @adapter = mock("adpater", :send_message => nil) RosettaQueue::Adapter.stub!(:instance).and_return(@adapter) end - + # it "should look up the destination defined on the class" do # Destinations.should_receive(:lookup).with(:test_queue).and_return("/queue/test_queue") # # when # @gateway.publish('some message') # end - it "should publish messages to queue with the options defined in the class" do + it "should publish messages to queue with the options defined in the class" do # TO DO: REFACTOR #publish METHOD SO THAT YOU PASS IN MESSAGE HANDLER AS WITH CONSUMER pending # expect @@ -46,14 +46,14 @@ module RosettaQueue end end - + describe ".publish" do # it "should look up the destination defined on the class" do # Destinations.should_receive(:lookup).with(:test_queue).and_return("/queue/test_queue") # # when # Producer.publish(:test_queue, "blah") # end - + it "should send the message to the adpater along with the options" do # expect @adapter.should_receive(:send_message).with("/queue/test_queue", "Hello World!", {:persistent => true}) @@ -61,6 +61,6 @@ module RosettaQueue Producer.publish(:test_queue, "Hello World!", {:persistent => true}) end end - + end -end \ No newline at end of file +end spec/rosetta_queue/producer_spec.rb @@ -1,21 +1,21 @@ module RosettaQueue - describe "a messaging gateway object", :shared => true do - - it "#unsubscribe should be delegated to the adapter" do + describe "a messaging gateway object", :shared => true do + + it "#unsubscribe should be delegated to the adapter" do pending # expect adapter.should_receive("unsubscribe") # when gateway.unsubscribe end - + it "#disconnect should be delegated to the adapter" do # expect adapter.should_receive("disconnect") # when gateway.disconnect end - + end -end \ No newline at end of file +end spec/rosetta_queue/shared_messaging_behavior.rb 7a01bf490ee1e6f69b8c26ad8b12e999dee4fc91 Ben Mabey ben@benmabey.com http://github.com/bmabey/rosetta_queue/commit/1e9832131bdef842ef98c99ef0181f69f8f20875 1e9832131bdef842ef98c99ef0181f69f8f20875 2009-09-23T13:57:32-07:00 2009-09-23T13:57:32-07:00 whitespace begone! addb2fc3b058c9e275fcceffc029807a7d5dee98 Ben Mabey ben@benmabey.com