diff options
| -rw-r--r-- | app/controllers/api/v1/netex_imports_controller.rb | 4 | ||||
| -rw-r--r-- | app/services/http_service.rb | 4 | ||||
| -rw-r--r-- | app/services/zip_service.rb | 9 | ||||
| -rw-r--r-- | app/workers/workbench_import_worker.rb | 56 |
4 files changed, 33 insertions, 40 deletions
diff --git a/app/controllers/api/v1/netex_imports_controller.rb b/app/controllers/api/v1/netex_imports_controller.rb index 7dc18a7be..fd28ba45a 100644 --- a/app/controllers/api/v1/netex_imports_controller.rb +++ b/app/controllers/api/v1/netex_imports_controller.rb @@ -44,10 +44,8 @@ module Api ReferentialMetadataKludge.make_metadata_from_name! netex_import_params['name'], referential_id: @new_referential.id # <<< REMOVE ME !!!! rescue ActiveRecord::RecordInvalid - require 'pry' - binding.pry # render json: {errors: @new_referential.errors}, status: 406 - render json: {errors: ErrorFormat.detailed(@new_referential)}, status: 406 + render json: {errors: ErrorFormat.details(@new_referential)}, status: 406 finish_action! end diff --git a/app/services/http_service.rb b/app/services/http_service.rb index 6cf8ef93b..d3999f293 100644 --- a/app/services/http_service.rb +++ b/app/services/http_service.rb @@ -26,13 +26,11 @@ module HTTPService extend self # params: { netex_import: {referential_id: 13, workbench_id: 1}}, # upload: {file: [StringIO.new('howdy'), 'application/zip', 'greeting']}) def post_resource(host:, path:, token: nil, params: {}, upload: nil) - Faraday.new(url: host) do |c| + result = Faraday.new(url: host) do |c| c.headers['Authorization'] = "Token token=#{token.inspect}" if token c.request :multipart c.request :url_encoded c.adapter Faraday.default_adapter - require 'pry' - binding.pry return c.post path, params end end diff --git a/app/services/zip_service.rb b/app/services/zip_service.rb index f2e7fc2ed..cab301b01 100644 --- a/app/services/zip_service.rb +++ b/app/services/zip_service.rb @@ -1,12 +1,15 @@ class ZipService + # TODO: Remove me before merge https://github.com/rubyzip/rubyzip - class Subdir < Struct.new( :name, :stream) + class Subdir < Struct.new(:name, :stream) end attr_reader :current_key, :current_output, :yielder def initialize data - @zip_data = StringIO.new(data) + @zip_data = StringIO.new(data) + @current_key = nil + @current_output = nil end def subdirs @@ -38,6 +41,7 @@ class ZipService end def write_to_current_output input_stream + # the condition below is true for directory entries return if Zip::NullInputStream == input_stream current_output.write input_stream.read end @@ -58,6 +62,7 @@ class ZipService end def entry_key entry + # last dir name File.dirname.split("/").last entry.name.split('/', -1)[-2] end end diff --git a/app/workers/workbench_import_worker.rb b/app/workers/workbench_import_worker.rb index d06d4607c..325ff401d 100644 --- a/app/workers/workbench_import_worker.rb +++ b/app/workers/workbench_import_worker.rb @@ -3,15 +3,13 @@ class WorkbenchImportWorker include Rails.application.routes.url_helpers include Configurable - RETRY_DELAYS = [3, 5, 8] - # Workers # ======= def perform(import_id) @workbench_import = WorkbenchImport.find(import_id) @response = nil - @workbench_import.update_attributes(status: 'running') + @workbench_import.update(status: 'running') downloaded = download zip_service = ZipService.new(downloaded) upload zip_service @@ -34,49 +32,38 @@ class WorkbenchImportWorker params: params(eg_file, eg_name)) end - def log_failure reason, count - logger.warn "HTTP POST failed with #{reason}, count = #{count}, response=#{@response}" - end - - def try_upload_entry_group eg_name, eg_file - result = execute_post eg_name, eg_file - return Result.ok(result) if result && result.status < 400 - Result.error(JSON.parse result.body) - end - def upload zip_service entry_group_streams = zip_service.subdirs - @workbench_import.update_attributes total_steps: entry_group_streams.size + @workbench_import.update total_steps: entry_group_streams.size entry_group_streams.each_with_index(&method(:upload_entry_group)) - rescue StopIteration - @workbench_import.update_attributes( current_step: entry_group_streams.size, status: 'failed' ) + rescue Exception => e + logger.error e.message + @workbench_import.update( current_step: entry_group_streams.size, status: 'failed' ) + raise end def upload_entry_group entry_pair, element_count - @workbench_import.update_attributes( current_step: element_count.succ ) - retry_service = RetryService.new( - delays: RETRY_DELAYS, - rescue_from: [HTTPService::Timeout], - logger: logger, - &method(:log_failure)) - status = retry_service.execute(&upload_entry_group_proc(entry_pair)) - raise StopIteration unless status.ok? - end - - def upload_entry_group_proc entry_pair + @workbench_import.update( current_step: element_count.succ ) + # status = retry_service.execute(&upload_entry_group_proc(entry_pair)) eg_name = entry_pair.name eg_stream = entry_pair.stream - eg_file = Tempfile.new("WorkbenchImport_#{eg_name}_#{$$}").tap do |file| - eg_stream.read + eg_file = File.new(Rails.root.join('tmp', "WorkbenchImport_#{eg_name}_#{$$}.zip"), 'wb').tap do |file| + eg_stream.rewind + file.write eg_stream.read end eg_file.rewind - -> do - try_upload_entry_group(eg_name, eg_file) + result = execute_post eg_name, eg_file + if result && result.status < 400 + result + else + raise StopIteration, result.body end + ensure + eg_file.close rescue nil + eg_file.unlink rescue nil end - # Queries # ======= @@ -112,6 +99,11 @@ class WorkbenchImportWorker end def params file, name + if dest = ENV["DEBUG_TEMPFILE"] + require 'pry' + binding.pry + %x{unzip -oqq #{file.path} -d #{dest}} + end { netex_import: { referential_id: @workbench_import.referential_id, workbench_id: @workbench_import.workbench_id, |