lib/activemessaging/adapters/reliable_msg.rb
# 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)
            puts "sleep #{@poll_interval}..." if (current_subscription == start)
            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]
              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.id, reliable_msg.object, reliable_msg.headers, destination_name, 'MESSAGE', @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 ) puts "sleep #{@poll_interval}..." if ( current_subscription == start ) 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 ] 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 . id , reliable_msg . object , reliable_msg . headers , destination_name , 'MESSAGE' , @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