vendor/bundle/ruby/1.8/gems/activemessaging-0.9.0/lib/activemessaging/adapters/reliable_msg.rb
def receive

        raise "No subscriptions to receive messages from." if (subscriptions.nil? || subscriptions.empty?)
        start = current_subscription
        while true
          self.current_subscription = ((current_subscription < subscriptions.length-1) ? current_subscription + 1 : 0)
          sleep poll_interval if (current_subscription == start)
          destination_name = subscriptions.keys.sort[current_subscription]
          destination = destinations[destination_name]
          unless destination.nil?
            # from the way we use this, assume this is the start of a transaction, 
            # there should be no current transaction
            ctx = Thread.current[::ReliableMsg::Client::THREAD_CURRENT_TX]
            raise "There should not be an existing reliable-msg transaction. #{ctx.inspect}" if ctx

            # start a new transaction
            @tx = {:qm=>destination.queue_manager}
            @tx[:tid] = @tx[:qm].begin @tx_timeout
            Thread.current[::ReliableMsg::Client::THREAD_CURRENT_TX] = @tx
            begin

              # now call a get on the destination - it will use the transaction
              #the commit or the abort will occur in the received or unreceive methods
              reliable_msg = destination.get subscriptions[destination_name].headers[:selector]
              @tx[:qm].commit(@tx[:tid]) if reliable_msg.nil?

            rescue Object=>err
              #abort the transaction on error
              @tx[:qm].abort(@tx[:tid])

              raise err unless reliable
              puts "receive failed, will retry in #{@poll_interval} seconds"
              sleep poll_interval
            end
            return Message.new(reliable_msg.object, reliable_msg.id, reliable_msg.headers, destination_name, @tx) if reliable_msg
            
            Thread.current[::ReliableMsg::Client::THREAD_CURRENT_TX] = nil
          end
        end
      end
# receive a single message from any of the subscribed destinations
# check each destination once, then sleep for poll_interval
receive a single message from any of the subscribed destinations
check each destination once, then sleep for poll_interval
receive a single message from any of the subscribed destinations check each destination once then sleep for poll_interval
def receive
        raise "No subscriptions to receive messages from." if (subscriptions.nil? || subscriptions.empty?)
        start = current_subscription
        while true
          self.current_subscription = ((current_subscription < subscriptions.length-1) ? current_subscription + 1 : 0)
          sleep poll_interval if (current_subscription == start)
          destination_name = subscriptions.keys.sort[current_subscription]
          destination = destinations[destination_name]
          unless destination.nil?
            ctx = Thread.current[::ReliableMsg::Client::THREAD_CURRENT_TX]
            raise "There should not be an existing reliable-msg transaction. #{ctx.inspect}" if ctx
            @tx = {:qm=>destination.queue_manager}
            @tx[:tid] = @tx[:qm].begin @tx_timeout
            Thread.current[::ReliableMsg::Client::THREAD_CURRENT_TX] = @tx
            begin
              reliable_msg = destination.get subscriptions[destination_name].headers[:selector]
              @tx[:qm].commit(@tx[:tid]) if reliable_msg.nil?
            rescue Object=>err
              @tx[:qm].abort(@tx[:tid])
              raise err unless reliable
              puts "receive failed, will retry in #{@poll_interval} seconds"
              sleep poll_interval
            end
            return Message.new(reliable_msg.object, reliable_msg.id, reliable_msg.headers, destination_name, @tx) if reliable_msg
            Thread.current[::ReliableMsg::Client::THREAD_CURRENT_TX] = nil
          end
        end
      end
def receive raise "No subscriptions to receive messages from." if ( subscriptions . nil? || subscriptions . empty? ) start = current_subscription while true self . current_subscription = ( ( current_subscription < subscriptions . length - 1 ) ? current_subscription + 1 : 0 ) sleep poll_interval if ( current_subscription == start ) destination_name = subscriptions . keys . sort [ current_subscription ] destination = destinations [ destination_name ] unless destination . nil? ctx = Thread . current [ :: ReliableMsg :: Client :: THREAD_CURRENT_TX ] raise "There should not be an existing reliable-msg transaction. #{ctx.inspect}" if ctx @tx = { :qm => destination . queue_manager } @tx [ :tid ] = @tx [ :qm ] . begin @tx_timeout Thread . current [ :: ReliableMsg :: Client :: THREAD_CURRENT_TX ] = @tx begin reliable_msg = destination . get subscriptions [ destination_name ] . headers [ :selector ] @tx [ :qm ] . commit ( @tx [ :tid ] ) if reliable_msg . nil? rescue Object => err @tx [ :qm ] . abort ( @tx [ :tid ] ) raise err unless reliable puts "receive failed, will retry in #{@poll_interval} seconds" sleep poll_interval end return Message . new ( reliable_msg . object , reliable_msg . id , reliable_msg . headers , destination_name , @tx ) if reliable_msg Thread . current [ :: ReliableMsg :: Client :: THREAD_CURRENT_TX ] = nil end end end
receive a single message from any of the subscribed destinations
check each destination once, then sleep for poll_interval
receive a single message from any of the subscribed destinations check each destination once then sleep for poll_interval
# from the way we use this, assume this is the start of a transaction,  # there should be no current transaction # start a new transaction # now call a get on the destination - it will use the transaction #the commit or the abort will occur in the received or unreceive methods #abort the transaction on error