Skip to content

Commit

Permalink
Support keyword arguments in iteration jobs
Browse files Browse the repository at this point in the history
It's now more common to have jobs that take keyword arguments, like
`MyJob.perform_later(foo: "bar")`. But under the current implementation,
`build_enumerator` can only take them as a hash argument instead of
keyword arguments:

```ruby
def build_enumerator(params)
  foo = params[:foo]
  # ...
end
```

This commit fixes that by allowing keyword arguments to be passed to
`build_enumerator` and `each_iteration`.
  • Loading branch information
st0012 committed Jan 22, 2025
1 parent 5f44ee6 commit 54febaf
Show file tree
Hide file tree
Showing 2 changed files with 25 additions and 8 deletions.
17 changes: 9 additions & 8 deletions lib/job-iteration/iteration.rb
Original file line number Diff line number Diff line change
Expand Up @@ -107,8 +107,8 @@ def deserialize(job_data) # @private
self.total_time = Float(job_data["total_time"] || 0.0)
end

def perform(*params) # @private
interruptible_perform(*params)
def perform(*args, **kwargs) # @private
interruptible_perform(*args, **kwargs)

nil
end
Expand All @@ -128,12 +128,12 @@ def enumerator_builder
JobIteration.enumerator_builder.new(self)
end

def interruptible_perform(*arguments)
def interruptible_perform(*args, **kwargs)
self.start_time = Time.now.utc

enumerator = nil
ActiveSupport::Notifications.instrument("build_enumerator.iteration", instrumentation_tags) do
enumerator = build_enumerator(*arguments, cursor: cursor_position)
enumerator = build_enumerator(*args, **kwargs, cursor: cursor_position)
end

unless enumerator
Expand All @@ -153,7 +153,7 @@ def interruptible_perform(*arguments)
end

completed = catch(:abort) do
iterate_with_enumerator(enumerator, arguments)
iterate_with_enumerator(enumerator, args, kwargs)
end

run_callbacks(:shutdown)
Expand All @@ -170,8 +170,9 @@ def interruptible_perform(*arguments)
end
end

def iterate_with_enumerator(enumerator, arguments)
arguments = arguments.dup.freeze
def iterate_with_enumerator(enumerator, args, kwargs)
args = args.dup.freeze
kwargs = kwargs.dup.freeze
found_record = false
@needs_reenqueue = false

Expand All @@ -183,7 +184,7 @@ def iterate_with_enumerator(enumerator, arguments)
ActiveSupport::Notifications.instrument("each_iteration.iteration", tags) do
found_record = true
run_callbacks(:iterate) do
each_iteration(object_from_enumerator, *arguments)
each_iteration(object_from_enumerator, *args, **kwargs)
end
self.cursor_position = cursor_from_enumerator
end
Expand Down
16 changes: 16 additions & 0 deletions test/unit/active_job_iteration_test.rb
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,16 @@ def each_iteration(_record, params)
end
end

class KwargsIterationJob < SimpleIterationJob
def build_enumerator(times:, cursor:)
enumerator_builder.build_times_enumerator(times, cursor: cursor)
end

def each_iteration(_record, times:)
self.class.records_performed << { times: times }
end
end

class ActiveRecordIterationJob < SimpleIterationJob
def build_enumerator(cursor:)
enumerator_builder.active_record_on_records(
Expand Down Expand Up @@ -605,6 +615,12 @@ def test_passes_params_to_each_iteration
assert_equal([params, params], ParamsIterationJob.records_performed)
end

def test_passes_kwargs_to_each_iteration
KwargsIterationJob.perform_later(times: 3)
work_one_job
assert_equal([{ times: 3 }, { times: 3 }, { times: 3 }], KwargsIterationJob.records_performed)
end

def test_passes_params_to_each_iteration_without_extra_information_on_interruption
iterate_exact_times(1.times)
params = { "walrus" => "yes", "morewalrus" => "si" }
Expand Down

0 comments on commit 54febaf

Please sign in to comment.