initial import of failover code

lib/amqp-failover.rb

@@ -1,5 +0,0 @@
-module Amqp
-  module Failover
-    # Your code goes here...
-  end
-end

lib/amqp/failover.rb

@@ -0,0 +1,18 @@
+# encoding: utf-8
+
+require 'amqp'
+require 'yaml'
+
+require 'amqp/failover/basic_client'
+require 'amqp/failover/config'
+require 'amqp/failover/fallback'
+require 'amqp/failover/logger'
+require 'amqp/failover/logic'
+require 'amqp/failover/logic/failed_config'
+require 'amqp/failover/version'
+
+module AMQP
+  module Failover
+    
+  end
+end

lib/amqp/failover/basic_client.rb

@@ -0,0 +1,55 @@
+# encoding: utf-8
+
+module AMQP
+  module Failover
+    module BasicClient
+      include AMQP::BasicClient
+      
+      class Error < Exception; end
+      
+      attr_accessor :on_disconnect
+      attr_accessor :settings
+      
+      def self.extended(base)
+        base.on_disconnect = proc {
+          OnDisconnect.new(base).call
+        }
+      end
+      
+      def logger
+        @logger ||= Logger.new
+      end
+      
+      def failover_conf
+        @failover_conf ||= Config.new
+      end
+      
+      def configs
+        failover_conf.configs
+      end
+      
+      def clean_exit(msg = nil)
+        msg ||= "clean exit"
+        logger.info(msg)
+        logger.error(msg)
+        Process.exit
+      end
+      
+      def process_frame(frame)
+        if mq = channels[frame.channel]
+          mq.process_frame(frame)
+          return
+        end
+        
+        if frame.is_a?(::AMQP::Frame::Method) && (method = frame.payload).is_a?(::AMQP::Protocol::Connection::Close)
+          if method.reply_text =~ /^NOT_ALLOWED/
+            raise ::AMQP::Error, "#{method.reply_text} in #{::AMQP::Protocol.classes[method.class_id].methods[method.method_id]}"
+          end
+        end
+        super(frame)
+      end
+      
+    end # BasicClient
+  end # Failover
+end # AMQP
+

lib/amqp/failover/basic_client/on_disconnect.rb

@@ -0,0 +1,46 @@
+# encoding: utf-8
+
+module AMQP
+  module Failover
+    module BasicClient
+      class OnDisconnect
+        
+        attr_accessor :base
+        attr_accessor :failover
+        attr_accessor :configs
+        attr_accessor :logger
+        
+        def initialize(base)
+          @base = base
+          @configs = @base.configs
+          @failover_conf = @base.failover_conf
+          @logger = @base.logger
+        end
+        
+        def call
+          @logic ||= Logic.new(@configs.configs, @failover_conf.get_primary, @failover_conf.failover_config)
+          if (new_settings = @logic.failover_from(@base.settings))
+            log_message = "Could not connect to or lost connection to server #{@base.settings[:host]}:#{@base.settings[:port]}. " +
+                          "Attempting connection to: #{new_settings[:host]}:#{new_settings[:port]}"
+            @logger.error(log_message)
+            @logger.info(log_message)
+            
+            if @failover_conf.get_primary == @base.settings
+              FallbackMonitor.monitor(@failover_conf.get_primary) do
+                @base.clean_exit("Primary server (#{@failover_conf.get_primary[:host]}:#{@failover_conf.get_primary[:port]}) is back. " +
+                                 "Performing clean exit to be relaunched with primary config.")
+              end
+            end
+            
+            @base.settings = new_settings
+            @base.reconnect
+          else
+            raise Error, "Could not connect to server #{@base.settings[:host]}:#{@base.settings[:port]}"
+          end
+        end
+        
+      end # OnDisconnect
+    end # BasicClient
+  end # Failover
+end # AMQP
+

lib/amqp/failover/config.rb

@@ -0,0 +1,94 @@
+# encoding: utf-8
+
+module AMQP
+  module Failover
+    class Config
+      
+      attr_accessor :configs
+      attr_accessor :failover_config
+      
+      def failover_config
+        @failover_config ||= { :retry_timeout => 30 }
+      end
+      
+      def refs
+        @refs ||= {}
+      end
+      
+      def configs
+        @configs ||= []
+      end
+      
+      def primary
+        @primary ||= 0
+      end
+      
+      def primary=(ref)
+        @primary = ref
+      end
+      
+      def get_primary
+        get(primary) || default_config
+      end
+      
+      def set_primary(conf = {})
+        set(conf, primary)
+      end
+      
+      def get(ref = nil)
+        return configs[ref] if ref.is_a?(Fixnum)
+        configs[refs[ref]] if refs[ref]
+      end
+      
+      def set(conf = {}, ref = nil)
+        conf = default_config.merge(conf)
+        configs << conf if (index = configs.index(conf)).nil?
+        if ref
+          refs[ref] = (index || configs.index(conf))
+        end
+      end
+      
+      def find_next(conf = {})
+        current = configs.index(conf)
+        configs[(current+1 == configs.size) ? 0 : current+1] if current
+      end
+      
+      def load_file(file, env = nil)
+        raise ArgumentError, "Can't find #{file}" unless File.exists?(file)
+        load(YAML.load_file(file)[env || "development"])
+      end
+      
+      def load_yaml(data, env = nil)
+        load(YAML.load(data)[env || "development"])
+      end
+      
+      def load(conf)
+        if conf.is_a?(::Array)
+          load_array(conf)
+        elsif conf.is_a?(::Hash)
+          load_hash(conf)
+        end
+      end
+      
+      def load_array(confs = [])
+        @configs = nil
+        confs.each do |conf|
+          load_hash(conf)
+        end
+      end
+      
+      def load_hash(conf = {})
+        conf = conf.inject({}) do |result, (key, value)|
+          result[key.is_a?(String) ? key.to_sym : key] = value
+          result
+        end
+        self.set(conf)
+      end
+      
+      def default_config
+        AMQP.settings
+      end
+      
+    end # Config
+  end # Failover
+end # AMQP

lib/amqp/failover/fallback.rb

@@ -0,0 +1,44 @@
+# encoding: utf-8
+
+module AMQP
+  module Failover
+    class Fallback < EM::Connection
+      
+      class << self
+        attr_accessor :connection
+      end
+      
+      def initialize(args)
+        @done = args[:done]
+        @timer = args[:timer]
+      end
+      
+      def connection_completed
+        @done.call
+        @timer.cancel
+        close_connection
+      end
+      
+      def self.monitor(conf = {}, &block)
+        if EM.reactor_running?
+          start_monitoring(conf, &block)
+        else
+          EM.run { start_monitoring(conf, &block) }
+        end
+      end
+      
+      def self.start_monitoring(conf = {}, &block)
+        conf = conf.clone
+        conf[:done] = block
+        conf[:timer] = EM::PeriodicTimer.new(conf[:retry_interval] || 5) do
+          @connection = connect(conf)
+        end
+      end
+      
+      def self.connect(conf)
+        EM.connect(conf[:host], conf[:port], self, conf)
+      end
+      
+    end # Fallback
+  end # Failover
+end # AMQP

lib/amqp/failover/logger.rb

@@ -0,0 +1,31 @@
+# encoding: utf-8
+
+module AMQP
+  module Failover
+    class Logger
+    
+      attr_accessor :enabled
+    
+      def initialize(enabled = nil)
+        @enabled = enabled || true
+      end
+    
+      def error(*msg)
+        msg[0] = "[ERROR]: " + msg[0] if msg[0].is_a?(String)
+        write(*msg)
+      end
+    
+      def info(*msg)
+        write(*msg)
+      end
+    
+      private
+    
+      def write(*msg)
+        return if !@enabled
+        puts *msg
+      end
+    
+    end # Logger
+  end # Failover
+end # AMQP

lib/amqp/failover/logic.rb

@@ -0,0 +1,86 @@
+# encoding: utf-8
+
+module AMQP
+  module Failover
+    class Logic
+      
+      attr_reader :latest_failed
+      attr_accessor :primary
+      attr_accessor :retry_timeout
+      attr_accessor :fallback
+      
+      def initialize(confs = nil, primary = nil, options = {})
+        @primary = primary
+        @retry_timeout = (options.delete(:retry_timeout) || 30)
+        self.configs = confs if !confs.nil?
+      end
+      
+      def refs
+        @refs ||= {}
+      end
+      
+      def configs
+        @configs ||= []
+      end
+      
+      def configs=(confs = [])
+        @configs = []
+        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 << FailedConfig.new(conf) if index.nil?
+        refs[ref] = (index || configs.index(conf)) if !ref.nil?
+      end
+      
+      def failover_from(conf = {}, time = nil)
+        failed_with(conf, nil, time)
+        next_config
+      end
+      
+      def failed_with(conf = {}, ref = nil, time = nil)
+        time ||= Time.now
+        if index = configs.index(conf)
+          configs[index].last_fail = time
+          @latest_failed = configs[index]
+        else
+          configs << FailedConfig.new(conf, time)
+          @latest_failed = configs.last
+        end
+        refs[ref] = (index || configs.index(conf)) if !ref.nil?
+      end
+      
+      def next_config(retry_timeout = nil, after = nil)
+        return nil if configs.size <= 1
+        retry_timeout ||= @retry_timeout
+        after ||= @latest_failed
+        index = configs.index(after)
+        available = (index > 0) ? configs[index+1..-1] + configs[0..index-1] : configs[1..-1]
+        available.each do |conf|
+          return conf if conf.last_fail.nil? || (conf.last_fail + retry_timeout.seconds) < Time.now
+        end
+        return nil
+      end
+      
+      def last_fail_of(match)
+        ((match.is_a?(Hash) ? get_by_conf(match) : get_by_ref(match)) || FailedConfig.new).last_fail
+      end 
+      
+      def get_by_conf(conf = {})
+        configs[configs.index(conf)]
+      end
+      
+      def get_by_ref(ref = nil)
+        configs[refs[ref]] if refs[ref]
+      end
+      
+    end # Logic
+  end # Failover
+end # AMQP

lib/amqp/failover/logic/failed_config.rb

@@ -0,0 +1,33 @@
+# encoding: utf-8
+
+module AMQP
+  module Failover
+    class Logic
+      class FailedConfig < ::Hash
+        
+        attr_accessor :last_fail
+        
+        def initialize(hash = {}, last_fail_date = nil)
+          self.replace(hash)
+          self.last_fail = last_fail_date if last_fail_date
+        end
+        
+        # order by latest fail, potentially useful if random config selection is used
+        def <=>(other)
+          if self.respond_to?(:last_fail) && other.respond_to?(:last_fail)
+            if self.last_fail.nil? && other.last_fail.nil?
+              return 0
+            elsif self.last_fail.nil? && !other.last_fail.nil?
+              return 1
+            elsif !self.last_fail.nil? && other.last_fail.nil?
+              return -1
+            end
+            return other.last_fail <=> self.last_fail
+          end
+          return 0
+        end
+        
+      end # FailedConfig
+    end # Logic
+  end # Failover
+end # AMQP

lib/amqp/failover/version.rb

@@ -1,4 +1,6 @@
-module Amqp
+# encoding: utf-8
+
+module AMQP
   module Failover
     VERSION = "0.0.1"
   end