diff --git a/lib/amqp/failover.rb b/lib/amqp/failover.rb index df6f4b6..4de6b3d 100644 --- a/lib/amqp/failover.rb +++ b/lib/amqp/failover.rb @@ -20,7 +20,7 @@ module AMQP attr_accessor :fallback def initialize(confs = nil, opts = {}) - @configs = Configs.new(confs) + @configs = Failover::Configs.new(confs) @options = default_options.merge(opts) end @@ -32,7 +32,7 @@ module AMQP def default_options { :retry_timeout => 1, :selection => :sequential, #TODO: Impliment next server selection algorithm - :fallback => false, + :fallback => false, #TODO: Enable by default once a sane solution is found :fallback_interval => 10 } end @@ -56,17 +56,6 @@ module AMQP @configs ||= Config.new end - def configs=(confs = []) - @configs = nil - confs.each do |conf| - if conf.is_a?(Array) - add_config(conf[1], conf[0]) - else - add_config(conf) - end - end - end - def add_config(conf = {}, ref = nil) index = configs.index(conf) configs << Config::Failed.new(conf) if index.nil? diff --git a/lib/amqp/failover/config.rb b/lib/amqp/failover/config.rb index 784492c..dbaef58 100644 --- a/lib/amqp/failover/config.rb +++ b/lib/amqp/failover/config.rb @@ -7,10 +7,14 @@ module AMQP attr_accessor :last_fail def initialize(hash = {}, last_fail_date = nil) - self.replace(symbolize_keys(hash)) + self.replace(symbolize_keys(defaults.merge(hash))) self.last_fail = last_fail_date if last_fail_date end + def defaults + AMQP.settings + end + def symbolize_keys(hash = {}) hash.inject({}) do |result, (key, value)| result[key.is_a?(String) ? key.to_sym : key] = value @@ -30,7 +34,7 @@ module AMQP end return other.last_fail <=> self.last_fail end - return 0 + super(other) end end # Config diff --git a/lib/amqp/failover/configs.rb b/lib/amqp/failover/configs.rb index 251b458..4858b55 100644 --- a/lib/amqp/failover/configs.rb +++ b/lib/amqp/failover/configs.rb @@ -46,7 +46,7 @@ module AMQP end def set(conf = {}, ref = nil) - conf = Config.new(default_config.merge(conf)) + conf = Failover::Config.new(conf) if !conf.is_a?(Failover::Config) self << conf if (index = self.index(conf)).nil? if ref refs[ref] = (index || self.index(conf)) @@ -75,7 +75,7 @@ module AMQP end end - def self.load_array(confs = []) + def load_array(confs = []) self.clear confs.each do |conf| conf = AMQP::Client.parse_amqp_url(conf) if conf.is_a?(::String) @@ -84,11 +84,7 @@ module AMQP end def load_hash(conf = {}) - set(Config.new(conf)) - end - - def default_config - AMQP.settings + set(conf) end end # Config diff --git a/lib/amqp/failover_client.rb b/lib/amqp/failover_client.rb index 0f06f9a..fd6c048 100644 --- a/lib/amqp/failover_client.rb +++ b/lib/amqp/failover_client.rb @@ -38,7 +38,9 @@ module AMQP logger.error(log_message) logger.info(log_message) - fallback(@failover.primary, @failover.fallback_interval) if @failover.primary == @settings + if @failover.options[:fallback] && @failover.primary == @settings + fallback(@failover.primary, @failover.fallback_interval) + end @settings = new_settings reconnect else @@ -53,6 +55,7 @@ module AMQP end def fallback_callback + #TODO: Figure out a way to artificially trigger EM to disconnect on fallback without channels being closed. @fallback_callback ||= proc { |conf, retry_interval| clean_exit("Primary server (#{conf[:host]}:#{conf[:port]}) is back. " + "Performing clean exit to be relaunched with primary config.") diff --git a/spec/unit/amqp/failover/config_spec.rb b/spec/unit/amqp/failover/configs_spec.rb similarity index 61% rename from spec/unit/amqp/failover/config_spec.rb rename to spec/unit/amqp/failover/configs_spec.rb index 2a1ae41..f9a79f6 100644 --- a/spec/unit/amqp/failover/config_spec.rb +++ b/spec/unit/amqp/failover/configs_spec.rb @@ -3,52 +3,39 @@ $LOAD_PATH.unshift File.expand_path(File.dirname(__FILE__)) require 'spec_helper' -describe 'AMQP::Failover::Config' do - - before(:all) do - # @conf = AMQP::Failover::Config.new - end +describe 'AMQP::Failover::Configs' do before(:each) do - @conf = AMQP::Failover::Config.new - # [:primary, :configs, :refs].each do |var| - # @conf.instance_variable_set("@#{var}", nil) - # end + @conf = AMQP::Failover::Configs.new @raw_configs = [ {:host => 'rabbit3.local'}, {:host => 'rabbit2.local'}, {:host => 'rabbit2.local', :port => 5673} ] - @configs = @raw_configs.map { |conf| @conf.default_config.merge(conf) } - end - - after(:each) do - # [:primary, :configs, :refs].each do |var| - # @conf.instance_variable_set("@#{var}", nil) - # end + @configs = @raw_configs.map { |conf| AMQP.settings.merge(conf) } end it "should set and get configs" do @conf.primary.should == 0 - @conf.configs.should have(0).items + @conf.should have(0).items @conf.set(@raw_configs[0]) - @conf.configs.should have(1).items + @conf.should have(1).items @conf.get(0).should == @configs[0] @conf[0].should == @configs[0] @conf.set(@raw_configs[1]) - @conf.configs.should have(2).items + @conf.should have(2).items @conf.get(1).should == @configs[1] @conf[1].should == @configs[1] @conf.set(@raw_configs[1], :the_one) - @conf.configs.should have(2).items + @conf.should have(2).items @conf.get(1).should == @configs[1] @conf[:the_one].should == @configs[1] @conf.load_array(@raw_configs) - @conf.configs.should have(3).items + @conf.should have(3).items @conf.get_primary.should == @configs[0] @conf.primary = 1 @conf.get_primary.should == @configs[1] @@ -57,7 +44,7 @@ describe 'AMQP::Failover::Config' do it "should #find_next" do @conf.load(@raw_configs) - @conf.configs.should have(3).items + @conf.should have(3).items @conf.find_next(@configs[0]).should == @configs[1] @conf.find_next(@configs[1]).should == @configs[2] @conf.find_next(@configs[2]).should == @configs[0]