<commit_before>class StreamWorkflowTask
  include WorkflowRESTHelpers
  include RbbtRESTHelpers

  def initialize(app)
    @app = app
  end

  EOL = "\r\n"

  def parse_uri(env)
    uri = env["REQUEST_URI"]
    _n, workflow, task = uri.split("/")
    workflow = begin
                 Kernel.const_get(workflow)
               rescue
                 raise "Could not accept task for workflow: #{ workflow }"
               end
    [workflow, task]
  end

  def read_normal_inputs(io, boundary, stream_input)
    inputs = {}

    input_name = nil
    variable_chunk = nil
    filename = nil

    while line = io.gets
      line.chomp!

      chunk_start = line == "--" + boundary

      if chunk_start
        if input_name
          inputs[input_name] = variable_chunk
        end
        content_start = false
      elsif content_start  
        if variable_chunk.empty?
          variable_chunk << line
        else
          variable_chunk << "\n" << line
        end
      elsif line =~ /^Content.* name="([^\s;"]*)"/
        input_name = $1
        filename = line.match(/filename="([^"]+)"/)[1] if line =~ /filename/
      elsif line.empty?
        variable_chunk = ""
        break if input_name == stream_input
        content_start = true
      end
    end
    
    [inputs, filename]
  end

  def copy_until_boundary(sin, sout, boundary)
    while line = sin.gets
      break if line.include? boundary
      sout.write line
    end
  end

  def get_inputs(content_type, stream)
    boundary = content_type.match(/boundary=([^\s;]*)/)[1]
    stream_input = content_type.match(/stream=([^\s;]*)/)[1]
    inputs, filename = read_normal_inputs(stream, boundary, stream_input)

    IndiferentHash.setup(inputs)

    [inputs, stream_input, filename, stream, boundary]
  end

  def run_job(workflow, task, inputs, stream_input, stream, boundary, filename = nil)
    name = inputs.delete "jobname"

    task_parameters = prepare_job_inputs(workflow, task, inputs)

    Misc.add_stream_filename(stream, filename) if filename

    task_parameters[stream_input] = stream

    task = task.to_sym

    job = workflow.job(task, name, task_parameters)

    execution_type = case
                     when workflow.exec_exports.include?(task)
                       "exec"
                     when workflow.synchronous_exports.include?(task)
                       "synchronous"
                     when workflow.asynchronous_exports.include?(task)
                       "asynchronous"
                     else
                       raise "No known export type for #{ workflow } #{ task }. Accesses denied"
                     end 

    execution_type = "exec" if inputs["_cache_type"] == 'exec'

    begin
      case execution_type<commit_msg>Avoid streaming the end of the multipart section into the job<commit_after>class StreamWorkflowTask
  include WorkflowRESTHelpers
  include RbbtRESTHelpers

  def initialize(app)
    @app = app
  end

  EOL = "\r\n"

  def parse_uri(env)
    uri = env["REQUEST_URI"]
    _n, workflow, task = uri.split("/")
    workflow = begin
                 Kernel.const_get(workflow)
               rescue
                 raise "Could not accept task for workflow: #{ workflow }"
               end
    [workflow, task]
  end

  def read_normal_inputs(io, boundary, stream_input)
    inputs = {}

    input_name = nil
    variable_chunk = nil
    filename = nil

    while line = io.gets
      line.chomp!

      chunk_start = line == "--" + boundary

      if chunk_start
        if input_name
          inputs[input_name] = variable_chunk
        end
        content_start = false
      elsif content_start  
        if variable_chunk.empty?
          variable_chunk << line
        else
          variable_chunk << "\n" << line
        end
      elsif line =~ /^Content.* name="([^\s;"]*)"/
        input_name = $1
        filename = line.match(/filename="([^"]+)"/)[1] if line =~ /filename/
      elsif line.empty?
        variable_chunk = ""
        break if input_name == stream_input
        content_start = true
      end
    end
    
    [inputs, filename]
  end

  def copy_until_boundary(sin, sout, boundary)
    last_line = nil
    while line = sin.gets
      break if line.include? boundary
      sout.write last_line
      last_line = line
    end
    sout.write last_line.strip unless last_line == EOL
  end

  def get_inputs(content_type, stream)
    boundary = content_type.match(/boundary=([^\s;]*)/)[1]
    stream_input = content_type.match(/stream=([^\s;]*)/)[1]
    inputs, filename = read_normal_inputs(stream, boundary, stream_input)

    IndiferentHash.setup(inputs)

    [inputs, stream_input, filename, stream, boundary]
  end

  def run_job(workflow, task, inputs, stream_input, stream, boundary, filename = nil)
    name = inputs.delete "jobname"

    task_parameters = prepare_job_inputs(workflow, task, inputs)

    Misc.add_stream_filename(stream, filename) if filename

    clean_stream = Misc.open_pipe do |sin|
      copy_until_boundary(stream, sin, boundary)
    end

    task_parameters[stream_input] = clean_stream

    task = task.to_sym

    job = workflow.job(task, name, task_parameters)

    execution_type = case
                     when workflow.exec_exports.include?(task)
                       "exec"
                     when workflow.synchronous_exports.include?(task)
                       "synchronous"
                     when workflow.asynchronous_exports.include?(task)
                       "asynchronous"
                     else
                       raise "No known export type for #{ workflow } #{ task }. Accesses denied"
                     end 

    execution_type = "exec" if inputs["_cache_type"] == 'exec'

    begin
      case execution_type
{'fraction_non_alphanumeric': 0.09568545581071677, 'fraction_numerical': 0.0013917884481558804, 'mean_word_length': 2.1143011917659806, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 3, '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': '3853127', 'n_tokens_mistral': 1715, 'n_tokens_neox': 1633, 'n_words': 412}