def on_message(params, &block)
      raise 'JMS::Connection must be connected prior to calling JMS::Connection::on_message' unless @sessions && @consumers

      consumer_count = params[:session_count] || 1
      consumer_count.times do
        session  = self.create_session(params)
        consumer = session.consumer(params)
        if session.transacted?
          consumer.on_message(params) do |message|
            begin
              block.call(message) ? session.commit : session.rollback
            rescue => exc
              session.rollback
              throw exc
            end
          end
        else
          consumer.on_message(params, &block)
        end
        @consumers << consumer
        @sessions << session
      end
    end
def on_message(params, &block)
      raise 'JMS::Connection must be connected prior to calling JMS::Connection::on_message' unless @sessions && @consumers

      consumer_count = params[:session_count] || 1
      consumer_count.times do
        session  = self.create_session(params)
        consumer = session.consumer(params)
        if session.transacted?
          consumer.on_message(params) do |message|
            begin
              block.call(message) ? session.commit : session.rollback
            rescue => exc
              session.rollback
              throw exc
            end
          end
        else
          consumer.on_message(params, &block)
        end
        @consumers << consumer
        @sessions << session
      end
    end
Receive messages in a separate thread when they arrive

 Allows messages to be received Asynchronously in a separate thread.
 This method will return to the caller before messages are processed.
 It is then the callers responsibility to keep the program active so that messages
 can then be processed.

 Session Parameters:
  transacted: true or false
      Determines whether transactions are supported within this session.
      I.e. Whether commit or rollback can be called
      Default: false
      Note: :options below are ignored if this value is set to :true

  options: any of the JMS::Session constants:
     Note: :options are ignored if transacted: true
     JMS::Session::AUTO_ACKNOWLEDGE
        With this acknowledgment mode, the session automatically acknowledges
        a client's receipt of a message either when the session has successfully
        returned from a call to receive or when the message listener the session has
        called to process the message successfully returns.
     JMS::Session::CLIENT_ACKNOWLEDGE
        With this acknowledgment mode, the client acknowledges a consumed
        message by calling the message's acknowledge method.
     JMS::Session::DUPS_OK_ACKNOWLEDGE
        This acknowledgment mode instructs the session to lazily acknowledge
        the delivery of messages.
     JMS::Session::SESSION_TRANSACTED
        This value is returned from the method getAcknowledgeMode if the
        session is transacted.
     Default: JMS::Session::AUTO_ACKNOWLEDGE

   :session_count : Number of sessions to create, each with their own consumer which
                    in turn will call the supplied code block.
                    Note: The supplied block must be thread safe since it will be called
                          by several threads at the same time.
                          I.e. Don't change instance variables etc. without the
                          necessary semaphores etc.
                    Default: 1

 Consumer Parameters:
   queue_name: String: Name of the Queue to return
                  Symbol: temporary: Create temporary queue
                  Mandatory unless :topic_name is supplied
     Or,
   topic_name: String: Name of the Topic to write to or subscribe to
                  Symbol: temporary: Create temporary topic
                  Mandatory unless :queue_name is supplied
     Or,
   destination:Explicit javaxJms::Destination to use

   selector:   Filter which messages should be returned from the queue
                  Default: All messages

   no_local:   Determine whether messages published by its own connection
                  should be delivered to the supplied block
                  Default: false

   :statistics Capture statistics on how many messages have been read
      true  : This method will capture statistics on the number of messages received
              and the time it took to process them.
              The timer starts when each() is called and finishes when either the last message was received,
              or when Destination::statistics is called. In this case MessageConsumer::statistics
              can be called several times during processing without affecting the end time.
              Also, the start time and message count is not reset until MessageConsumer::each
              is called again with statistics: true

 Usage: For transacted sessions the block supplied must return either true or false:
          true => The session is committed
          false => The session is rolled back
          Any Exception => The session is rolled back

 Note: Separately invoke Connection#on_exception so that connection failures can be handled
       since on_message will Not be called if the connection is lost