 ##
 # Adds a new stream to the this flow.
 ##
-Flow::add = ( priority, name, streams ) ->
+Flow::add = ( priority, name, streams, options ) ->
   if @usedPriorities.indexOf( priority ) isnt -1
     warlock.log.warning "Warning loading stream '#{name}': multiple tasks in '#{@name}' flow are " +
       "set to run at #{priority}. There is no guarantee which will run first."

    name: name
    priority: priority
    run: streams

  warlock.debug.log "[#{@name}] Added stream to queue: #{name}"
  this

  @stream = @_mergeArrayOfStreams @queues.pre

  steps.forEach ( step ) =>
    # Only pipe to this stream if not disabled
    if warlock.task.isPrevented step.name
      warlock.verbose.log "[#{@name}] Stream #{step.name} prevented from running."
    else
      dest = step.run

      # If it's a raw handler, we pass it a stream and trust that it returns one as well.
      if step.options?.raw
        warlock.verbose.log "[#{@name}] Handing stream off to #{step.name}."
        @stream = dest warlock.config( "tasks.#{step.name}" ), warlock.streams.highland( @stream )

        warlock.fatal "#{@name}.#{step.name} did not return a stream." if not warlock.util.isStream @stream
      else
        warlock.verbose.log "[#{@name}] Piping stream to #{step.name}."
        if warlock.util.isFunction dest
          dest = dest warlock.config( "tasks.#{step.name}" )

        if warlock.util.isArray dest
          dest.push @stream
          @stream = @_mergeArrayOfStreams dest
        else
          @stream = @stream.pipe dest

  # Add it into the post queue with anything another flow may have merged here.
  @queues.post.push @stream
