<commit_before>require 'benchmark'

module ModernTimes
  module JMS
  
    # Base Worker Class for any class that will be processing messages from topics or queues
    # By default, it will consume messages from a queue with the class name minus the Worker postfix.
    # For example, the queue call is unneccessary as it will default to a value of 'Foo' anyways:
    #  class FooWorker < ModernTimes::JMS::Worker
    #    queue 'Foo'
    #    def perform(obj)
    #      # Perform work on obj
    #    end
    #  end
    #
    # A topic can be specified using virtual_topic as follows (ActiveMQ only).  Multiple separate workers can
    # subscribe to the same topic (under ActiveMQ - see http://activemq.apache.org/virtual-destinations.html):
    #  class FooWorker < ModernTimes::JMS::Worker
    #    virtual_topic 'Zulu'
    #    def perform(obj)
    #      # Perform work on obj
    #    end
    #  end
    #
    # Filters can also be specified within the class:
    #  class FooWorker < ModernTimes::JMS::Worker
    #    filter 'age > 30'
    #    def perform(obj)
    #      # Perform work on obj
    #    end
    #  end
    #
    #
    module Worker
      include ModernTimes::Base::Worker

      attr_reader :session, :error_count, :time_track
      attr_accessor :message

      module ClassMethods
        def create_supervisor(manager, worker_options)
          Supervisor.new(manager, self, {}, worker_options)
        end

        def destination_options
          options = dest_options.dup
          # Default the queue name to the Worker name if a destinations hasn't been specified
          if options.keys.select {|k| [:virtual_topic_name, :queue_name, :destination].include?(k)}.empty?
            options[:queue_name] = default_name
          end
          return options
        end

        def virtual_topic(name, opts={})
          # ActiveMQ only
          dest_options[:virtual_topic_name] = name.to_s
        end

        def queue(name, opts={})
          dest_options[:queue_name] = name.to_s
        end

       def dest_options
          @dest_options ||= {}
        end

        def failure_queue(name, opts={})
          @failure_queue_name = name.to_s
        end

        def failure_queue_name
          # Don't overwrite if the user set to false, only if it was never set
          @failure_queue_name = "#{default_name}Failure" if @failure_queue_name.nil?
          @failure_queue_name
        end
      end

      def self.included(base)
        base.extend(ModernTimes::Base::Worker::ClassMethods)
        base.extend(ClassMethods)
      end

      def initialize(opts={})
        super
        @status        = 'initialized'
        # Supervisor will ask for counts in a separate thread
        @time_track    = ModernTimes::TimeTrack.new
        @error_count   = 0
      end

      def message_count
        @time_track.total_count
      end

      def status
        @status || "Processing message #{@time_track.total_count}"
      end

      def real_destination_options
        options = self.class.destination_options
        virtual_topic_name = options.delete(:virtual_topic_name)
        options[:queue_name] = "Consumer.#{name}.VirtualTopic.#{virtual_topic_name}" if virtual_topic_name
        return options
      end

      # Start the event loop for handling messages off the queue
      def start
        @session = Connection.create_consumer_session
        @consumer = @session.consumer(real_destination_options)
        @session.start

        ModernTimes.logger.debug "#{self}: Starting receive loop"
        @status = nil
        while msg = @consumer.receive
          @time_track.perform do
            on_message(msg)
            msg.acknowledge
          end
          ModernTimes.logger.info {"#{self}::on_message (#{('%.1f' % @time_track.last_time)}ms)"} if ModernTimes::JMS::Connection.log_times?
        end
        @status = 'Exited'
        ModernTimes.logger.info "#{self}: Exiting"
#      rescue javax.jms.IllegalStateException => e
#        #if e.cause.code == Java::org.jms.api.core.JMSException::OBJECT_CLOSED
#          # Normal exit
#          @status = 'Exited'
#          ModernTimes.logger.info "#{self}: Exiting due to possible close: #{e.message}\n\t#{e.backtrace.join("\n\t")}"<commit_msg>Fix logtime display so we're actually displaying ms like we say we are<commit_after>require 'benchmark'

module ModernTimes
  module JMS
  
    # Base Worker Class for any class that will be processing messages from topics or queues
    # By default, it will consume messages from a queue with the class name minus the Worker postfix.
    # For example, the queue call is unneccessary as it will default to a value of 'Foo' anyways:
    #  class FooWorker < ModernTimes::JMS::Worker
    #    queue 'Foo'
    #    def perform(obj)
    #      # Perform work on obj
    #    end
    #  end
    #
    # A topic can be specified using virtual_topic as follows (ActiveMQ only).  Multiple separate workers can
    # subscribe to the same topic (under ActiveMQ - see http://activemq.apache.org/virtual-destinations.html):
    #  class FooWorker < ModernTimes::JMS::Worker
    #    virtual_topic 'Zulu'
    #    def perform(obj)
    #      # Perform work on obj
    #    end
    #  end
    #
    # Filters can also be specified within the class:
    #  class FooWorker < ModernTimes::JMS::Worker
    #    filter 'age > 30'
    #    def perform(obj)
    #      # Perform work on obj
    #    end
    #  end
    #
    #
    module Worker
      include ModernTimes::Base::Worker

      attr_reader :session, :error_count, :time_track
      attr_accessor :message

      module ClassMethods
        def create_supervisor(manager, worker_options)
          Supervisor.new(manager, self, {}, worker_options)
        end

        def destination_options
          options = dest_options.dup
          # Default the queue name to the Worker name if a destinations hasn't been specified
          if options.keys.select {|k| [:virtual_topic_name, :queue_name, :destination].include?(k)}.empty?
            options[:queue_name] = default_name
          end
          return options
        end

        def virtual_topic(name, opts={})
          # ActiveMQ only
          dest_options[:virtual_topic_name] = name.to_s
        end

        def queue(name, opts={})
          dest_options[:queue_name] = name.to_s
        end

       def dest_options
          @dest_options ||= {}
        end

        def failure_queue(name, opts={})
          @failure_queue_name = name.to_s
        end

        def failure_queue_name
          # Don't overwrite if the user set to false, only if it was never set
          @failure_queue_name = "#{default_name}Failure" if @failure_queue_name.nil?
          @failure_queue_name
        end
      end

      def self.included(base)
        base.extend(ModernTimes::Base::Worker::ClassMethods)
        base.extend(ClassMethods)
      end

      def initialize(opts={})
        super
        @status        = 'initialized'
        # Supervisor will ask for counts in a separate thread
        @time_track    = ModernTimes::TimeTrack.new
        @error_count   = 0
      end

      def message_count
        @time_track.total_count
      end

      def status
        @status || "Processing message #{@time_track.total_count}"
      end

      def real_destination_options
        options = self.class.destination_options
        virtual_topic_name = options.delete(:virtual_topic_name)
        options[:queue_name] = "Consumer.#{name}.VirtualTopic.#{virtual_topic_name}" if virtual_topic_name
        return options
      end

      # Start the event loop for handling messages off the queue
      def start
        @session = Connection.create_consumer_session
        @consumer = @session.consumer(real_destination_options)
        @session.start

        ModernTimes.logger.debug "#{self}: Starting receive loop"
        @status = nil
        while msg = @consumer.receive
          @time_track.perform do
            on_message(msg)
            msg.acknowledge
          end
          ModernTimes.logger.info {"#{self}::on_message (#{('%.1f' % (@time_track.last_time*1000.0))}ms)"} if ModernTimes::JMS::Connection.log_times?
        end
        @status = 'Exited'
        ModernTimes.logger.info "#{self}: Exiting"
#      rescue javax.jms.IllegalStateException => e
#        #if e.cause.code == Java::org.jms.api.core.JMSException::OBJECT_CLOSED
#          # Normal exit
#          @status = 'Exited'
#          ModernTimes.logger.info "#{self}: Exiting due to possible close: #{e.message}\n\t#{e.backtrace.join("\n\t")}"
{'fraction_non_alphanumeric': 0.08859730608453321, 'fraction_numerical': 0.0015095215977705527, 'mean_word_length': 2.2949502677888294, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6319896', 'n_tokens_mistral': 2489, 'n_tokens_neox': 2415, 'n_words': 677}