<commit_before>require 'travis/build'
require 'simple_states'
require 'multi_json'
require 'thread'

module Travis
  module Worker
    # Represents a single Worker which is bound to a single VM instance.
    class Worker
      autoload :Factory, 'travis/worker/worker/factory'
      autoload :Heart,   'travis/worker/worker/heart'

      include SimpleStates
      include Util::Logging

      def self.create(name, config)
        Factory.new(name, config).worker
      end

      states :created, :starting, :ready, :working, :stopping, :stopped, :errored

      attr_accessor :state
      attr_reader :name, :vm, :queue, :reporter, :logger, :config, :payload, :last_error

      # Instantiates a new worker.
      #
      # queue - The Amqp builds queue.
      # vm    - The virtual machine to be used by the worker.
      def initialize(name, vm, queue, reporter, logger, config)
        @name     = name
        @vm       = vm
        @queue    = queue
        @reporter = reporter
        @logger   = logger
        @config   = config
      end

      # Boots the worker by preparing the VM and subscribing to the builds queue.
      def start
        self.state = :starting
        vm.prepare
        heart.beat
        self.state = :ready
        queue.subscribe(:ack => true, :blocking => false, &method(:process))
      end
      log :start

      # Stops the worker by cancelling the builds queue subscription.
      def stop(options = {})
        self.state = :stopping
        queue.unsubscribe
        kill if options[:force]
      end
      log :stop

      # Forcefully stops the current job
      def kill
        vm.shell.terminate("Worker #{name} was stopped forcefully.")
      end

      def report
        { :state => state, :last_error => last_error, :payload => payload }
      end

      protected

        def process(message, payload)
          work(message, payload)
        rescue Errno::ECONNREFUSED, Exception => error
          # puts error.message, error.backtrace
          error(error, message)
        end

        def work(message, payload)
          payload = prepare(payload)<commit_msg>be sure to set the worker state to :stopped if the worker is not working<commit_after>require 'travis/build'
require 'simple_states'
require 'multi_json'
require 'thread'

module Travis
  module Worker
    # Represents a single Worker which is bound to a single VM instance.
    class Worker
      autoload :Factory, 'travis/worker/worker/factory'
      autoload :Heart,   'travis/worker/worker/heart'

      include SimpleStates
      include Util::Logging

      def self.create(name, config)
        Factory.new(name, config).worker
      end

      states :created, :starting, :ready, :working, :stopping, :stopped, :errored

      attr_accessor :state
      attr_reader :name, :vm, :queue, :reporter, :logger, :config, :payload, :last_error

      # Instantiates a new worker.
      #
      # queue - The Amqp builds queue.
      # vm    - The virtual machine to be used by the worker.
      def initialize(name, vm, queue, reporter, logger, config)
        @name     = name
        @vm       = vm
        @queue    = queue
        @reporter = reporter
        @logger   = logger
        @config   = config
      end

      # Boots the worker by preparing the VM and subscribing to the builds queue.
      def start
        self.state = :starting
        vm.prepare
        heart.beat
        self.state = :ready
        queue.subscribe(:ack => true, :blocking => false, &method(:process))
      end
      log :start

      # Stops the worker by cancelling the builds queue subscription.
      def stop(options = {})
        self.state = :stopping
        queue.unsubscribe
        kill if options[:force]
        self.state = :stopped unless working?
      end
      log :stop

      # Forcefully stops the current job
      def kill
        vm.shell.terminate("Worker #{name} was stopped forcefully.")
      end

      def report
        { :state => state, :last_error => last_error, :payload => payload }
      end

      protected

        def process(message, payload)
          work(message, payload)
        rescue Errno::ECONNREFUSED, Exception => error
          # puts error.message, error.backtrace
          error(error, message)
        end

        def work(message, payload)
          payload = prepare(payload)
{'fraction_non_alphanumeric': 0.08606839568510442, 'fraction_numerical': 0.0, 'mean_word_length': 2.142033165104542, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, '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': '6930756', 'n_tokens_mistral': 1221, 'n_tokens_neox': 1187, 'n_words': 401}