Class: S3DataPacker::Packer

Inherits:
Object
  • Object
show all
Defined in:
lib/s3_data_packer/packer.rb

Defined Under Namespace

Modules: Error

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(opts = {}) ⇒ Packer

Returns a new instance of Packer.



9
10
11
12
13
 
# File 'lib/s3_data_packer/packer.rb', line 9

def initialize opts = {}
  @source = opts[:source]
  @target = opts[:target]
  @output = opts[:output] || S3DataPacker::JSONBatch.new
end

Instance Attribute Details

#output ⇒ Object (readonly)

Returns the value of attribute output.



7
8
9
 
# File 'lib/s3_data_packer/packer.rb', line 7

def output
  @output
end

#source ⇒ Object (readonly)

Returns the value of attribute source.



7
8
9
 
# File 'lib/s3_data_packer/packer.rb', line 7

def source
  @source
end

#target ⇒ Object (readonly)

Returns the value of attribute target.



7
8
9
 
# File 'lib/s3_data_packer/packer.rb', line 7

def target
  @target
end

Instance Method Details

#logger ⇒ Object 



19
20
21
 
# File 'lib/s3_data_packer/packer.rb', line 19

def logger
  @logger ||= S3DataPacker.logger
end

#pack! ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/s3_data_packer/packer.rb', line 27

def pack!
  log "Packing data from #{source.name} to #{target.name} ..."
  boot_workers!

  @start_time = Time.now
  begin
    each_item { |item| workers.queue.add!(item) }
    finalize_processing!
  rescue Exception => e
    log "Unexpected error, killing threads", :error
    raise e
  ensure
    workers.kill!
  end
end

#process_item(data) ⇒ Object 



43
44
45
46
47
48
49
50
 
# File 'lib/s3_data_packer/packer.rb', line 43

def process_item(data)
  output.append_data! data
  summary.count_processed
  if output.full?
    flush_batch!
    output.new_file!
  end
end

#summary ⇒ Object 



15
16
17
 
# File 'lib/s3_data_packer/packer.rb', line 15

def summary
  @summary ||= S3DataPacker::Summary.new
end

#workers ⇒ Object 



23
24
25
 
# File 'lib/s3_data_packer/packer.rb', line 23

def workers
  @workers ||= S3DataPacker::ThreadSet.new
end