def event_loop
      Qwirk.logger.debug "#{self}: Starting receive loop"
      @start_worker_time = Time.now
      until @stopped || (config.stopped? && @impl.ready_to_stop?)
        Qwirk.logger.debug "#{self}: Waiting for read"
        @start_read_time = Time.now
        msg = @impl.receive_message
        if msg
          @start_processing_time = Time.now
          Qwirk.logger.debug {"#{self}: Done waiting for read in #{@start_processing_time - @start_read_time} seconds"}
          delta = config.timer.measure do
            @processing_mutex.synchronize do
              on_message(msg)
              @impl.acknowledge_message(msg)
            end
          end
          Qwirk.logger.info {"#{self}::on_message (#{'%.1f' % delta}ms)"} if self.config.log_times
          Qwirk.logger.flush if Qwirk.logger.respond_to?(:flush)
        end
      end
      Qwirk.logger.info "#{self}: Exiting"
    rescue Exception => e
      @status = "Terminated: #{e.message}"
      Qwirk.logger.error "#{self}: Exception, thread terminating: #{e.message}\n\t#{e.backtrace.join("\n\t")}"
    ensure
      @status = 'Stopped'
      Qwirk.logger.flush if Qwirk.logger.respond_to?(:flush)
      config.worker_stopped(self)
    end
def event_loop
      Qwirk.logger.debug "#{self}: Starting receive loop"
      @start_worker_time = Time.now
      until @stopped || (config.stopped? && @impl.ready_to_stop?)
        Qwirk.logger.debug "#{self}: Waiting for read"
        @start_read_time = Time.now
        msg = @impl.receive_message
        if msg
          @start_processing_time = Time.now
          Qwirk.logger.debug {"#{self}: Done waiting for read in #{@start_processing_time - @start_read_time} seconds"}
          delta = config.timer.measure do
            @processing_mutex.synchronize do
              on_message(msg)
              @impl.acknowledge_message(msg)
            end
          end
          Qwirk.logger.info {"#{self}::on_message (#{'%.1f' % delta}ms)"} if self.config.log_times
          Qwirk.logger.flush if Qwirk.logger.respond_to?(:flush)
        end
      end
      Qwirk.logger.info "#{self}: Exiting"
    rescue Exception => e
      @status = "Terminated: #{e.message}"
      Qwirk.logger.error "#{self}: Exception, thread terminating: #{e.message}\n\t#{e.backtrace.join("\n\t")}"
    ensure
      @status = 'Stopped'
      Qwirk.logger.flush if Qwirk.logger.respond_to?(:flush)
      config.worker_stopped(self)
    end
def event_loop Qwirk . logger . debug "#{self}: Starting receive loop" @start_worker_time = Time . now until @stopped || ( config . stopped? && @impl . ready_to_stop? ) Qwirk . logger . debug "#{self}: Waiting for read" @start_read_time = Time . now msg = @impl . receive_message if msg @start_processing_time = Time . now Qwirk . logger . debug { "#{self}: Done waiting for read in #{@start_processing_time - @start_read_time} seconds" } delta = config . timer . measure do @processing_mutex . synchronize do on_message ( msg ) @impl . acknowledge_message ( msg ) end end Qwirk . logger . info { "#{self}::on_message (#{'%.1f' % delta}ms)" } if self . config . log_times Qwirk . logger . flush if Qwirk . logger . respond_to? ( :flush ) end end Qwirk . logger . info "#{self}: Exiting" rescue Exception => e @status = "Terminated: #{e.message}" Qwirk . logger . error "#{self}: Exception, thread terminating: #{e.message}\n\t#{e.backtrace.join("\n\t")}" ensure @status = 'Stopped' Qwirk . logger . flush if Qwirk . logger . respond_to? ( :flush ) config . worker_stopped ( self ) end
Start the event loop for handling messages off the queue
Start the event loop for handling messages off the queue