require 'starling'
require 'timeout'

module Airbrush
  include Timeout
  
  class Client
    DEFAULT_INCOMING_QUEUE = 'airbrush_incoming_queue'
    DEFAULT_RESPONSE_TIMEOUT = 2.minutes
    DEFAULT_QUEUE_VALIDITY = 0 #This should be changed back to a valid timeout once we know the problem.
    
    attr_reader :host, :incoming_queue, :response_timeout, :queue_validity

    def initialize(host, incoming_queue = DEFAULT_INCOMING_QUEUE, response_timeout = DEFAULT_RESPONSE_TIMEOUT, queue_validity = DEFAULT_QUEUE_VALIDITY)
      @host = host
      @server = Starling.new(@host)
      @incoming_queue = incoming_queue
      @response_timeout = response_timeout
      @queue_validity = queue_validity
    end
    
    def process(id, command, args = {})
      raise 'No job id specified' unless id
      raise 'No command specified' unless command
      raise "Invalid arguments #{args}" unless args.is_a? Hash
      
      send_and_receive(id, command, args)
    end
    
    private
    
      def send_and_receive(id, command, args)
        @server.set(@incoming_queue, { :id => id, :command => command, :args => args }, @queue_validity, false)
        queue = unique_name(id)
        Timeout::timeout(@response_timeout) do
          return @server.get(queue)
        end
      end
      
      # REVISIT: share implementation with server?
      def unique_name(id)
        id.to_s
      end 
  end
end
require 'starling'
require 'timeout'

module Airbrush
  include Timeout

  class Client
    DEFAULT_INCOMING_QUEUE   = 'airbrush_incoming_queue'
    DEFAULT_RESPONSE_TIMEOUT = 2.minutes
    DEFAULT_QUEUE_VALIDITY   = 10.minutes

    attr_reader :host, :incoming_queue, :response_timeout, :queue_validity

    def initialize(host, incoming_queue = DEFAULT_INCOMING_QUEUE, response_timeout = DEFAULT_RESPONSE_TIMEOUT, queue_validity = DEFAULT_QUEUE_VALIDITY)
      @host = host
      @server = Starling.new(@host)
      @incoming_queue = incoming_queue
      @response_timeout = response_timeout
      @queue_validity = queue_validity
    end

    def process(id, command, args = {})
      raise 'No job id specified' unless id
      raise 'No command specified' unless command
      raise "Invalid arguments #{args}" unless args.is_a? Hash

      send_and_receive(id, command, args)
    end

    private

      def send_and_receive(id, command, args)
        @server.set(@incoming_queue, { :id => id, :command => command, :args => args }, @queue_validity, false)
        queue = unique_name(id)
        Timeout::timeout(@response_timeout) do
          return @server.get(queue)
        end
      end

      # REVISIT: share implementation with server?
      def unique_name(id)
        id.to_s
      end
  end
end

Update queue validity to 10 minutes to expire items in the airbursh queue
Update queue validity to 10 minutes to expire items in the airbursh queue


git-svn-id: 6d9b9a8f21f96071dc3ac7ff320f08f2a342ec80@80 daa9635f-4444-0410-9b3d-b8c75a019348
