Method: Que::Job._bulk_enqueue_insert

Defined in:

lib/que/job.rb

._bulk_enqueue_insert(args_and_kwargs_array, job_options: {}, notify:) ⇒ Object

# File 'lib/que/job.rb', line 155

def _bulk_enqueue_insert(args_and_kwargs_array, job_options: {}, notify:)
  raise 'Unexpected bulk args format' if !args_and_kwargs_array.is_a?(Array) || !args_and_kwargs_array.all? { |a| a.is_a?(Hash) }

  if job_options[:tags]
    if job_options[:tags].length > MAXIMUM_TAGS_COUNT
      raise Que::Error, "Can't enqueue a job with more than #{MAXIMUM_TAGS_COUNT} tags! (passed #{job_options[:tags].length})"
    end

    job_options[:tags].each do |tag|
      if tag.length > MAXIMUM_TAG_LENGTH
        raise Que::Error, "Can't enqueue a job with a tag longer than 100 characters! (\"#{tag}\")"
      end
    end
  end

  args_and_kwargs_array = args_and_kwargs_array.map do |args_and_kwargs|
    args_and_kwargs.merge(
      args: args_and_kwargs.fetch(:args, []),
      kwargs: args_and_kwargs.fetch(:kwargs, {}),
    )
  end

  attrs = {
    queue:    job_options[:queue]    || resolve_que_setting(:queue) || Que.default_queue,
    priority: job_options[:priority] || resolve_que_setting(:priority),
    run_at:   job_options[:run_at]   || resolve_que_setting(:run_at),
    args_and_kwargs_array: args_and_kwargs_array,
    data:     job_options[:tags] ? { tags: job_options[:tags] } : {},
    job_class: \
      job_options[:job_class] || name ||
        raise(Error, "Can't enqueue an anonymous subclass of Que::Job"),
  }

  if attrs[:run_at].nil? && resolve_que_setting(:run_synchronously)
    args_and_kwargs_array = Que.deserialize_json(Que.serialize_json(attrs.delete(:args_and_kwargs_array)))
    args_and_kwargs_array.map do |args_and_kwargs|
      _run_attrs(
        attrs.merge(
          args: args_and_kwargs.fetch(:args),
          kwargs: args_and_kwargs.fetch(:kwargs),
        ),
      )
    end
  else
    attrs.merge!(
      args_and_kwargs_array: Que.serialize_json(attrs[:args_and_kwargs_array]),
      data: Que.serialize_json(attrs[:data]),
    )
    values_array =
      Que.transaction do
        Que.execute('SET LOCAL que.skip_notify TO true') unless notify
        Que.execute(
          :bulk_insert_jobs,
          attrs.values_at(:queue, :priority, :run_at, :job_class, :args_and_kwargs_array, :data),
        )
      end
    values_array.map(&method(:new))
  end
end