Class: Etna::Clients::Metis::IngestMetisDataWorkflow

Inherits:
Struct
  • Object
show all
Defined in:
lib/etna/clients/metis/workflows/ingest_metis_data_workflow.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#ingest_filesystem ⇒ Object

Returns the value of attribute ingest_filesystem

Returns:

  • (Object) —

    the current value of ingest_filesystem



8
9
10
 
# File 'lib/etna/clients/metis/workflows/ingest_metis_data_workflow.rb', line 8

def ingest_filesystem
  @ingest_filesystem
end

#logger ⇒ Object

Returns the value of attribute logger

Returns:

  • (Object) —

    the current value of logger



8
9
10
 
# File 'lib/etna/clients/metis/workflows/ingest_metis_data_workflow.rb', line 8

def logger
  @logger
end

#metis_filesystem ⇒ Object

Returns the value of attribute metis_filesystem

Returns:

  • (Object) —

    the current value of metis_filesystem



8
9
10
 
# File 'lib/etna/clients/metis/workflows/ingest_metis_data_workflow.rb', line 8

def metis_filesystem
  @metis_filesystem
end

Instance Method Details

#copy_file(dest:, src:, &block) ⇒ Object 



36
37
38
39
40
41
 
# File 'lib/etna/clients/metis/workflows/ingest_metis_data_workflow.rb', line 36

def copy_file(dest:, src:, &block)
  ingest_filesystem.with_readable(src, "r") do |io|
    metis_filesystem.do_streaming_upload(io, dest, ingest_filesystem.stat(src).size)
    yield src, true if block_given?
  end
end

#copy_files(files, &block) ⇒ Object

Since we are doing manual triage of files, do not automatically copy directory trees. srcs must be a list of full paths to files.



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/etna/clients/metis/workflows/ingest_metis_data_workflow.rb', line 12

def copy_files(files, &block)
  files.each do |file|
    if file.is_a?(Array)
      src, dest = file
    else
      src = dest = file
    end

    if !ingest_filesystem.exist?(src)
      logger&.warn("#{src} does not exist on source filesystem. Skipping.")
      yield src, false if block_given?
      next
    end

    logger&.info("Copying file #{src} (#{Etna::Formatting.as_size(ingest_filesystem.stat(src).size)})")

    begin
      copy_file(dest: dest, src: src, &block)
    rescue Exception => e
      yield src, false if block_given?
    end
  end
end