diff options
author | Robert Speicher <rspeicher@gmail.com> | 2021-01-20 13:34:23 -0600 |
---|---|---|
committer | Robert Speicher <rspeicher@gmail.com> | 2021-01-20 13:34:23 -0600 |
commit | 6438df3a1e0fb944485cebf07976160184697d72 (patch) | |
tree | 00b09bfd170e77ae9391b1a2f5a93ef6839f2597 /lib/bulk_imports/pipeline/runner.rb | |
parent | 42bcd54d971da7ef2854b896a7b34f4ef8601067 (diff) | |
download | gitlab-ce-6438df3a1e0fb944485cebf07976160184697d72.tar.gz |
Add latest changes from gitlab-org/gitlab@13-8-stable-eev13.8.0-rc42
Diffstat (limited to 'lib/bulk_imports/pipeline/runner.rb')
-rw-r--r-- | lib/bulk_imports/pipeline/runner.rb | 30 |
1 files changed, 13 insertions, 17 deletions
diff --git a/lib/bulk_imports/pipeline/runner.rb b/lib/bulk_imports/pipeline/runner.rb index 88b96f0ab6e..11fb9722173 100644 --- a/lib/bulk_imports/pipeline/runner.rb +++ b/lib/bulk_imports/pipeline/runner.rb @@ -12,25 +12,15 @@ module BulkImports info(context, message: 'Pipeline started', pipeline_class: pipeline) - extractors.each do |extractor| - data = run_pipeline_step(:extractor, extractor.class.name, context) do - extractor.extract(context) + Array.wrap(extracted_data_from(context)).each do |entry| + transformers.each do |transformer| + entry = run_pipeline_step(:transformer, transformer.class.name, context) do + transformer.transform(context, entry) + end end - if data && data.respond_to?(:each) - data.each do |entry| - transformers.each do |transformer| - entry = run_pipeline_step(:transformer, transformer.class.name, context) do - transformer.transform(context, entry) - end - end - - loaders.each do |loader| - run_pipeline_step(:loader, loader.class.name, context) do - loader.load(context, entry) - end - end - end + run_pipeline_step(:loader, loader.class.name, context) do + loader.load(context, entry) end end @@ -55,6 +45,12 @@ module BulkImports mark_as_failed(context) if abort_on_failure? end + def extracted_data_from(context) + run_pipeline_step(:extractor, extractor.class.name, context) do + extractor.extract(context) + end + end + def mark_as_failed(context) warn(context, message: 'Pipeline failed', pipeline_class: pipeline) |