<commit_before>
module Distribot
  class Workflow
    attr_accessor :id, :name, :phases, :consumer, :finished_callback_queue

    def initialize(attrs={})
      self.id = attrs[:id]
      self.name = attrs[:name]
      self.phases = [ ]
      if attrs.has_key? :phases
        attrs[:phases].each do |options|
          self.add_phase(options)
        end
      end
    end

    def validate!
      # Make sure the phases make a continuous line:
      self.phases.each do |phase|
        next if phase.is_final
        unless (self.phases.map{|x| x.name } - [phase.name]).include? phase.transitions_to
          raise "Phase '#{phase.name}' transitions to invalid phase '#{phase.transitions_to}'"
        end
        # Make sure every handler is actively watched:
        phase.handlers.each do |handler|
          queue_names = [
            "distribot.workflow.handler.#{handler}.enumerate",
            "distribot.workflow.handler.#{handler}.tasks",
          ]
          queue_names.each do |queue_name|
            unless Distribot.queue_exists?(queue_name)
              raise "The worker queue '#{queue_name}' for handler '#{handler}' does not yet exist. Make sure the handler is active within a worker."
            end
          end
        end
      end

      # Make sure the engine appears to be running:
      engine_queues = %w(
        distribot.workflow.created
        distribot.workflow.phase.started
        distribot.workflow.task.finished
        distribot.workflow.handler.finished
        distribot.workflow.phase.finished
        distribot.workflow.finished
      )
      missing_queues = engine_queues.reject{|queue| Distribot.queue_exists?(queue) }
      unless missing_queues.empty?
        raise "The following engine queues are missing. Ensure their workers are enabled. #{missing_queues.join(", ")}"
      end

      # Finally:
      true
    end

    def self.create!(attrs={})
      obj = self.new(attrs)
      obj.save!
      return obj
    end

    def save!(&block)
      self.id = SecureRandom.uuid
      record_id = self.redis_id + ':definition'
      is_new = redis.keys(record_id).count <= 0
      redis.set record_id, serialize

      if is_new
        Distribot.publish! 'distribot.workflow.created', {
          workflow_id: self.id
        }
        if block_given?
          Thread.new do
            while true do
              sleep 1
              if self.finished?
                block.call( workflow_id: self.id )
                break
              end
            end
          end
        end
        self.add_transition( :to => self.current_phase, :timestamp => Time.now.utc.to_f )
      end<commit_msg>Fixed bug where we were announcing a transition to the next phase before the transition happened. This caused the wrong transition to occur within the handler.<commit_after>
module Distribot
  class Workflow
    attr_accessor :id, :name, :phases, :consumer, :finished_callback_queue

    def initialize(attrs={})
      self.id = attrs[:id]
      self.name = attrs[:name]
      self.phases = [ ]
      if attrs.has_key? :phases
        attrs[:phases].each do |options|
          self.add_phase(options)
        end
      end
    end

    def validate!
      # Make sure the phases make a continuous line:
      self.phases.each do |phase|
        next if phase.is_final
        unless (self.phases.map{|x| x.name } - [phase.name]).include? phase.transitions_to
          raise "Phase '#{phase.name}' transitions to invalid phase '#{phase.transitions_to}'"
        end
        # Make sure every handler is actively watched:
        phase.handlers.each do |handler|
          queue_names = [
            "distribot.workflow.handler.#{handler}.enumerate",
            "distribot.workflow.handler.#{handler}.tasks",
          ]
          queue_names.each do |queue_name|
            unless Distribot.queue_exists?(queue_name)
              raise "The worker queue '#{queue_name}' for handler '#{handler}' does not yet exist. Make sure the handler is active within a worker."
            end
          end
        end
      end

      # Make sure the engine appears to be running:
      engine_queues = %w(
        distribot.workflow.created
        distribot.workflow.phase.started
        distribot.workflow.task.finished
        distribot.workflow.handler.finished
        distribot.workflow.phase.finished
        distribot.workflow.finished
      )
      missing_queues = engine_queues.reject{|queue| Distribot.queue_exists?(queue) }
      unless missing_queues.empty?
        raise "The following engine queues are missing. Ensure their workers are enabled. #{missing_queues.join(", ")}"
      end

      # Finally:
      true
    end

    def self.create!(attrs={})
      obj = self.new(attrs)
      obj.save!
      return obj
    end

    def save!(&block)
      self.id = SecureRandom.uuid
      record_id = self.redis_id + ':definition'
      is_new = redis.get(record_id).to_s == ''
      redis.set record_id, serialize
      redis.sadd 'distribot.workflows.active', self.id

      if is_new
        self.add_transition( :to => self.current_phase, :timestamp => Time.now.utc.to_f )
        Distribot.publish! 'distribot.workflow.created', {
          workflow_id: self.id
        }
        if block_given?
          Thread.new do
            while true do
              sleep 1
              if self.finished?
                block.call( workflow_id: self.id )
                break
              end
            end
          end
        end
      end
{'fraction_non_alphanumeric': 0.09231048240841494, 'fraction_numerical': 0.000544069640914037, 'mean_word_length': 2.0021774632553075, 'pattern_counts': {'":': 0, '<': 4, '<?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': '3363129', 'n_tokens_mistral': 1610, 'n_tokens_neox': 1591, 'n_words': 408}